#! /usr/bin/env python ############################################################################# ## ## ## scapy6.py --- IPv6 support for Scapy ## ## see http://namabiiru.hongo.wide.ad.jp/scapy6/ ## ## for more informations ## ## ## ## Copyright (C) 2005 Guillaume Valadon ## ## Arnaud Ebalard ## ## ## ## This program is free software; you can redistribute it and/or modify it ## ## under the terms of the GNU General Public License version 2 as ## ## published by the Free Software Foundation; version 2. ## ## ## ## This program is distributed in the hope that it will be useful, but ## ## WITHOUT ANY WARRANTY; without even the implied warranty of ## ## MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU ## ## General Public License for more details. ## ## ## ############################################################################# from scapy import * ############ ## Consts ## ############ ETH_P_IPV6 = 0x86dd ############################## ## Routing/Interfaces stuff ## ############################## class Route6: def __init__(self): self.routes = read_routes6() if self.routes == []: log_loading.info("No IPv6 support in Linux kernel") def route(self, x): return Route6.route_iface(self, x, conf.iface) def route_iface(self, x, iface=''): x = socket.inet_pton(socket.AF_INET6, x) pathes = [] for d,p,nh,i,a in self.routes: t = in6_cidr2mask(10) if (in6_and(x,t) == socket.inet_pton(socket.AF_INET6, 'fe80::') and i != iface): # link-local addresses: fe080::/10 continue t = in6_cidr2mask(8) if (in6_and(x,t) == socket.inet_pton(socket.AF_INET6, 'ff00::') and i != iface): # multicast link-local addresses continue t = socket.inet_pton(socket.AF_INET6, d) cm = in6_cidr2mask(p) if in6_and(x, cm) == in6_and(t, cm): pathes.append((p,(i,a,nh))) if not pathes: raise Exception("no IPv6 route found") pathes.sort() return pathes[-1][1] def __repr__(self): rt = "%-46s %-42s %-10s %s\n" % ('Destination', 'Next Hop', 'Iface', 'Output IPv6s') for net,msk,gw,iface,addr in self.routes: t = '%s/%i'% (net,msk) rt += "%-46s %-42s %-10s %s\n" % (t, gw, iface, addr) return rt def get_if_raw_addr6(iff): r = filter(lambda x: x[2] == iff and x[1] == '00', in6_getifaddr()) if len(r) == 0: return None else: r = r[0][0] return socket.inet_pton(socket.AF_INET6, r) if not LINUX: def in6_getifaddr(): ret = [] i = dnet.intf() for int in i: ifname = int['name'] if int.has_key('alias_addrs'): v6 = i.get(ifname)['alias_addrs'] else: ret += [ (None, None, ifname) ] for a in v6: if a.type != dnet.ADDR_TYPE_IP6: continue xx = str(a).split('/')[0] xx = in6_ptop(xx) scope = '20' x = a.ip6 if in6_and(x,in6_cidr2mask(10)) == socket.inet_pton(socket.AF_INET6, 'fe80::'): scope = '20' elif in6_and(x,in6_cidr2mask(3)) == socket.inet_pton(socket.AF_INET6, '2000::'): scope = '00' ret += [ (xx, scope, ifname) ] return ret def read_routes6(): f=os.popen("netstat -rn -f inet6") ok = 0 mtu = False routes = [] lifaddr = in6_getifaddr() for l in f.readlines(): if not l: break l = l.strip() if l.find("Destination") >= 0: ok = 1 continue if ok == 0: continue if not l: break dest,nh,fl,dev = l.split()[:4] if 'H' in fl: continue if dest == "default": d = '::' dp = 0 if '%' in nh: nh,dev = nh.split('%') ifaddr = filter(lambda x: x[2] == dev and x[1] == '00', lifaddr)[0][0] else: d = dest dp = 128 ifaddr = None nh = '::' if '/' in dest: d,dp = dest.split("/") dp = int(dp) if '%' in d: d,ifaddr = d.split('%') if '%' in nh: nh,ifaddr = nh.split('%') x = socket.inet_pton(socket.AF_INET6, d) if in6_and(x, in6_cidr2mask(8)) == socket.inet_pton(socket.AF_INET6, 'ff00::'): ifaddr = filter(lambda x: x[2] == dev and x[1] == '20', lifaddr)[0][0] elif dev != 'lo' and in6_and(x,in6_cidr2mask(10)) == socket.inet_pton(socket.AF_INET6, 'fe80::'): ifaddr = filter(lambda x: x[2] == dev and x[1] == '20', lifaddr)[0][0] elif dev != 'lo' and in6_and(x,in6_cidr2mask(3)) == socket.inet_pton(socket.AF_INET6, '2000::'): ifaddr = filter(lambda x: x[2] == dev and x[1] == '00', lifaddr)[0][0] routes.append((d, dp, nh, dev, ifaddr)) f.close() return routes else: def in6_getifaddr(): ret = [] try: f = open("/proc/net/if_inet6","r") except IOError, err: return ret l = f.readlines() for i in l: # addr, index, plen, scope, flags, ifname tmp = i.split() t = struct.unpack('4s4s4s4s4s4s4s4s', tmp[0]) t = ':'.join(t) t = in6_ptop(t) ret += [ (t, tmp[3], tmp[5]) ] return ret def read_routes6(): try: f = open("/proc/net/ipv6_route","r") except IOError, err: return [] # 1. destination network # 2. destination prefix length # 3. source network displayed # 4. source prefix length # 5. next hop # 6. metric # 7. reference counter (?!?) # 8. use counter (?!?) # 9. flags # 10. device name routes = [] def proc2r(p): ret = struct.unpack('4s4s4s4s4s4s4s4s', p) ret = ':'.join(ret); return in6_ptop(ret) def pl2i(pl): return 16*int(pl[0])+int(pl[1]) lifaddr = in6_getifaddr() for l in f.readlines(): d,dp,s,sp,nh,m,rc,us,fl,dev = l.split() d = proc2r(d) dp = pl2i(dp) s = proc2r(s) sp = pl2i(sp) nh = proc2r(nh) x = socket.inet_pton(socket.AF_INET6, d) if dev != 'lo' and in6_and(x, in6_cidr2mask(8)) == socket.inet_pton(socket.AF_INET6, 'ff00::'): ifaddr = filter(lambda x: x[1] == '20' and x[2] == dev, lifaddr) ifaddr = map(lambda x: x[0], ifaddr) elif dev != 'lo' and in6_and(x,in6_cidr2mask(10)) == socket.inet_pton(socket.AF_INET6, 'fe80::'): ifaddr = filter(lambda x: x[1] == '20' and x[2] == dev, lifaddr) ifaddr = map(lambda x: x[0], ifaddr) elif dev != 'lo' and in6_and(x,in6_cidr2mask(3)) == socket.inet_pton(socket.AF_INET6, '2000::'): ifaddr = filter(lambda x: x[1] == '00' and x[2] == dev, lifaddr) ifaddr = map(lambda x: x[0], ifaddr) elif d == '::' and dp == 0: if nh != '::': ifaddr = filter(lambda x: x[1] == '00', lifaddr) ifaddr = map(lambda x: x[0], ifaddr) else: ifaddr = [None] elif dev == 'lo': ifaddr = [] for a,t,db in lifaddr: if in6_and(x, socket.inet_pton(socket.AF_INET6, a)) == x: ifaddr += [a] if ifaddr == []: ifaddr += ['::1'] else: ifaddr= [None] if ifaddr != [None]: ifaddr = ifaddr[0] # TODO: IPv6, check behavior routes.append((d, dp, nh, dev, ifaddr)) f.close() return routes def get_if_addr6(iff): addr = get_if_raw_addr6(iff) if addr is None: return None return socket.inet_ntop(socket.AF_INET6, addr) ########################## ## Neighbor cache stuff ## ########################## NEIGHTIMEOUT=120 ns_cache={} def getmacbyip6(ip6): # link-local multicast addresses x = socket.inet_pton(socket.AF_INET6, ip6) if in6_and(x, in6_cidr2mask(8)) == socket.inet_pton(socket.AF_INET6, 'ff00::'): if ns_cache.has_key(ip6): mac, timeout = ns_cache[ip6] if timeout and (time.time()-timeout < NEIGHTIMEOUT): return mac mac = in6_getnsmac(x) ns_cache[ip6] = (mac,time.time()) return mac iff,a,nh = conf.route6.route_iface(ip6, conf.iface) if iff == "lo": return "ff:ff:ff:ff:ff:ff" if nh != '::': ip6 = nh if ns_cache.has_key(ip6): mac, timeout = ns_cache[ip6] if timeout and (time.time()-timeout < NEIGHTIMEOUT): return mac nsma = in6_getnsma(socket.inet_pton(socket.AF_INET6, ip6)) d = socket.inet_ntop(socket.AF_INET6, nsma) dm = in6_getnsmac(nsma) p = Ether(dst=dm)/IPv6(dst=d, src=a, hlim=255) p /= ICMPv6ND_NS(tgt=ip6) p /= ICMPv6NDOptSrcLLAddr(lladdr=get_if_hwaddr(iff)) res = srp1(p, type=ETH_P_IPV6, iface=iff, timeout=2, verbose=0) if res is not None: mac = res.lladdr ns_cache[ip6] = (mac,time.time()) return mac return None class Net6(Gen): # syntax ex. fec0::/126 """Generate a list of IPv6s from a network address or a name""" name = "ipv6" ipaddress = re.compile(r"^([a-fA-F0-9:]+)(/[1]?[0-3]?[0-9])?$") def __init__(self, net): self.repr = net tmp = net.split('/')+["128"] if not self.ipaddress.match(net): tmp[0]=socket.getaddrinfo(tmp[0], None, socket.AF_INET6)[0][-1][0] netmask = int(tmp[1]) self.net = socket.inet_pton(socket.AF_INET6, tmp[0]) self.mask = in6_cidr2mask(netmask) self.plen = netmask def __iter__(self): def m8(i): if i % 8 == 0: return i tuple = filter(lambda x: m8(x), xrange(8, 129)) a = in6_and(self.net, self.mask) tmp = map(lambda x: x, struct.unpack('16B', a)) def parse_digit(a, netmask): netmask = min(8,max(netmask,0)) a = (int(a) & (0xffL<>(8-netmask)))+1) return a self.parsed = map(lambda x,y: parse_digit(x,y), tmp, map(lambda x,nm=self.plen: x-nm, tuple)) def rec(n, l): if n and n % 2 == 0: sep = ':' else: sep = '' if n == 16: return l else: ll = [] for i in xrange(*self.parsed[n]): for y in l: ll += [y+sep+'%.2x'%i] return rec(n+1, ll) return iter(rec(0, [''])) def __repr__(self): return "" % self.repr ####################################### ## IPv6 addresses manipulation tools ## ####################################### # TODO : Ca ne doit pas etre le meilleur moyen de faire. A revoir def in6_isaddr6to4(x): our = socket.inet_pton(socket.AF_INET6, x)[0:2] prefix = socket.inet_pton(socket.AF_INET6, "2002::")[0:2] return prefix == our # Temporary Teredo prefix (MS Pool, 6bone@) : 3ffe:83f1 # Should be modified before 6 June 2006 (6bone phaseout deadline) def in6_isaddrTeredo(x): our = socket.inet_pton(socket.AF_INET6, x)[0:4] teredoprefix = socket.inet_pton(socket.AF_INET6, "3ffe:831f::")[0:4] return teredoprefix == our def in6_iseui64(x): eui64 = socket.inet_pton(socket.AF_INET6, '::ff:fe00:0') x = in6_and(x, eui64) return x == socket.inet_pton(socket.AF_INET6, '::ff:fe00:0') def in6_isanycast(x): # RFC 2526 if in6_iseui64(x): x = in6_and(x, socket.inet_pton(socket.AF_INET6, '::fdff:ffff:ffff:ff10')) print socket.inet_ntop(socket.AF_INET6, x) return x == socket.inet_pton(socket.AF_INET6, '::fdff:ffff:ffff:ff10') else: # not EUI-64 #| n bits | 121-n bits | 7 bits | #+---------------------------------+------------------+------------+ #| subnet prefix | 1111111...111111 | anycast ID | #+---------------------------------+------------------+------------+ # | interface identifier field | warning('in6_isanycast(): TODO not EUI-64') return 0 def in6_hasConedFlag(x): return socket.inet_pton(socket.AF_INET6, x)[8:10] == '\x80\x00' def in6_bitops(a1, a2, operator=0): a1 = struct.unpack('4I', a1) a2 = struct.unpack('4I', a2) fop = [ lambda x,y: x | y, lambda x,y: x & y, lambda x,y: x ^ y ] ret = map(fop[operator%len(fop)], a1, a2) t = ''.join(map(lambda x: struct.pack('I', x), ret)) return t def in6_or(a1, a2): return in6_bitops(a1, a2, 0) def in6_and(a1, a2): return in6_bitops(a1, a2, 1) def in6_xor(a1, a2): return in6_bitops(a1, a2, 2) def in6_cidr2mask(m): t = [] for i in xrange(0, 4): t.append(max(0, 2**32 - 2**(32-min(32, m)))) m -= 32 mask = ''.join(map(lambda x: struct.pack('!I', x), t)) return mask def in6_getnsma(a): # return link-local solicited-node multicast address for given address r = in6_and(a, socket.inet_pton(socket.AF_INET6, '::ff:ffff')) r = in6_or(socket.inet_pton(socket.AF_INET6, 'ff02::1:ff00:0'), r) return r def in6_getnsmac(a): # return multicast Ethernet address associated with multicast v6 destination a = struct.unpack('16B', a)[-4:] mac = '33:33:' mac += ':'.join(map(lambda x: '%.2x' %x, a)) return mac def in6_getha(a): r = in6_and(socket.inet_pton(socket.AF_INET6, a), in6_cidr2mask(64)) r = in6_or(r, socket.inet_pton(socket.AF_INET6, '::fdff:ffff:ffff:fffe')) return socket.inet_ntop(socket.AF_INET6, r) def in6_ptop(str): # return str as a compressed address return socket.inet_ntop(socket.AF_INET6, socket.inet_pton(socket.AF_INET6, str)) # return True if address belongs to _allocated_ link-local (fe80::/64). def in6_islladdr(str): temp = socket.inet_pton(socket.AF_INET6, str)[:8] return ('\xfe\x80' + '\x00'*6) == temp def in6_isaddrllallnodes(str): return (socket.inet_pton(socket.AF_INET6, "ff02::1") == socket.inet_pton(socket.AF_INET6, v6tgt)) def in6_chksum(nh, u, p): ph6 = PseudoIPv6() ph6.nh = nh rthdr = 0 hahdr = 0 while u != None and not isinstance(u, IPv6): if (isinstance(u, IPv6OptionHeaderRouting) and u.segleft != 0 and len(u.addresses) != 0): rthdr = u.addresses[-1] elif isinstance(u, IPv6OptionHeaderHomeAddress): hahdr = u.ha u = u.underlayer if u == None: warning("No IPv6 underlayer to compute checksum. Leaving null.") return p if hahdr: ph6.src = hahdr else: ph6.src = u.src if rthdr: ph6.dst = rthdr else: ph6.dst = u.dst ph6.uplen = len(p) ph6s = str(ph6) return checksum(ph6s+p) class IP6Field(Field): def __init__(self, name, default): Field.__init__(self, name, default, "16s") def h2i(self, pkt, x): if type(x) is str: try: x = in6_ptop(x) except socket.error: x = Net6(x) elif type(x) is list: x = map(Net6, x) return x def i2m(self, pkt, x): return socket.inet_pton(socket.AF_INET6, x) def m2i(self, pkt, x): return socket.inet_ntop(socket.AF_INET6, x) def any2i(self, pkt, x): return self.h2i(pkt,x) def i2repr(self, pkt, x): if x is None: return self.h2i(pkt,x) elif not isinstance(x, Net6) and not type(x) is list: if in6_isaddrTeredo(x): # print Teredo info server, flag, maddr, mport = teredoAddrExtractInfo(x) return "%s [Teredo srv: %s cli: %s:%s]" % (self.i2h(pkt, x), server, maddr,mport) elif in6_isaddr6to4(x): # print encapsulated address addr = socket.inet_pton(socket.AF_INET6, x) vaddr = socket.inet_ntop(socket.AF_INET, addr[2:6]) return "%s [6to4 GW: %s]" % (self.i2h(pkt, x), vaddr) return self.i2h(pkt, x) # No specific information to return class SourceIP6Field(IP6Field): def __init__(self, name, dstname): IP6Field.__init__(self, name, None) self.dstname = dstname def i2m(self, pkt, x): if x is None: dst=getattr(pkt,self.dstname) iff,x,nh = conf.route6.route(dst) return IP6Field.i2m(self, pkt, x) def i2h(self, pkt, x): if x is None: dst=getattr(pkt,self.dstname) if isinstance(dst,Gen): r = map(conf.route6.route, dst) r.sort() if r[0] == r[-1]: x=r[0][1] else: warning("More than one possible route for %s"%repr(dst)) return None else: iff,x,nh = conf.route6.route(dst) return IP6Field.i2h(self, pkt, x) class AddNewFields(type): def __new__(cls, name, bases, dct): nf = dct["new_fields"] fields = None for b in bases: if hasattr(b,"fields_desc"): fields = b.fields_desc[:] break if fields is None: raise Exception("No fields_desc in superclasses") del(dct["new_fields"]) dct["fields_desc"] = fields + nf return super(AddNewFields, cls).__new__(cls, name, bases, dct) #################### ## IPv6 main part ## #################### ipv6nh = { 0:"IPv6 Hop-by-Hop Option", 4:"IP", 6:"TCP", 17:"UDP", 41:"IPv6", 43:"Routing Header", 44:"Fragment Header", 47:"GRE", 50:"ESP Header", 51:"AH Header", 58:"ICMPv6", 59:"No Next Header", 60:"Destination Option Header", 135:"Mobility Header"} ipv6nhcls = { 0: "IPv6OptionHeaderHopByHop", 4: "IP", 6: "TCP", 17: "UDP", 43: "IPv6OptionHeaderRouting", 44: "IPv6OptionHeaderFragment", #50: "ESPHeader", #51: "AHHeader", 58: "_ICMPv6", # ?!? 59: "Raw", #60: "_IPv6DestOpt", } class IP6ListField(StrField): islist = 1 def i2repr(self,pkt,x): s = [] if x == None: return "[]" for y in x: s.append('%s' % y) return "[ %s ]" % (" ".join(s)) def getfield(self, pkt, s): return "", self.m2i(pkt, s) def m2i(self, pkt, x): r = [] while len(x) != 0: r.append(socket.inet_ntop(socket.AF_INET6, x[:16])) x = x[16:] return r def i2m(self, pkt, x): s = '' for y in x: s += socket.inet_pton(socket.AF_INET6, y) return s class _IPv6GuessPayload: name = "Dummy class that implements guess_payload_class() for IPv6" def default_payload_class(self,p): if self.nh == 58: if len(p) > 2: type = ord(p[0]) if type == 139: # Node Info Query specific stuff code = ord(p[1]) if code == 0: return ICMPv6NIQueryIPv6 elif code == 1: return ICMPv6NIQueryName elif code == 2: return ICMPv6NIQueryIPv4 return Raw elif type == 140: # Node Info Reply specific stuff code = ord(p[1]) if code == 0: if len(p) > 6: qtype = ord(p[4])*16+ord(p[5]) if qtype in [ 0, 1 ]: return ICMPv6NIReplySuccess elif qtype == 2: return ICMPv6NIReplySuccessName elif qtype == 3: return ICMPv6NIReplySuccessIPv6 elif qtype == 4: return ICMPv6NIReplySuccessIPv4 elif code == 1: return ICMPv6NIReplyRefuse elif code == 2: return ICMPv6NIReplyUnknown return Raw return globals().get(icmp6typescls.get(type,"Raw"), "Raw") elif self.nh == 60: if len(p) > 6: otype = ord(p[6]) if otype == 201: return IPv6OptionHeaderHomeAddress return Raw elif self.nh == 135: if len(p) > 3: mhtype = ord(p[2]) if mhtype == 5: return IPv6MobilityHeader_BU elif mhtype == 6: return IPv6MobilityHeader_BACK elif mhtype == 7: return IPv6MobilityHeader_BE else: return Raw else: return globals().get(ipv6nhcls.get(self.nh,"Raw"), "Raw") class IPv6(_IPv6GuessPayload, Packet, IPTools): name = "IPv6" fields_desc = [ BitField("version" , 6 , 4), BitField("tc", 0, 8), #TODO: IPv6, ByteField ? BitField("fl", 0, 20), ShortField("plen", None), ByteEnumField("nh", 0, ipv6nh), ByteField("hlim", 64), SourceIP6Field("src", "dst"), IP6Field("dst", None) ] def mysummary(self): return "%s > %s (%i)" % (self.src,self.dst, self.nh) def post_build(self, p): if self.plen is None: l = len(p) - 40 p = p[:4]+struct.pack("!H", l)+p[6:] return p def extract_padding(self, s): l = self.plen return s[:l], s[l:] def hashret(self): if self.nh == 58 and isinstance(self.payload, _ICMPv6): if self.payload.type < 128: return self.payload.payload.hashret() elif (self.payload.type in [133,134,135,136,144,145]): return struct.pack("B", self.nh)+self.payload.hashret() nh = self.nh sd = self.dst ss = self.src if self.nh == 43 and isinstance(self.payload, IPv6OptionHeaderRouting): # With routing header, the destination is the last # address of the IPv6 list if segleft > 0 nh = self.payload.nh try: sd = self.addresses[-1] except IndexError: sd = '::1' # TODO: big bug with ICMPv6 error messages as the destination of IPerror6 # could be anything from the original list ... if 1: sd = socket.inet_pton(socket.AF_INET6, sd) for a in self.addresses: a = socket.inet_pton(socket.AF_INET6, a) sd = strxor(sd, a) sd = socket.inet_ntop(socket.AF_INET6, sd) if self.nh == 60 and isinstance(self.payload, IPv6OptionHeaderHomeAddress): nh = self.payload.nh ss = self.payload.ha if conf.checkIPsrc and conf.checkIPaddr: sd = socket.inet_pton(socket.AF_INET6, sd) ss = socket.inet_pton(socket.AF_INET6, self.src) return struct.pack("B",nh)+self.payload.hashret() else: return struct.pack("B", nh)+self.payload.hashret() def answers(self, other): if not isinstance(other, IPv6): return 0 if conf.checkIPaddr: ss = socket.inet_pton(socket.AF_INET6, self.src) sd = socket.inet_pton(socket.AF_INET6, self.dst) os = socket.inet_pton(socket.AF_INET6, other.src) od = socket.inet_pton(socket.AF_INET6, other.dst) # other.dst was a mutlicast address # sd == od if (in6_and(sd, in6_cidr2mask(8)) != socket.inet_pton(socket.AF_INET6, 'ff00::') and in6_xor(sd, od) != socket.inet_pton(socket.AF_INET6, '::')): # other.dst was an anycast address # os == ss #if (in6_isanycast(od) # and in6_xor(os, ss) != socket.inet_pton(socket.AF_INET6, '::')): if in6_xor(sd, os) != socket.inet_pton(socket.AF_INET6, '::'): return 0 if self.nh == 58 and isinstance(self.payload, _ICMPv6) and self.payload.type < 128: # ICMPv6 Error message -> generated by IPv6 packet # Note : at the moment, we jump the ICMPv6 specific class # to call answers() method of erroneous packet (over # initial packet). There can be cases where an ICMPv6 error # class could implement a specific answers method that perform # a specific task. Currently, don't see any use ... return self.payload.payload.answers(other) elif other.nh == 43 and isinstance(other.payload, IPv6OptionHeaderRouting): return self.payload.answers(other.payload.payload) # Buggy if self.payload is a IPv6OptionHeaderRouting elif other.nh == 60 and isinstance(other.payload, IPv6OptionHeaderHomeAddress): return self.payload.payload.answers(other.payload.payload) else: if (self.nh != other.nh): return 0 return self.payload.answers(other.payload) import scapy scapy.IPv6 = IPv6 class IPerror6(IPv6): name = "IPv6 in ICMPv6" def answers(self, other): if not isinstance(other, IPv6): return 0 sd = socket.inet_pton(socket.AF_INET6, self.dst) ss = socket.inet_pton(socket.AF_INET6, self.src) od = socket.inet_pton(socket.AF_INET6, other.dst) os = socket.inet_pton(socket.AF_INET6, other.src) # Make sure that the ICMPv6 error is related to the packet scapy sent if isinstance(self.underlayer, _ICMPv6) and self.underlayer.type < 128: #if sd == od: if ss == os and sd == od: return self.payload.answers(other.payload) #else: # return 0 if not conf.checkIPsrc or sd != od: if ss != os and self.nh != other.nh: return 0 return self.payload.answers(other.payload) def mysummary(self): return Packet.mysummary(self) icmp6typescls = { 1: "ICMPv6DestUnreach", 2: "ICMPv6PacketTooBig", 3: "ICMPv6TimeExceeded", 4: "ICMPv6ParamProblem", 128: "ICMPv6EchoRequest", 129: "ICMPv6EchoReply", 130: "ICMPv6MLQuery", 131: "ICMPv6MLReport", 132: "ICMPv6MLDone", 133: "ICMPv6ND_RS", 134: "ICMPv6ND_RA", 135: "ICMPv6ND_NS", 136: "ICMPv6ND_NA", 137: "ICMPv6ND_Redirect", #138: Do Me 139: "ICMPv6NIQuery", 140: "ICMPv6NIReply", 141: "ICMPv6ND_INDSol", 142: "ICMPv6ND_INDAdv", 144: "ICMPv6HAADRequest", 145: "ICMPv6HAADReply", 146: "ICMPv6MPSol", 147: "ICMPv6MPAdv" } icmp6types = { 1 : "Destination unreachable", 2 : "Packet too big", 3 : "Time exceeded", 4 : "Parameter problem", 128 : "Echo request", 129 : "Echo reply", 130 : "MLD Query", 131 : "MLD Report", 132 : "MLD Done", 133 : "Router Solicitation", 134 : "Router Advertisement", 135 : "Neighbor Solicitation", 136 : "Neighbor Advertisement", 137 : "Redirect", 138 : "Router Renumbering", 139 : "ICMP Node Information Query", 140 : "ICMP Node Information Response", 141 : "Inverse Neighbor Discovery Solicitation Message", 142 : "Inverse Neighbor Discovery Advertisement Message", 143 : "Version 2 Multicast Listener Report", 144 : "Home Agent Address Discovery Request Message", 145 : "Home Agent Address Discovery Reply Message", 146 : "Mobile Prefix Solicitation", 147 : "Mobile Prefix Advertisement", 148 : "Certification Path Solicitation", 149 : "Certification Path Advertisement" } # ipv6 pseudo-header for checksum computation class PseudoIPv6(Packet): name = "Pseudo IPv6 Header" fields_desc = [ IP6Field("src", "::"), IP6Field("dst", "::"), ShortField("uplen", None), BitField("zero", 0, 24), ByteField("nh", 0) ] class _IPv6OptionHeader(_IPv6GuessPayload, Packet): name = 'Abstract IPV6 Option Header' aliastypes = [IPerror6, IPv6] # TODO ... import scapy scapy._IPv6OptionHeader = _IPv6OptionHeader # Decoding only ... class HopByHopOptionsField(StrLenField): islist=1 def getfield(self, pkt, s): return self._do(pkt, s) def m2i(self, pkt, x): return self._do(pkt,x)[1] def _do(self, pkt, x): # TODO: change this method #print "_do x=%s len=%d" % (x, len(x)) opt = [] l = getattr(pkt, self.fld) l = 2**l l = l*8 - 2 while l > 0: o = ord(x[0]) if o == 0: # Pad1 l = l-1 opt.append('Pad1') x = x[1:] elif o == 1:# PadN t = ord(x[1]) x = x[t+2:] l = l-(t+2) opt.append(('PadN', t+2)) elif o == 0x5: # Router Alert - RFC 2711 if ord(x[1]) == 2: l = l-4 v = struct.unpack('!B', x[2])[0] v << 8 v = v + struct.unpack('!B', x[3])[0] x = x[4:] opt.append(('RouterAlert', v)) else: continue return x,opt class IPv6OptionHeaderHopByHop(_IPv6OptionHeader): name = "IPv6 Option Header Hop-by-Hop" fields_desc = [ ByteEnumField("nh", 0, ipv6nh), FieldLenField("len", None, "data", "B"), # (2^len *8) -2 bytes #StrLenField("options", "", "len") #StrFixedLenField("options", "", 6) # Should be a list of options ... HopByHopOptionsField("options", [], "len") ] overload_fields = {IPv6: { "nh": 0 }} # Take "len" field of the class class IP6RoutingHeaderListField(IP6ListField): islist=1 def getfield(self, pkt, s): l = 8*getattr(pkt, "len") return s[l:], self.m2i(pkt, s[:l]) class IPv6OptionHeaderRouting(_IPv6OptionHeader): name = "IPv6 Option Header Routing" fields_desc = [ ByteEnumField("nh", 0, ipv6nh), ByteField("len", None), ByteField("type", 0), ByteField("segleft", None), BitField("reserved", 0, 32), # There is meaning in this field ... IP6RoutingHeaderListField("addresses", []) ] overload_fields = {IPv6: { "nh": 43 }} def post_build(self, pkt): if self.len is None: pkt = pkt[0]+struct.pack("!B", 2*len(self.addresses))+pkt[2:] if self.segleft is None: pkt = pkt[:3]+struct.pack("!B", len(self.addresses))+pkt[4:] return _IPv6OptionHeader.post_build(self, pkt) class _ICMPv6(Packet): name = "ICMPv6 dummy class" overload_fields = {IPv6: {"nh": 58}} def post_build(self, p): if self.cksum == None: chksum = in6_chksum(58, self.underlayer, p) p = p[:2]+struct.pack("!H", chksum)+p[4:] return p def hashret(self): return self.payload.hashret() def answers(self, other): # isinstance(self.underlayer, _IPv6OptionHeader) may introduce a bug ... if isinstance(self.underlayer, IPerror6) or isinstance(self.underlayer, _IPv6OptionHeader) and isinstance(other, _ICMPv6): if not ((self.type == other.type) and (self.code == other.code)): return 0 return 1 return 0 ipv6Opt = { 0 : "IPv6 Pad1 Option", 1 : "IPv6 PadN Option"} ipv6HopByHopRoutingHeaderOpt = { 0: "IPv6 Hop-by-Hop Routing Header Option" } ipv6HopByHopRoutingHeaderOptval = { 0: "Datagram contains a MLD message", 1: "Datagram contains RSVP message", 2: "Datagram contains an Active Network message" } # Hop-by-Hop Options # Routing (Type 0) # Fragment # Destination Options # Authentication # Encapsulating Security Payload # Options TLV # type: two highest-order bits code problems; third bit code for en-route changes of encoded data # len # (data) # Padding # Pad1: adds one byte of padding # PadN: N-2 octets of padding are present in V field (type=1, len=N-2) class Pad1(ByteField): def __init__(self, name): ByteField.__init__(self, name, 0) class PadN(IntField): def __init__(self, name, len): IntField.__init__(self, name, len) def getfield(self, pkt, s): l = ord(s[1]) return s[l+2:], l def addfield(self, pkt, s, val): h = struct.pack('!B', 1) + struct.pack('!B', val) return s+h+(val*'\0') def i2m(self, pkt, x): h = struct.pack('!B', 1) + struct.pack('!B', val) return pkt+h class IPv6OptionHeaderFragment(_IPv6OptionHeader): name = "IPv6 Fragment Header Option" fields_desc = [ ByteEnumField("nh", None, ipv6nh), BitField("len", 0, 8), BitField("offset", 0, 13), BitField("res", 0, 2), BitField("m", 0, 1), IntField("id", 0) ] overload_fields = {IPv6: { "nh": 44 }} # Ce serait pas mal de transformer le premier champs en 2 + 6 # On devrait avoir un post_build pour faire en sorte de mettre a une # valeur donnee le champ val en fonction de certaines info, comme la # presence d'un message MLD dans le paquet. Voir si on le fait dans le # post_build ou dans un overload_fields class IPv6HopByHopRoutingHeaderOpt(_IPv6OptionHeader): # NEVER TESTED !!! name = "IPv6 Hop-by-Hop Routing Header Option" fields_desc = [ ByteEnumField("type",0, ipv6HopByHopRoutingHeaderOpt), ByteField("len",2), ShortEnumField("val",None,ipv6HopByHopRoutingHeaderOptval)] # Temporairement, pour pas que ca gueule trop : class IPv6PacketField(StrField): name = "toto" class _ICMPv6Error(_ICMPv6): name = "ICMPv6 errors dummy class" def guess_payload_class(self,p): return IPerror6 # Voir ce qui concerne la gestion de la contrainte de MTU minimale IPv6. class ICMPv6DestUnreach(_ICMPv6Error): name = "ICMPv6 Destination Unreachable" fields_desc = [ ByteEnumField("type",1, icmp6types), ByteEnumField("code",0, { 0: "No route to destination", 1: "Communication with destination administratively prohibited", 3: "Address unreachable", 4: "Port unreachable" }), XShortField("cksum", None), XIntField("unused",0x00000000)] class ICMPv6PacketTooBig(_ICMPv6Error): name = "ICMPv6 Packet Too Big" fields_desc = [ ByteEnumField("type",2, icmp6types), ByteField("code",0), XShortField("cksum", None), IntField("mtu",1280)] class ICMPv6TimeExceeded(_ICMPv6Error): name = "ICMPv6 Time Exceeded" fields_desc = [ ByteEnumField("type",3, icmp6types), ByteField("code",{ 0: "hop limit exceeded in transit", 1: "fragment reassembly time exceeded"}), XShortField("cksum", None), XIntField("unused",0x00000000)] # The default pointer value is for the next header of the encapsulated # IPv6 packet. class ICMPv6ParamProblem(_ICMPv6Error): name = "ICMPv6 Parameter Problem" fields_desc = [ ByteEnumField("type",4, icmp6types), ByteEnumField("code",0, {0: "erroneous header field encountered", 1: "unrecognized Next Header type encountered", 2: "unrecognized IPv6 option encountered"}), XShortField("cksum", None), IntField("ptr",6)] # voir s'il n'y a pas moyen de randomiser les champs # identifier et sequence number # Un peu de code pour traquer les valeurs qui arrivent pourrait etre sympa # genre un ping responder. class ICMPv6EchoRequest(_ICMPv6): name = "ICMPv6 Echo Request" fields_desc = [ ByteEnumField("type", 128, icmp6types), ByteField("code", 0), XShortField("cksum", None), XShortField("id",0), XIntField("seq",0), StrField("data", "")] def mysummary(self): return self.sprintf("%name% (id: %id% seq: %seq%)") def hashret(self): return struct.pack("HI",self.id,self.seq)+self.payload.hashret() class ICMPv6EchoReply(ICMPv6EchoRequest): name = "ICMPv6 Echo Reply" __metaclass__ = ChangeDefaultValues new_default_values = { "type": 129 } def answers(self, other): # We could match data content between request and reply. return (isinstance(other, ICMPv6EchoRequest) and self.id == other.id and self.seq == other.seq and self.data == other.data) ############################################################################### # ICMPv6 Multicast Listener Discovery : A FAIRE ############################################################################### # tous les messages MLD sont emis avec une adresse source lien-locale # -> Y veiller dans le post_build si aucune n'est specifiee # La valeur de Hop-Limit doit etre de 1 # "and an IPv6 Router Alert option in a Hop-by-Hop Options # header. (The router alert option is necessary to cause routers to # examine MLD messages sent to multicast addresses in which the router # itself has no interest" # class _ICMPv6ML(_ICMPv6): fields_desc = [ ByteEnumField("type", 130, icmp6types), ByteField("code", 0), XShortField("cksum", None), ShortField("mrd", 0), ShortField("reserved", 0), IP6Field("mladdr",None)] # general queries are sent to the link-scope all-nodes multicast # address ff02::1, with a multicast address field of 0 and a MRD of # [Query Response Interval] # Default value for mladdr is set to 0 for a General Query, and # overloaded by the user for a Multicast Address specific query class ICMPv6MLQuery(_ICMPv6ML): # RFC 2710 name = "MLD - Multicast Listener Query" __metaclass__ = ChangeDefaultValues new_default_values = {"type": 130, "mrd":10000, "mladdr": "::"} # 10s for mrd overload_fields = {IPv6: { "dst": "ff02::1", "hlim": 1 }, IPv6HopByHopRoutingHeaderOpt: {"val": 0}} def hashret(self): if self.mladdr != "::": return struct.pack("HH",self.mladdr)+self.payload.hashret() else: return self.payload.hashret() class ICMPv6MLReport(_ICMPv6ML): # RFC 2710 name = "MLD - Multicast Listener Report" __metaclass__ = ChangeDefaultValues new_default_values = {"type": 131} overload_fields = {IPv6: {"hl": 1}, IPv6HopByHopRoutingHeaderOpt: {"val": 0}} # implementer le hashret et le answers # When a node ceases to listen to a multicast address on an interface, # it SHOULD send a single Done message to the link-scope all-routers # multicast address (FF02::2), carrying in its multicast address field # the address to which it is ceasing to listen class ICMPv6MLDone(_ICMPv6ML): # RFC 2710 name = "MLD - Multicast Listener Done" __metaclass__ = ChangeDefaultValues new_default_values = {"type": 132} overload_fields = {IPv6: { "dst": "ff02::2", "hlim": 1}, IPv6HopByHopRoutingHeaderOpt: {"val": 0}} ############################################################################### # ICMPv6 Neighbor Discovery (RFC 2461) ############################################################################### icmp6ndopts = { 1: "Source Link-Layer Address", 2: "Target Link-Layer Address", 3: "Prefix Information", 4: "Redirected Header", 5: "MTU", #6: #7: #8: 9: "Source Address List", 10: "Target Address List" } icmp6ndoptscls = { 1: "ICMPv6NDOptSrcLLAddr", 2: "ICMPv6NDOptDstLLAddr", 3: "ICMPv6NDOptPrefixInfo", 4: "ICMPv6NDOptRedirectedHdr", 5: "ICMPv6NDOptMTU", #6: 7: "ICMPv6NDOptAdvInterval", 8: "ICMPv6NDOptHAInfo", 9: "ICMPv6NDOptSrcAddrList", 10: "ICMPv6NDOptTgtAddrList" } class _ICMPv6NDGuessPayload: name = "Dummy ND class that implements guess_payload_class()" def guess_payload_class(self,p): if len(p) > 1: return globals().get(icmp6ndoptscls.get(ord(p[0]),"Raw"), "Raw") # s/Raw/ICMPv6NDOptUnknown/g ? class ICMPv6NDOptUnknown(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery Option - Scapy Unimplemented" fields_desc = [ ByteField("type",None), FieldLenField("len",None,"data","B",shift=-2), StrLenField("data","","len") ] # NOTE: len includes type and len field. Expressed in unit of 8 bytes # TODO: Revoir le coup du ETHER_ANY class ICMPv6NDOptSrcLLAddr(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery Option - Source Link-Layer Address" fields_desc = [ ByteField("type",1), ByteField("len",1), MACField("lladdr",ETHER_ANY) ] def mysummary(self): return self.sprintf("%name% %lladdr%") class ICMPv6NDOptDstLLAddr(ICMPv6NDOptSrcLLAddr): name = "ICMPv6 Neighbor Discovery Option - Destination Link-Layer Address" __metaclass__ = ChangeDefaultValues new_default_values = { 'type': 2 } class ICMPv6NDOptPrefixInfo(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery Option - Prefix Information" fields_desc = [ ByteField("type",3), ByteField("len",4), ByteField("prefixlen",None), BitField("L",1,1), BitField("A",1,1), BitField("R",0,1), BitField("reserved1",None,0), XIntField("validlifetime",0xffffffffL), XIntField("preferredlifetime",0xffffffffL), XIntField("reserved2",0x00000000), IP6Field("prefix","::") ] def mysummary(self): return self.sprintf("%name% %prefix%") # Faire un post_build pour le recalcul de la taille (en multiple de 8 octets) class ICMPv6NDOptRedirectedHdr(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery Option - Redirected Header" fields_desc = [ ByteField("type",4), ByteField("len",None), XShortField("reserved",0) ] # Voir la valeur de MTU a utiliser par defaut au lieu du 1280 class ICMPv6NDOptMTU(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery Option - MTU" fields_desc = [ ByteField("type",5), ByteField("len",1), XShortField("reserved",0), IntField("mtu",1280)] class ICMPv6NDOptAdvInterval(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery - Interval Advertisement" fields_desc = [ ByteField("type",7), ByteField("len",1), ShortField("reserved", 0), IntField("advint", 0) ] def mysummary(self): return self.sprintf("%name% %advint% milliseconds") class ICMPv6NDOptHAInfo(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Neighbor Discovery - Home Agent Information" fields_desc = [ ByteField("type",8), ByteField("len",1), ShortField("reserved", 0), ShortField("pref", 0), ShortField("lifetime", 1)] def mysummary(self): return self.sprintf("%name% %pref% %lifetime% seconds") class ICMPv6ND_RS(_ICMPv6NDGuessPayload, _ICMPv6): name = "ICMPv6 Neighbor Discovery - Router Solicitation" fields_desc = [ ByteEnumField("type", 133, icmp6types), ByteField("code",0), XShortField("cksum", None), IntField("reserved",0) ] overload_fields = {IPv6: { "nh": 58, "dst": "ff02::2", "hlim": 255 }} # TODO : gerer le draft router preference class ICMPv6ND_RA(_ICMPv6NDGuessPayload, _ICMPv6): name = "ICMPv6 Neighbor Discovery - Router Advertisement" fields_desc = [ ByteEnumField("type", 134, icmp6types), ByteField("code",0), XShortField("cksum", None), ByteField("chlim",0), BitField("M",0,1), BitField("O",0,1), BitField("H",0,1), BitField("reserved",0,5), ShortField("routerlifetime",0), IntField("reachabletime",0), IntField("restranstimer",0) ] overload_fields = {IPv6: { "nh": 58, "dst": "ff02::1", "hlim": 255 }} def answers(self, other): return isinstance(other, ICMPv6ND_RS) class ICMPv6ND_NS(_ICMPv6NDGuessPayload, _ICMPv6, Packet): name = "ICMPv6 Neighbor Discovery - Neighbor Solicitation" fields_desc = [ ByteEnumField("type",135, icmp6types), ByteField("code",0), XShortField("cksum", None), BitField("R",0,1), BitField("S",0,1), BitField("O",0,1), XBitField("reserved",0,29), IP6Field("tgt","::") ] overload_fields = {IPv6: { "nh": 58, "dst": "ff02::1", "hlim": 255 }} def mysummary(self): return self.sprintf("%name% (tgt: %tgt%)") def hashret(self): return self.tgt+self.payload.hashret() class ICMPv6ND_NA(ICMPv6ND_NS): name = "ICMPv6 Neighbor Discovery - Neighbor Advertisement" __metaclass__ = ChangeDefaultValues new_default_values = { 'type': 136, 'R': 1, 'O': 1 } def answers(self, other): return isinstance(other, ICMPv6ND_NS) and self.tgt == other.tgt # possible option : target link-layer option, Redirected header class ICMPv6ND_Redirect(_ICMPv6NDGuessPayload, _ICMPv6, Packet): name = "ICMPv6 Neighbor Discovery - Redirect" fields_desc = [ ByteEnumField("type",137, icmp6types), ByteField("code",0), XShortField("cksum", None), XIntField("reserved",0), IP6Field("tgt","::"), IP6Field("dst","::") ] overload_fields = {IPv6: { "nh": 58, "dst": "ff02::1", "hlim": 255 }} ############################################################################### # ICMPv6 Node Information : draft-ietf-ipngwg-icmp-name-lookups-12.txt ############################################################################### icmp6niqtypes = { 0: "NOOP", 1: "unused", 2: "Node Name", 3: "Node Addresses", 4: "IPv4 Address" } class _ICMPv6NIHashret(_ICMPv6): def hashret(self): return str(self.nonce) class _ICMPv6NIAnswers: def answers(self, other): return self.nonce == other.nonce # Buggy; always returns the same value during a session class NonceField(StrFixedLenField): def __init__(self, name, default): StrFixedLenField.__init__(self, name, default, 8) self.default = str(self.randval()) # Compute the NI group Address. Can take a FQDN as input parameter def computeNIGroupAddr(name): name = string.lower(name).split(".")[0] record = chr(len(name))+name h = md5.new(record) h = h.digest() addr = "ff02::2:%x%x:%x%x" % (ord(h[0]),ord(h[1]),ord(h[2]),ord(h[3])) return addr class NIQueryDataField(StrField): guessed_code = None def __init__(self, name, default): StrField.__init__(self, name, default) # Returns the part of s corresponding to the field and the rest of s def getfield(self, pkt, s): code = getattr(pkt, "code") if code == 0: # IPv6 Addr l = 16 elif code == 2: # IPv4 Addr l = 4 else: # Name or Unknown l = len(s) # TODO: Buggy ... return s[l:],s[:l] def addfield(self, pkt, s, val): code = getattr(pkt,"code") if code == 0: # IPv6 Addr return s+struct.pack("16s", self.i2m(pkt,val)) elif code == 2: # IPv4 Addr return s+struct.pack("4s", self.i2m(pkt,val)) else: # Name or Unknown return s+self.i2m(pkt,val) # internal 2 machine : converts internal representation interne to *network output* def i2m(self, pkt, x): code = getattr(pkt,"code") if code == 0: # IPv6 Addr return IP6Field('', x).i2m(pkt, x) elif code == 2: # IPv4 Addr self.data = IPField('', x) return IPField('', x).i2m(pkt, x) else: # Name or Unknown return DNSStrField('', x).i2m(pkt, x) class NIReplyNamesField(StrField): islist=1 def i2repr(self,pkt,x): s = [] if x == None: return "[]" for i in x: s.append('%s' % i) return "[ %s ]" % (" ".join(s)) def getfield(self, pkt, s): return "", self.m2i(pkt, s) def m2i(self, pkt, x): r = [] while x: o = ord(x[0]) x,t = DNSStrField('', '').getfield(pkt, x) r += [t] return r def i2m(self, pkt, x): s = '' for i in x: s += DNSStrField('', i).i2m(pkt, i) return s class NIReplyIP6Field(StrField): islist=1 def i2repr(self,pkt,x): s = [] if x == None: return "[]" for (x,y) in x: s.append('(ttl: %s, ip: %s)' % (x,y)) return "[ %s ]" % (" ".join(s)) def getfield(self, pkt, s): return "", self.m2i(pkt, s) def m2i(self, pkt, x): r = [] while x: x, ttl = IntField('', '').getfield(pkt, x) x, ip6 = IP6Field('', None).getfield(pkt, x) r += [(ttl, ip6)] return r def i2m(self, pkt, x): s = '' for (x,y) in x: s += struct.pack('!I', x)+IP6Field('', y).i2m(pkt, y) return s class NIReplyIP4Field(StrField): islist=1 def i2repr(self,pkt,x): s = [] if x == None: return "[]" for (x,y) in x: s.append('(ttl: %s, ip: %s)' % (x,y)) return "[ %s ]" % (" ".join(s)) def getfield(self, pkt, s): return "", self.m2i(pkt, s) def m2i(self, pkt, x): r = [] while x: x, ttl = IntField('', '').getfield(pkt, x) x, ip = IPField('', None).getfield(pkt, x) r += [(ttl, ip)] return r def i2m(self, pkt, x): s = '' for (x,y) in x: s += struct.pack('!I', x)+IPField('', y).i2m(pkt, y) return s class ICMPv6NIQuery(_ICMPv6NIHashret, Packet): name = "ICMPv6 Node Information Query" fields_desc = [ ByteEnumField("type", 139, icmp6types), ByteEnumField("code", 0, { 0: "IPv6 Query", 1: "Name Query", 2: "IPv4 Query"}), XShortField("cksum", None), ShortEnumField("qtype", 0, icmp6niqtypes), ByteField("unused", 0), ByteEnumField("flags", 0, { 2: "All unicast addresses", 4: "IPv4 addresses", 8: "Link-local addresses", 16: "Site-local addresses", 32: "Global addresses" }), NonceField("nonce", None), NIQueryDataField("data",None) ] # Cosmetic class - not meant to be use directly unless # the data field is added as payload. class ICMPv6NIReply(_ICMPv6NIAnswers, _ICMPv6NIHashret, Packet): name = "ICMPv6 Node Information Reply (Successful Reply)" fields_desc = [ ByteEnumField("type", 140, icmp6types), ByteEnumField("code", 0, { 0: "Successful Reply", 1: "Response Refusal", 3: "Unknown query type" }), XShortField("cksum", None), ShortEnumField("qtype", 0, icmp6niqtypes), ByteField("unused", 0), ByteEnumField("flags", 0, { 1: "Reply sert incomplete", 2: "All unicast addresses", 4: "IPv4 addresses", 8: "Link-local addresses", 16: "Site-local addresses", 32: "Global addresses" }), NonceField("nonce", None), # See classes below for data field support ] # Class implementing different data field according to qtype class ICMPv6NIReplySuccess(ICMPv6NIReply): name = "ICMPv6 Node Information Reply (Sucessful Reply)" class ICMPv6NIReplySuccessName(ICMPv6NIReply): name = "ICMPv6 Node Information Reply (Sucessful Reply - Name)" __metaclass__ = AddNewFields new_fields = [ IntField("ttl", 0), NIReplyNamesField("data", None) ] class ICMPv6NIReplySuccessIPv6(ICMPv6NIReply): name = "ICMPv6 Node Information Reply (Successful Reply - IPv6)" __metaclass__ = AddNewFields new_fields = [ NIReplyIP6Field("data", None) ] class ICMPv6NIReplySuccessIPv4(ICMPv6NIReply): name = "ICMPv6 Node Information Reply (Successful Reply - IPv4)" __metaclass__ = AddNewFields new_fields = [ NIReplyIP4Field("data", None) ] class ICMPv6NIReplyRefuse(ICMPv6NIReply): name = "ICMPv6 Node Information Reply (Responder refuses to supply answer)" __metaclass__ = ChangeDefaultValues new_default_values = { 'code': 1 } class ICMPv6NIReplyUnknown(ICMPv6NIReply): name = "ICMPv6 Node Information Reply (Qtype unknown to the responder)" __metaclass__ = ChangeDefaultValues new_default_values = { 'code': 2 } # Convenient classes to generate queries # - the expected data is an IPv6 address # - refer to comments for testing class ICMPv6NIQueryIPv6(ICMPv6NIQuery): # IPv6 -> IPv6 [ aka NI ping ] name = "ICMPv6 Node Information Query (IPv6 Query)" __metaclass__ = ChangeDefaultValues new_default_values = { 'code': 0, 'qtype': 3 } class ICMPv6NIQueryLocal(ICMPv6NIQuery): # IPv6-> local IPv6 name = "ICMPv6 Node Information Query (IPv6 Link-Local Address Query)" __metaclass__ = ChangeDefaultValues new_default_values = { 'code': 0, 'qtype': 3, 'flags': 8 } class ICMPv6NIQueryName(ICMPv6NIQuery): # IPv6-> Name name = "ICMPv6 Node Information Query (IPv6 Address Query)" __metaclass__ = ChangeDefaultValues new_default_values = { 'code': 0, 'qtype': 2 } class ICMPv6NIQueryIPv4(ICMPv6NIQuery): # IPv6 -> IPv4 name = "ICMPv6 Node Information Query (IPv4 Address Query)" __metaclass__ = ChangeDefaultValues new_default_values = { 'code': 0, 'qtype': 4 } ############################################################################### # ICMPv6 Inverse Neighbor Discovery : RFC 3122 ############################################################################### # Developper un field specifique pour la liste d'adresse # Voir le calcul de la longueur en fonction du nombre d'elements # de la liste d'adresses. # n = (len-1)/2 class ICMPv6NDOptSrcAddrList(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Inverse Neighbor Discovery Option - Source Address List" fields_desc = [ ByteField("type",9), ByteField("len",None), StrFixedLenField("reserved","",6) #IP6ListField("addrlist",None,"len") ] class ICMPv6NDOptTgtAddrList(_ICMPv6NDGuessPayload, Packet): name = "ICMPv6 Inverse Neighbor Discovery Option - Target Address List" fields_desc = [ ByteField("type",10), ByteField("len",None), StrFixedLenField("reserved","",6) #IP6ListField("addrlist",None,"len") ] # RFC3122 # Options requises : source lladdr et target lladdr # Autres options valides : source address list, MTU # - Comme precise dans le document, il serait bien de prendre l'adresse L2 # demandee dans l'option requise target lladdr et l'utiliser au niveau # de l'adresse destination ethernet si aucune adresse n'est precisee # - ca semble pas forcement pratique si l'utilisateur doit preciser toutes # les options. # Ether() must use the target lladdr as destination class ICMPv6ND_INDSol(_ICMPv6NDGuessPayload, _ICMPv6): name = "ICMPv6 Inverse Neighbor Discovery Solicitation" fields_desc = [ ByteEnumField("type",141, icmp6types), ByteField("code",0), ShortField("cksum",None), XIntField("reserved",0) ] overload_fields = {IPv6: { "nh": 58, "dst": "ff02::1", "hlim": 255 }} # Options requises : target lladdr, target address list # Autres options valides : MTU class ICMPv6ND_INDAdv(_ICMPv6NDGuessPayload, _ICMPv6): name = "ICMPv6 Inverse Neighbor Discovery Advertisement" fields_desc = [ ByteEnumField("type",142, icmp6types), ByteField("code",0), ShortField("cksum",None), XIntField("reserved",0) ] overload_fields = {IPv6: { "nh": 58, "dst": "ff02::1", "hlim": 255 }} #### Early Teredo support # returns server, flag, mapped addr and mapped port from Teredo address def teredoAddrExtractInfo(x): addr = socket.inet_pton(socket.AF_INET6, x) server = socket.inet_ntop(socket.AF_INET, addr[4:8]) flag = struct.unpack("H",addr[8:10])[0] mappedport = struct.unpack("H",strxor(addr[10:12],'\xff'*2))[0] mappedaddr = socket.inet_ntop(socket.AF_INET, strxor(addr[12:16],'\xff'*4)) return server, flag, mappedaddr, mappedport class TeredoPeerList: l={} # simply add an entry to the list of peers (with the time) # address is the pair (v4dst, v4port). def add(self, dst6, address): # TODO : see if peer GW can change. self.l[dst6]=address, time.time() # returns 3600 if peer is not known, else, number of seconds # since last contact def elapsed(self, dst6): if self.l.has_key(dst6): return (time.time() - self.l[dst6][1]) return 3600 # remove the entry from the list of peers # TODO : suppress me, I'm useless def remove(self, dst6): if self.l.has_key(dst6): del self.l[dst6] def has(self, dst6): return self.l.has_key(dst6) def get(self, dst6): if self.l.has_key(dst6): return self.l[dst6][0] else: return None # if time associated with return the pair (v4dst, port) # TODO : integrer la limite de temps dans le dictionnaire. def get_v4dst(self, dst6): if self.l.has_key(dst6): address, t = self.l[dst6] else: return None if time.time() - t < 30: return address else: del self.l[dst6] return None return None class TeredoSocket: # variables de classe, a conserver status="initial" # interface status coned=0 # are we behind Cone NAT oursrvaddr = None # our Teredo server address ourv6addr= None # Our IPv6 Teredo address serviceport = None # Our service port maddr = None # our mapped address mport = None # our mapped port lastemission = None # Last time we contact our server s = None # UDP socket bound to service port peers=TeredoPeerList() relays={} # Pieces of information that WILL be used for secure qualification clientid=None # Client identifier secret=None # Shared secret authalg=None # Authentication Algorithm # What I do : # - open a UDP socket for receiving Teredo traffic # - perform Teredo qualification procedure # - launch the connection maintainer that peridically sent bubbles # to our Teredo server. def __init__(self, server1=None, server2=None, serviceport=None, iface=None, verbose=None): # Open a UDP socket if self._openTeredoSocket(serviceport=serviceport, iface=None, verbose=verbose) is False: return self.ourv6addr = self._qualify(server1, server2, serviceport=serviceport,verbose=verbose) # In a near future, we could maintain the communication with our server # self._maintain() def __close__(self): self.s.close() def _openTeredoSocket(self, serviceport=None, iface=None, verbose=1): """ Internal function. Opens a UDP socket to listen for incoming Teredo traffic on service port. 1 is return on error. 0 on success. serviceport : specifies a specific TEredo service port for the socket. if this parameter is not provided, a random port is used. iface : specifies the interface or address to bind our socket on. if this parameter is omitted, conf.iface is used. verbose : should I print error messages ? """ self.s = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) if iface is None and conf.iface is not None: iface = conf.iface retries=0 if serviceport is None: maxretries=4 else: # no port randomization. We won't retry on error maxretries=1 while(retries < maxretries): if maxretries > 1: # implies port randomization serviceport = 0 while serviceport < 1024: serviceport = RandShort() try: self.s.bind((get_if_addr(iface),serviceport)) break except: if verbose: print "Unable to bind socket to port %d on %s : %s" % (serviceport, iface, sys.exc_info()[1]) retries += 1 if retries == maxretries: if verbose: print "After %d retries, we were Unable to bind our socket. Exiting" % retries return False self.serviceport=serviceport return True def _sendRSAndReceiveRA(self, server, src6lladdr, T): """ Sends a Router Solicitation message to our Teredo server and filters incoming traffic on our socket against Router Advertisements. You should be aware that other kind of traffic received on the socket is dropped. No RA validation is performed. server : the Teredo server we are sending traffic to. src6lladdr : the IPv6 source link-local address used as source of RS message. T : The time we wait for a RA after a RS has been sent. """ s = self.s # TODO : construct the RS with a Teredo Auth header. Made it optional ? rs = TeredoAuth()/IPv6(src=src6lladdr, dst="ff02::2")/ICMPv6ND_RS() # Send RS to our server # TODO : # - test if sending was OK # - something to do with blocking/nonblocking mode ? s.sendto(str(rs), (server, 3544)) emitted = time.time() elapsed = 0 while (T - elapsed) > 0: s.settimeout(float(T-elapsed)) try: # TODO : 200 ? string, address = s.recvfrom(200) # TODO : we should have different kinds of exception. This # one is quite dummy. except: # Houston, we have a problem ! elapsed = time.time() - emitted continue ip, port = address coned = in6_hasConedFlag(src6lladdr) if coned or (not coned and ip == server and port == 3544): # At least reply seems to come from our server # Let's try to dissect the response try: resp = TeredoAuth(string) except: elapsed = time.time() - emitted continue if rs > resp: return resp elapsed = time.time() - emitted continue return None def _validateTeredoRA(self, r, server, src6lladdr, verbose=0): """ This internal function validates received RA message (Teredo Origin Indication header must be included) and returns mapped UDP IP and port from Origin Indication header. None is returned on error, a message being displayed if verbose parameter is not 0. r : UDP Payload IPv6 packet to be validated as RA. server : IPv4 address of the Teredo server. Used to validate correctness of returned prefix. src6lladdr : Link-Local IPv6 address used as source in RS. Expected as destination address in RA. Used for validation. verbose : should I be verbose on error. """ # TODO : should deal with Athentication Header more specifically # Let's get Teredo Origin Indication (packet can have TeredoAuth _before_) r = r[TeredoOrigInd] if r is None: if verbose: print "No Teredo Origin Indication in Response from server. Discarding." # TODO : revoir la sortie return None # mapped address and port are stored to be returned after validation oport = r.oport origip = r.origip r = r.payload # After Teredo Origin Indication, IPv6 must follow if not isinstance(r, IPv6): if verbose: print "Teredo Origin Indication not followed by IPv6 packet. Discarding." # TODO : revoir la sortie return None # IPv6 destination address of RA must be source address of RS, # given as parameter. if r.dst != src6lladdr: if verbose: print "RA Destination address differs from source address of RS. Discarding." # TODO : revoir la sortie return None if r.hlim != 255: if verbose: print "Hop-Limit in RA is %d. Expected 255. Discarding." % r.hlim # TODO : revoir la sortie return None # We also store the source address of the RA and test it is link-local srvlladdr = r.src if not in6_islladdr(srvlladdr): if verbose: print "Source address of RA is not Link-Local. Discarding." # TODO : revoir la sortie return None r = r.payload # validating IPv6 packet is carrying a RA. if not isinstance(r, ICMPv6ND_RA): if verbose: print "Packet is not a valid RA. Discarding." # TODO : revoir la sortie return None r = r.payload # As specified in Teredo draft, RA must contain exactly one Prefix # Information Option, with a valid Teredo prefix. This is the purpose # of following tests. if not ICMPv6NDOptPrefixInfo in r: if verbose: print "RA does not contain a Prefix Information option. Discarding." # TODO : revoir la sortie return None r = r[ICMPv6NDOptPrefixInfo] if r.prefixlen != 64: if verbose: print "Carried prefix length is invalid. Discarding." # TODO : revoir la sortie return None # test if carried prefix is Teredo prefix = r.prefix if not in6_isaddrTeredo(prefix): if verbose: print "Prefix is not a valid Teredo Prefix. Discarding." # TODO : revoir la sortie return None if (teredoAddrExtractInfo(prefix)[0] != server): if verbose: print "Address encapsulated in received Teredo prefix does" print "not match Teredo server address. Discarding." return None # TODO : no test is performed against the public/private status of # server address. r = r.payload # Test if RA does not contain another ND Prefix Information option. if ICMPv6NDOptPrefixInfo in r: if verbose: print "RA contains 2 Prefix Information options. Discarding." # TODO : revoir la sortie return None return origip, oport # Note : no verification is performed on given parameters. def _constructTeredoAddress(self, server, cone, maddr, mport): """ This function construct a Teredo address using given Teredo server address, NAT type, mapped IP and port (typically extracted from Teredo Origin Indication header). Teredo address is returned in network format. server : the /64 Teredo prefix in printable format. For example, 3ffe:831f:4004:1950::. Typically the prefix found in a Prefix Information Option carried in a RA. cone : 1 for cone NAT flag. 0 for restricted NAT flag. maddr : mapped address of the client (public IPv4 address of the NAT GW), in printable format. mport : mapped port of the client (UDP port associated with the service port of the client on the NAT GW). """ # TODO : create a global variable for the teredo /32 prefix prefix = socket.inet_pton(socket.AF_INET6,"3ffe:831f::")[:4] prefix += socket.inet_pton(socket.AF_INET,server) # Create the interface ID by packing the 3 elements together maddr = socket.inet_pton(socket.AF_INET, maddr) if cone: flag = '\x80\x00' else: flag = '\x00\x00' ifaceid = flag+struct.pack("H",mport ^ 0xffff)+strxor('\xff'*4,maddr) return socket.inet_ntop(socket.AF_INET6, prefix+ifaceid) # This function performs the qualification procedure def _qualify(self, server1="64.4.25.80",server2="64.4.25.81",serviceport=None, verbose=0, testCone=0, T=4, N=3): """ This internal function performs the qualification procedure for the Teredo interface. server1 : our Teredo server address. By default, we use Microsoft one. verbose : do I display error messages. 0 by default. testCone : start qualification procedure by testing if we are behind a Cone NAT. Not performed by default. Makes qualification procedure faster. T : timeout value used when waiting for a RA from the Teredo server after a RS has been sent. Same value is used for Cone NAT flaged and non Cone NAT flaged RS. Default value is 4 seconds as specified in RFC 2461 N : number of retries if no RA is received for our RS. Same value is used for Cone NAT and non cone NAT test procedure. Default value is 3 as specified by RFC 2461. """ # simple sanity checks against current status if self.status != "initial": if self.status == "qualified": if verbose: print "Passing state from qualified to initial." print "Restarting qualification procedure." # Autre chose a reinitialiser ? self.status = "initial" elif self.status == "off-line": if verbose: print "Passing state from off-line to initial." print "Restarting qualification procedure." # Autre chose a reinitialiser ? self.status = "initial" else: if verbose: print "Don't modify internal variables by hand !!!" print "value \"%s\" for status variable is incorrect" % self.status if testCone: # Should I test if we are behind a Cone NAT ? currentShot=1 while currentShot <= N: # Our "Cone NAT" Link-Layer address (the one used by Win XP). src6lladdr = "fe80::8000:5445:5245:444f" r = self._sendRSAndReceiveRA(server1, src6lladdr, T) if r is None: # No RA received after T seconds currentShot += 1 continue mappedvals = self._validateTeredoRA(r, server1, src6lladdr, verbose=verbose) if mappedvals is None: # Validation failed : our server sends weird packets print "RA validation failed. Unable to get an address." print "Teredo server we are dealing with has a weird behavior." self.status="off-line" return None # Here, we are behind cone NAT, i.e. fully qualified. self.status="qualified" self.oursrvaddr = server1 self.coned = 1 maddr, mport = mappedvals return self._constructTeredoAddress(server1, 1, maddr, mport) # Behind that point, we know we are no more considering Cone NAT. # Only symmetric or restricted NAT currentShot=1 while currentShot <= N: # Our "non-Cone NAT" Link-Layer address (the one used by Win XP). src6lladdr = "fe80::5445:5245:444f" r = self._sendRSAndReceiveRA(server1, src6lladdr, T) if r is None: # No RA received currentShot += 1 continue mappedvals1 = self._validateTeredoRA(r, server1, src6lladdr, verbose=verbose) if mappedvals1 is None: # Validation failed : our server sends weird packets print "RA validation failed. Unable to get an address." print "Teredo server we are dealing with has a weird behavior." self.status="off-line" return None break # From that point, we know we can talk using UDP but we don't # know the specific kind of NAT our GW implements : restricted or # symmetric. Let's test it with our second Teredo server. CurrentShot=1 while CurrentShot <= N: r = self._sendRSAndReceiveRA(server2, src6lladdr, T) if r is None: # No RA received CurrentShot += 1 continue mappedvals2 = self._validateTeredoRA(r, server1, src6lladdr, verbose=verbose) # Pour le test suivant, on devrait peut-etre rajouter une comparaison # avec currenShot histoire de laisser une autre chance au serveur if mappedvals2 is None: print "RA validation failed. Unable to get an address." print "Teredo server we are dealing with has a weird behavior." self.status="off-line" return None break # Second server returned an address. Let's compare the mapped address # and port to the one first server returned to determine the kind of # NAT we are dealing with. if mappedvals1 != mappedvals2: if verbose: print "You are behind a symmetric NAT : Teredo service is not usable." self.status="off-line" return None # Here is the "Happy End" : we are behind a restricted NAT. Teredo # service is usable maddr, mport = mappedvals1 self.status="qualified" self.oursrvaddr = server1 self.coned=0 return self._constructTeredoAddress(server1, 0, maddr, mport) # TODO : There's some understanding to gather from experiment against # bubble sending. Both reference documents (draft and Teredo Overview) # do not explicitly specify the kinds of addresses to use as source # and destination of bubbles. I've got a capture with bubbles that # incorporate link-local unicast and multicast IPv6 addresses. # In fact, sect 5.2.6 of the draft states that when sending indirect bubbles, # we must use "extract the Teredo IPv4 server address from the Teredo prefix # of the IPv6 address of the target # What's the logic behind the following function ? # The "establish" function is (with the qualification function) a big part of # Teredo implementation. Depending on the kind of peer we are dealing with # (mainly native or teredo) and also our position (behind Cone or Restriced # NAT), our behavior and the emission of the first packet to the peer # can hugely change. # The first selection is performed against peer kind of address. # - Teredo: # - if peer is behind cone NAT, I can send the packet directly to its mapped # address. Its NAT is open and mine is either open or will be for him # as a result of the sending. # - if peer is behind restricted NAT, I have to inform him to open a mapping # in its NAT GW for me. This will be done by sending an indirect bubble via # its teredo server. Before that, and only if I'm behind a restricted NAT, # I open the door by sending a direct bubble (which will be dropped) to its # mapped address and port. I know can finally send my data packet after # receiving a direct bubble from the peer. # # - Non-Teredo (Native, 6to4, ...) : # Here, it is getting harder. The teredo relay enters the game. The idea is to # get the address of the relay that is closer to our peer. It is done by # sending the peer an Echo Request (through our Teredo server) and mainly # monitor the IPv4 source of the Reply. # More specifically, after sending the Echo Request to the peer through # our Teredo server, if we are behind a cone NAT, the relay that will forward # the Reply will know it (the flag in our address) and first send an indirect # bubble in order to warn us to open the door. For this reason, if we are # stuck behind a Cone NAT, we wait for an indirect bubble and reply by # a bubble when it comes. We store the information regarding the relay. # # After that (or directly if we are behind Cone) we simply wait for the Echo # Reply that will provide us (in the Cone case) the address of the relay. # # We send the data packet to the learned address and port of the relay. # # TODO : The bubble and Echo Req/Rep stuff should only be performed one # time by peer. The last reception time (aka the status of our NAT GW mapping # for this specific peer) should be maintained in order to avoid this heavy # mechanism. def establish(self, dst6): if in6_isaddrTeredo(dst6): # Destination is Teredo dserver, dflag, dmappedaddr, dmappedport = teredoAddrExtractInfo(dst6) if dflag: # Teredo peer is behind a Cone NAT : simple case self.peers.add(dst6, (dmappedaddr, dmappedport)) return (dmappedaddr, dmappedport) else: # Teredo peer is behind Restricted NAT if not self.coned: # I'm behind Restricted NAT. Let's open the door with a direct bubble bubble = IPv6(src=self.ourv6addr, dst=dst6, nh=59) self.s.sendto(str(bubble), (dmappedaddr, dmappedport)) self.peers.add(dst6, (dmappedaddr, dmappedport)) # TODO : should I add after sending a bubble. In fact, # Peer is not available after this simple bubble ? # Send indirect bubble to our peer's server and record the time # of sending # TODO : see if the information on servers and relays should not # be stored somewhere else. bubble = IPv6(src=self.ourv6addr, dst=dst6, nh=59) self.s.sendto(str(bubble), (dserver, 3544)) self.peers.add(dserver, (dserver, 3544)) T=3 emitted=time.time() elapsed=0 # Here, we'll wait for the direct bubble while (T-elapsed) > 0: self.s.settimeout(float(T-elapsed)) try: # TODO : 200 ? string, address = self.s.recvfrom(200) except: break if string[0:2] == '`\x00': resp=IPv6(string) if resp.nh == 59: self.peers.add(dst6, (dmappedaddr, dmappedport)) return (dmappedaddr, dmappedport) elapsed = time.time() - emitted # Should make something more specific if we don't receive a bubble. # like "Host seems down, no response from host" return None else: # Native IPv6 or 6to4 peer # To establish the connection, we have to find its relay, and open # mapping in our NAT (if restricted). If we had a out of date mapping # in our peers list, we can give it a try without the Echo Reques / # Echo Reply procedure oldrelay = self.peers.get(dst6) if oldrelay is not None: # TODO : maintain a list of IPv6 local addresses used by relays # to be able to suppress this static information # get v6 address of relay # TODO : create a specific class to deal with relays relayv6addr=self.relays[oldrelay[0]] bubble=IPv6(src=self.ourv6addr, dst=relayv6addr, nh=59) self.s.sendto(str(bubble), oldrelay) return oldrelay # No Relay info for peer. Let's perform the real procedure # Let's emit an indirect Echo Request through our Teredo server echoreq=IPv6(src=self.ourv6addr, dst=dst6)/ICMPv6EchoRequest(data=RandString(2)) strechoreq = str(echoreq) echoreq=IPv6(strechoreq) self.s.sendto(strechoreq, (self.oursrvaddr, 3544)) if not self.coned: # Here, we are waiting either for a bubble from our peer # or directly for the Echo Reply (can happen) self.s.settimeout(3.0) while True: try: string, address = self.s.recvfrom(300) except: return None # ok, I got data. Where does it come from and what kind of # data is it ? # Ok, last chance : Is this our Teredo Indication if string[0:2] == '\x00\x00': try: resp = TeredoOrigInd(string) except: return None if address != (self.oursrvaddr, 3544): # indirect bubble must come from our server continue # TODO : check TeredoOrigind content with mapped versions if (isinstance(resp.payload, IPv6) and resp.payload.nh == 59): self.relays[resp.origip]=resp.payload.src break else: return None # - In restricted NAT case, we determine the adddress # of peer's relay from the Origin Indication relayaddr = resp.origip relayport = resp.oport # - Open the door to the relay. # TODO : verifier que la source de la bubble est bien une adresse # teredo et que les parametres mappes correspondent bien # a l'adresse v4 du relay, i.e. puis-je utiliser l'addresse # source de la bubble recue pour emettre ma bubble # # En fait, si on suit de pret les considerations de la section 5.2.6 # du draft et le contenu de la section "Initial communication from a # Teredo client to an IPv6-only host", le mieux serait de verifier # que les parametres mappes presents dans l'adresse source de la # bubble doivent correspondre a ceux present dans le header # Origin indication. Dans ce cas, la ligne suivante est valide bubble = IPv6(src=self.ourv6addr, dst=resp.payload.src, nh=59) self.s.sendto(str(bubble), (relayaddr, relayport)) T=3 emitted=time.time() elapsed=0 # Here, we'll wait for the Echo Reply from the relay. while (T-elapsed) > 0: self.s.settimeout(float(T-elapsed)) try: # TODO : 200 ? string, relay = self.s.recvfrom(200) except: return None if string[0:2] == '`\x00': resp=IPv6(string) if echoreq > resp: if self.coned: # - In Cone NAT case, we determine the address of # peer's relay from Source addr and port of Echo Reply relayaddr, relayport = relay break elapsed = time.time() - emitted if (T - elapsed) <= 0: return None self.peers.add(dst6, (relayaddr, relayport)) return (relayaddr, relayport) # The idea : If we already have a conversation with the destination, # pkt is sent directly. If not, establishment procedure is # performed. If we got a response, we can send the packet # def send(self, pkt): dst6=pkt.dst dst4 = self.peers.get(dst6) if dst4 is not None and self.peers.elapsed(dst6) < 30: # Recent dialog with the peer. IPv4 destination is known # Contact it directly and log this contact self.peers.add(dst6, dst4) self.s.sendto(str(pkt), dst4) else: # Connection must be established before emitting # TODO : establish could return v4 addr and port dst4=self.establish(dst6) if dst4 is not None: self.peers.add(dst6, dst4) ret= self.s.sendto(str(pkt), dst4) return ret return 0 # x is the MTU def recv(self, x): return None class TeredoPortField(ShortField): def addfield(self, pkt, s, val): return s+struct.pack("H", self.i2m(pkt,val) ^ 0xffff ) def getfield(self, pkt, s): return s[2:], self.m2i(pkt, struct.unpack("H",s[:2])[0] ^0xffff) class TeredoIPField(IPField): def addfield(self, pkt, s, val): return s+struct.pack(self.fmt, strxor(self.i2m(pkt,val), '\xff\xff\xff\xff')) def getfield(self, pkt, s): return s[self.sz:], self.m2i(pkt, strxor(struct.unpack(self.fmt, s[:self.sz])[0], '\xff\xff\xff\xff')) class _TeredoGuessPayload: def guess_payload_class(self,p): if len(p) > 2: if (struct.unpack("B",p[0])[0] & 0xf0) == 0x60: return IPv6 dispatcher = struct.unpack("!H",p[0:2])[0] if dispatcher == 0x0000: # origin indication return TeredoOrigInd elif dispatcher == 0x0001: # Authentication return TeredoAuth else: # What is it ? return TeredoUnknown return Raw class TeredoUnknown(_TeredoGuessPayload, Packet): name = "Teredo Unknown (not implemened in Scapy)" fields_desc = [ StrField("data","") ] class TeredoOrigInd(_TeredoGuessPayload, Packet): name = "Teredo Origin Indication" fields_desc = [ XShortField("fixed", 0x0000), TeredoPortField("oport",0x0000), TeredoIPField("origip","0.0.0.0") ] def answers(self,other): return self.payload.answers(other) # randomiser le nonce class TeredoAuth(_TeredoGuessPayload, Packet): name = "Teredo Authentication" fields_desc = [ ShortField("fixed", 0x0001), FieldLenField("idlen", None, "clientid", "B", shift=0), FieldLenField("aulen", None, "authid", "B", shift=0), StrLenField("clientid", "", "idlen"), StrLenField("authid", "", "aulen"), StrFixedLenField("nonce", '\x00'*8, 8), ByteField("confirm",0) ] def default_payload_class(self, p): if len(p) > 2: dispatcher = struct.unpack("!H",p[0:2])[0] if dispatcher == 0x0000: # origin indication return TeredoOrigInd else: # must be IPv6 return IPv6 def hashret(self): return self.nonce+self.payload.hashret() # La section 5.2 du draft fournit les informations suffisantes pour gerer un # "Etat Teredo" : # "The client will maintain the following variables that reflect the # state of the Teredo service: # # - Teredo Connectivity status, # - Mapped address and port number associated with the Teredo Service # port, # - Teredo IPv6 prefix associated with the Teredo service port, # - Teredo IPv6 address or addresses derived from the prefix, # - Link Local address, # - Date and time of the last interaction with the Teredo server, # - Teredo Refresh Interval, # - Randomized Refresh Interval, # - List of recent Teredo peers. ############################################################################### ### DHCP6 (RFC 3315) ############################################################################### All_DHCP_Relay_Agents_and_Servers = "ff02::1:2" # scope lien local All_DHCP_Servers = "ff05::1:3" # scope site-local : deprecated by 3879 ? dhcp6opts = { # from sect 24.3 1:"OPTION_CLIENTID", 2:"OPTION_SERVERID", 3:"OPTION_IA_NA", 4:"OPTION_IA_TA", 5:"OPTION_IAADDR", 6:"OPTION_ORO", 7:"OPTION_PREFERENCE", 8:"OPTION_ELAPSED_TIME", 9:"OPTION_RELAY_MSG", 11:"OPTION_AUTH", 12:"OPTION_UNICAST", 13:"OPTION_STATUS_CODE", 14:"OPTION_RAPID_COMMIT", 15:"OPTION_USER_CLASS", 16:"OPTION_VENDOR_CLASS", 17:"OPTION_VENDOR_OPTS", 18:"OPTION_INTERFACE_ID", 19:"OPTION_RECONF_MSG", 20:"OPTION_RECONF_ACCEPT" } # sect 5.3 RFC 3315 dhcp6types = { 1:"SOLICIT", 2:"ADVERTISE", 3:"REQUEST", 4:"CONFIRM", 5:"RENEW", 6:"REBIND", 7:"REPLY", 8:"RELEASE", 9:"DECLINE", 10:"RECONFIGURE", 11:"INFORMATION-REQUEST", 12:"RELAY-FORW", 13:"RELAY-REPL" } ### Debut des DUID duidtypes = { 1: "Link-layer address plus time", 2: "Vendor-assigned unique ID based on Enterprise Number", 3: "Link-layer Address" } # duid hardware types - Voir RFC826 duidhwtypes = { } # Un champ contenat une valeur de temps au format UTC, i.e. en nombre # de secondes depuis 1970. A implementer. class UTCTimeField(ByteField): name = "toto" # temporairement pour que ca ce lance # Implementer le champ LLAddrField() qui (comme le StrLenField) # se base sur la valeur du champ hwtype pour determiner la taille # et le type d'adresse concernee. Ce doit etre possible a la # creation mais egalement a la dissection. # # Par defaut (certainement le cas unique), on va se retrouver # avec une adresse MAC # # Developper le champ UTCTimeField() # # Cette valeur de DUID, une fois generee, devrait etre stockee dans # la conf et eventuellement dans la session a la fin pour faire en # sort de conserver la valeur par la suite. L'ideal serait de # stocker ce DUID par type d'interface. Comme precise en section 9.2 # il est conseille de conserver le meme DUID pour toutes les # interfaces de meme type d'un systeme donne. # # Un autre point est a prendre en compte : les clients et serveurs # ne possedant pas de stockage a long terme (c'est un peu le cas # avec Scapy) ne doivent pas utiliser ce mecanisme. # # Conclusion : le plus simple est certainement de mettre par defaut # une valeur fixe pour le temps et d'avoir un DUID different # mais stable pour chacune des interfaces du systeme. Ca semble simple # et efficace. A voir. # # Voir section 9.2 # # TODO : Terminer cette classe class DHCP6_DUID_LLT(Packet): # sect 9.2 RFC 3315 name = "DUID - Link-layer address plus time" fields_desc = [ ShortEnumField("type",1,duidtypes), XShortField("hwtype",0), # revoir la valeur par defaut UTCTimeField("time",0),# a revoir #LLAddrField("lladdr") ] # bon, en fait, les enterprise-numbers, y'en a plus de 23000, le # fichier qui les contient est dispo sur le site de l'IANA mais il # fait 1,8 Mo # www.iana.org/asignments/enterprise-numbers # Le mieux, ce serait de prendre les principaux vendeurs de matos # reseau et les trucs connus. # Le principal probleme de cette merde, c'est la taille sur laquelle # est codee l'identifiant. Rien n'est precise. class DHCP6_DUID_EN(Packet): # sect 9.3 RFC 3315 name = "DUID - Assigned by Vendor Based on Enterprise Number" fields_desc = [ ShortEnumField("type",2,duidtypes), IntField("enterprise-number",0), # a revoir StrField("id","") ] # a revoir # A modifier apres avoir implemente LLAddrField # et avoir gerer les hwtype class DHCP6_DUID_LL(Packet): # sect 9.4 RFC 3315 name = "DUID - Based on Link-layer Address" fields_desc = [ ShortEnumField("type",3,duidtypes), XShortField("hwtype",0), # a revoir # LLAddrField("lladdr") # a revoir ] ### Fin des DUID - Debut des classes definissant le protocole en lui ### meme # A generic DHCPv6 Option class DHCP6OptUnknown(Packet): name = "Unknown DHCPv6 OPtion" fields_desc = [ ShortField("optcode", 1 ), FieldLenField("len", None, "data", "!H", 0 ), StrLenField("data","", "len")] # Il serait pas mal de mettre une valeur par defaut plus interessant # au niveau du champ lladdr. Voir egalement si ce ne serait pas plus # interessant de posseder un type de champ particulier class DHCP6OptClientIdWithDUID_LLT(Packet): # RFC sect 22.2 name = "DHCP6 Client Identifier Option With DUID LLT" fields_desc = [ ShortField("optcode", 1 ), FieldLenField("optlen", None, "duid", "!H", -12), ShortEnumField("type",1,duidtypes), XShortField("hwtype",""), UTCTimeField("time",0), # a revoir StrLenField("lladdr","","optlen") ] class DHCP6OptClientIdWithDUID_EN(Packet): # RFC sect 22.2 name = "DHCP6 Client Identifier Option With DUID EN" fields_desc = [ ShortField("optcode", 1 ), FieldLenField("optlen", None, "duid", "!H", -10), ShortEnumField("type",2,duidtypes), IntField("enterprise-number",0), # ??????????? StrLenField("id","","optlen") ] class DHCP6OptClientIdWithDUID_LL(Packet): # RFC sect 22.2 name = "DHCP6 Client Identifier Option With DUID LL" fields_desc = [ ShortField("optcode", 1 ), FieldLenField("optlen", None, "duid", "!H", -10), ShortEnumField("type",3,duidtypes), XShortField("hwtype",0), # ???????????? # LLAddrField("lladdr","","optlen") # ?????? ] # L'option ServerId est identique a l'option Client ID (les 3 # versions), seul l'optcode passe a la valeur 2 au lieu de 1 class DHCP6OptServerIdWithDUID_LLT(Packet): # RFC sect 22.3 name = "DHCP6 Server Identifier Option With DUID LLT" fields_desc = [ ShortField("optcode", 2 ), FieldLenField("optlen", None, "duid", "!H", -12), ShortEnumField("type",1,duidtypes), XShortField("hwtype",""), UTCTimeField("time", 0), # ????? StrLenField("lladdr","","optlen") ] class DHCP6OptServerIdWithDUID_EN(Packet): # RFC sect 22.2 name = "DHCP6 Server Identifier Option With DUID EN" fields_desc = [ ShortField("optcode", 2 ), FieldLenField("optlen", None, "duid", "!H", -10), ShortEnumField("type",2,duidtypes), IntField("enterprise-number",0), # ??????????? StrLenField("id","","optlen") ] class DHCP6OptServerIdWithDUID_LL(Packet): # RFC sect 22.2 name = "DHCP6 Server Identifier Option With DUID LL" fields_desc = [ ShortField("optcode", 2 ), FieldLenField("optlen", None, "duid", "!H", -10), ShortEnumField("type",3,duidtypes), XShortField("hwtype",0), # ????????? # LLAddrField("lladdr","","optlen") # ?????? ] # Le volume de code duplique sur les 6 classes precedentes est assez # impressionnant, ce serait pas mal de se simplifier la vie. # Faire qqch (genre plan epervier, ou truc plus violent) # Quelles sont les valeurs par defaut a utiliser pour t1 et t2 # Pour un client, les valeurs t1 et t2 a 0 indique que le client n'a # pas de preference et laisse le serveur specifier ces valeurs. # pour le cas d'un serveur, il seriat mieux que ceux-ci soit specifies # correctement # Que se passe-t-il si t1 et t2 sont mis a 0 pour les valeurs emises # par le serveur au client. Prend-il des valeurs par defaut dans ce # cas ? # Une valeur de 0xffffffff (inifinie)) semble une bonne solution # Il pourrait etre sympa de faire en sorte de definir dans la conf des # valeurs de t1 et t2 pour le client et pour le serveur et de les # utiliser en mode suivi de session. # Ce qui risque d'etre chaud, c'est de faire en sorte de gerer un etat # des clients en mode server. En fait ca risque de ne pas etre # faisable. # # Revoir le coup du champ d'options gere avec une chaine. # a la limite, l'utiliser pour extraire du padding. class DHCP6OptIA_NA(Packet): # RFC sect 22.4 name = "DHCP6 Identity Association for Non-temporary Addresses Option" fields_desc = [ ShortField("optcode", 3 ), FieldLenField("optlen", None, "ianaopts", "!H", -12 ), XIntField("iaid", None), IntField("T1", None), IntField("T2", None), StrLenField("ianaopts","", "optnlen")] def guess_payload_class(self,payload): if payload: return DHCP6OptIAAddress else: return Raw # transformer le dernier champ en champ d'option et non en stupide chaine # les options contenues dedans ne sont pas comprehensible dans ce cas. # c'est un peu triste (un peu) class DHCP6OptIA_TA(Packet): # RFC sect 22.5 name = "DHCP6 Identity Association for Temporary Addresses Option" fields_desc = [ ShortField("optcode", 4 ), FieldLenField("optlen", None, "iataopts", "!H", -4 ), XIntField("iaid", None), StrLenField("iataopts","", "optlen")] def guess_payload_class(self,payload): if payload: return DHCP6OptIAAddress else: return Raw # Cette option est encapsulee dans le champ d'option des options IA_NA # ou IA_TA. Elle ne peut apparaitre qu'a cet endroit class DHCP6OptIAAddress(Packet): # RFC sect 22.6 name = "DHCP6 IA Address Option (IA_TA or IA_NA suboption)" fields_desc = [ ShortField("optcode", 5 ), FieldLenField("optlen", None, "iaaddropts", "!H", -24 ), # IP6AddrField("address",None), IntField("", None), XIntField("iaid", None), StrLenField("iaaddropts","", "len")] # revoir ce que Guillaume a utilise pour contenir les addresses v6 # et l'utiliser pour le champ address # Creer une structure Time Field ? # Meme remarque que precedemment concernant les options (i.e le denier champ) # reactiver le champ IP6Addrfield lorsque les changements de Guillaume # ont ete merges # A client may include an ORO in a solicit, Request, Renew, Rebind, # Confirm or Information-request class DHCP6OptOptReq(Packet): # RFC sect 22.7 name = "DHCP6 Option Request Option" fields_desc = [ ShortField("optcode", 6 ), FieldLenField("optlen", None, "resopts", "!H", 24 ), StrLenField("reqopts","", "len")] # la, il faudrait regarder plus en details ce qui concerne # le champ optlen. La section 22.7 indique que la valeur correspond # a 2 fois le nombre d'options (chacune etant sur 16 bits). IL faudrait # peut-etre en tenir compte. # Il serait bon d'avoir une classe RequestedOptionsList qui permettrait # de creer facilement une liste d'option a ajouter et qui gererait la # commpletion automatique par tabulation. c'est pas gagne # a revoir lors d'une seconde passe # emise par un serveur pour affecter le choix fait par le client. Dans # les messages Advertise, a priori class DHCP6OptPref(Packet): # RFC sect 22.8 name = "DHCP6 Preference Option" fields_desc = [ ShortField("optcode", 7 ), ShortField("optlen", 1 ), ByteField("prefval",255) ] # Voir la valeur par defaut a donner a prefval (voir sect 17.1.3) # A client must include an Elapsed Time Option in messages to indicate # how long the client has been trying to complete a DHCP message # exchange. The elapsed time is measured from the time at which the # client sent the first message in the message exchange, and the # elapsed-time field is set to 0 in the first message in the message # exchange. see page 80. class DHCP6OptElapsedTime(Packet):# RFC sect 22.9 name = "DHCP6 Elapsed Time Option" fields_desc = [ ShortField("optcode", 8 ), ShortField("optlen", 2 ), ShortField("elapsedtime",None) ] # Elapsed time est exprime en 100eme de seconde. Voir quoi faire pour cela # creation d'une classe particuliere comme suggere precedemment # genre ByteTimeField, ShortTimeField heritant toutes deux d'un TimeField # avec une possibilite de definir l'unite representee. # il serait ptet plus malin de voir le paquet encapsule comme un # payload plutot que comme un PacketField class DHCP6OptRelayMsg(Packet):# RFC sect 22.10 name = "DHCP6 Relay Message Option" fields_desc = [ ShortField("optcode", 9 ), ShortField("optlen", None ), IPv6PacketField("relaymsg",None)] # remplacer le None # impossible d'associer une methode d'extraction de padding puisque # le paquet relaye est pris en charge charge directement par le # champ de type PacketField. Au final, que se passe-t-il si le paquet # relaye est coupe. Il serait peut-etre plus malin de marquer celui-ci # comme etant un paylod et faire en sorte d'extraire l'eventuel padding # non ? # pour prendre un cas concret, que se passe-t-il si une option relay message # est suivie par une autre option (possible ?). Dans ce cas, # l'option qui suit a des chances d'etre interpreteee comme option du paquet # relaye. Non? # ----> reprendre page 81 pour cette merde d'option d'authent qui ne sera # de toute maniere jamais utilisee par personne. Sa mere #class DHCP6OptAuth(Packet): # RFC sect 22.11 # name = "DHCP6 Authentication Option" # fields_desc = [ ShortField("optcode", 11 ), # Shortield("optlen", None ), # PacketField("relaymsg",Raw(),DHCP6) ] # p82/101 class DHCP6OptServerUnicast(Packet):# RFC sect 22.12 name = "DHCP6 Server Unicast Option" fields_desc = [ ShortField("optcode", 12 ), ShortField("optlen", 16 ), IP6Field("srvaddr",None) ] # a revoir au moment de l'integration du code de guillaume pour # ce qui concerne l'IP6Field. dhcp6statuscodes = { 0:"Success", # sect 24.4 1:"UnspecFail", 2:"NoAddrsAvail", 3:"NoBinding", 4:"NotOnLink", 5:"UseMulticast" } # A Status Code option may appear in the options field of a DHCP # message and/or in the options field of another option. If the status # code option does not appear in a message in which the option could # appear, the status of the message is assumed to be Success. see sect # 22.13 class DHCP6OptSatusCode(Packet):# RFC sect 22.13 name = "DHCP6 Status Code Option" fields_desc = [ ShortField("optcode", 13 ), FieldLenField("optlen", None, "statusmsg", "!H", shift=-2 ), ShortEnumField("statuscode",None,dhcp6statuscodes), StrLenField("statusmsg", "", "optlen") ] # this option is included by the client in the Solicit message in # order to get a reply message with commited assignments. class DHCP6OptRapidCommit(Packet): # RFC sect 22.14 name = "DHCP6 Rapid Commit Option" fields_desc = [ ShortField("optcode", 14 ), ShortField("optlen", 0)] # une option qui ne sert a rien. class DHCP6OptUserClass(Packet):# RFC sect 22.15 name = "DHCP6 User Class Option" fields_desc = [ ShortField("optcode", 15 ), FieldLenField("optlen", None, "userclassdata", "!H", 0 ), StrLenField("userclassdata", "", "optlen") ] # cf page 85 pour un sous-decoupage du champ userclassdata en # |len+opaquedata|len+opaquedata|.... Le truc qui fait bien chier et # qu'a l'air de servir strictement a rien. # encore une option qui va servir a rien. Elle va etre utilisee par # Cisco et microsoft et c'est tout. class DHCP6OptVendorClass(Packet):# RFC sect 22.16 name = "DHCP6 Vendor Class Option" fields_desc = [ ShortField("optcode", 16 ), FieldLenField("optlen", None, "vcdata", "!H", shift=-4 ), XIntField("enterprisenum",None), StrLenField("vcdata", "", "optlen") ] # cf page 86 pour un sous-decoupage du champ vcdata en # |len+opaquedata|len+opaquedata|.... # Voila le troisieme qui va servir a rien. class DHCP6OptVendorSpecificInfo(Packet):# RFC sect 22.17 name = "DHCP6 Vendor-specific Information Option" fields_desc = [ ShortField("optcode", 17 ), FieldLenField("optlen", None, "optdata", "!H", shift=-4 ), XIntField("enterprisenum",None), StrLenField("optdata", "", "optlen") ] # cf page 87 pour un sous-decoupage du champ optdata en # |opt-code+len+opaquedata|opt-code+len+opaquedata|.... # il serait pas mal d'aller chercher le nom du fabricant # comme le fait ethereal # Repasser sur cette option a la fin. Elle a pas l'air d'etre des # masses critique. class DHCP6OptIfaceId(Packet):# RFC sect 22.18 name = "DHCP6 Interface-Id Option" fields_desc = [ ShortField("optcode", 18 ), FieldLenField("optlen", None, "ifaceid", "!H", 0 ), StrLenField("ifaceid", "", "optlen") ] # A server includes a Reconfigure Message option in a Reconfigure # message to indicate to the client whether the client responds with a # renew message or an Informatiion-request message. class DHCP6OptReconfMsg(Packet): # RFC sect 22.19 name = "DHCP6 Reconfigure Message Option" fields_desc = [ ShortField("optcode", 19 ), ShortField("optlen", 1 ), ByteEnumField("msgtype",None, {5:"Renew Message", 11:"Information Request"})] # A client uses the Reconfigure Accept option to announce to the # server whether the client is willing to accept Recoonfigure # messages, and a server uses this option to tell the client whether # or not to accept Reconfigure messages. The default behavior in the # absence of this option, means unwillingness to accept reconfigure # messages, or instruction not to accept Reconfigure messages, for the # client and server messages, respectively. class DHCP6OptReconfAccept(Packet): # RFC sect 22.20 name = "DHCP6 Reconfigure Accept Option" fields_desc = [ ShortField("optcode", 20 ), ShortField("optlen", 0)] # Quelques variables d'etat du protocole qu'il serait sympa d'avoir # dans la conf pour eviter de les taper toutes les 3 secondes DHCP6RelayAgentUnicastAddr="" DHCP6RelayHopCount="" DHCP6ServerUnicastAddr="" DHCP6ClientUnicastAddr="" DHCP6ClientIA_TA="" DHCP6ClientIA_NA="" DHCP6ClientIAID="" T1="" # Voir 2462 T2="" # Voir 2462 DHCP6ServerDUID="" DHCP6CurrentTransactionID="" # devrait etre utilise pour matcher une # reponse et mis a jour en mode client par une valeur aleatoire pour # laquelle on attend un retour de la part d'un serveur. DHCP6PrefVal="" # la valeur de preference a utiliser dans # les options preference # inherited by classes that define types of message sent by clients # (used to set IPv6 src and dst addresses). # il serait peut-etre plus simple de faire une seule classe avec un # post_build dont heriteraient les autres classes avec un test sur le # type courant pour decider quelles adresses IP utiliser en source et # destination. Meme chose pour les ports UDP source et destination. # # Emis par : # - server : ADVERTISE, REPLY, RECONFIGURE, RELAY-REPL (vers relay) # - client : SOLICIT, REQUEST, CONFIRM, RENEW, REBIND, RELEASE, DECLINE, # INFORMATION REQUEST # - relay : RELAY-FORW (vers server) class _DHCP6FromClient(Packet): def post_build(): return # same for Server emitted messages class _DHCP6FromServer(Packet): def post_build(): return # same for Relay emitted messages class _DHCP6FromRelay(Packet): def post_build(): return class _DHCP6GuessPayload(Packet): def guess_payload_class(self, payload): if len(payload) > 1 : return globals().get(dhcp6opts.get(struct.unpack("!H", payload[0:2])[0],DHCP6OptUnknown), Raw) return Raw ## DHCPv6 messages sent between Clients and Servers (types 1 to 11) # Comme specifie en section 15.1 de la RFC 3315, les valeurs de # transaction id sont selectionnees de maniere aleatoire par le client # a chaque emission et doivent matcher dans les reponses faites par # les clients class DHCP6(_DHCP6GuessPayload,Packet): name = "DHCPv6 Generic Message)" fields_desc = [ ShortEnumField("msg-type",None,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Solicit(_DHCP6GuessPayload,Packet): name = "DHCPv6 Solicit Message" fields_desc = [ ShortEnumField("msg-type",1,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Advertise(_DHCP6GuessPayload,Packet): name = "DHCPv6 Advertise Message" fields_desc = [ ShortEnumField("msg-type",2,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Request(_DHCP6GuessPayload,Packet): name = "DHCPv6 Request Message" fields_desc = [ ShortEnumField("msg-type",3,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Confirm(_DHCP6GuessPayload,Packet): name = "DHCPv6 Confirm Message" fields_desc = [ ShortEnumField("msg-type",4,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Renew(_DHCP6GuessPayload,Packet): name = "DHCPv6 Renew Message" fields_desc = [ ShortEnumField("msg-type",5,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Rebind(_DHCP6GuessPayload,Packet): name = "DHCPv6 Rebing Message" fields_desc = [ ShortEnumField("msg-type",6,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Reply(_DHCP6GuessPayload,Packet): name = "DHCPv6 Reply Message" fields_desc = [ ShortEnumField("msg-type",7,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Release(_DHCP6GuessPayload,Packet): name = "DHCPv6 Release Message" fields_desc = [ ShortEnumField("msg-type",8,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Decline(_DHCP6GuessPayload,Packet): name = "DHCPv6 Decline Message" fields_desc = [ ShortEnumField("msg-type",9,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_Reconf(_DHCP6GuessPayload,Packet): name = "DHCPv6 Reconfigure Message" fields_desc = [ ShortEnumField("msg-type",10,dhcp6types), X3BytesField("trid",0x000000) ] class DHCP6_InfoRequest(_DHCP6GuessPayload,Packet): name = "DHCPv6 Information Request Message" fields_desc = [ ShortEnumField("msg-type",11,dhcp6types), X3BytesField("trid",0x000000) ] ## DHCPv6 messages sent between Relay Agents and Servers (types 12 and 13) # Normalement, doit inclure une option "Relay Message Option" # peut en inclure d'autres. # voir section 7.1 de la 3315 class DHCP6_RelayForward(_DHCP6GuessPayload,Packet): name = "DHCPv6 Relay Forward Message (Relay Agent/Server Message)" fields_desc = [ ShortEnumField("msg-type",12,dhcp6types), ShortField("count",None), IP6Field("linkaddr",None), IP6Field("peeraddr",None) ] def hashret(self): # we filter on peer address field return socket.inet_pton(socket.AF_INET6, self.peeraddr) # Normalement, doit inclure une option "Relay Message Option" # peut en inclure d'autres. # Les valeurs des champs hop-count, link-addr et peer-addr # sont copiees du messsage Forward associe. POur le suivi de session. # Pour le moment, comme decrit dans le commentaire, le hashret # se limite au contenu du champ peer address. # Voir section 7.2 de la 3315. class DHCP6_RelayReply(DHCP6_RelayForward): name = "DHCPv6 Relay Reply Message (Relay Agent/Server Message)" fields_desc = [ ShortEnumField("msg-type",13,dhcp6types), ShortField("count",None), IP6Field("linkaddr",None), IP6Field("peeraddr",None) ] def hashret(self): # We filter on peer address field. return socket.inet_pton(socket.AF_INET6, self.peeraddr) def answers(self, other): return (isinstance(other, DHCP6_RelayForward) and self.count == other.count and self.linkaddr == other.linkaddr and self.peeraddr == other.peeraddr ) # Solicit Message : sect 17.1.1 RFC3315 # - sent by client # - must include a client identifier option # - the client may include IA options for any IAs to which it wants the # server to assign address # - The client use IA_NA options to request the assignment of # non-temporary addresses and uses IA_TA options to request the # assignment of temporary addresses # - The client should include an Option Request option to indicate the # options the client is interested in receiving (eventually # including hints) # - The client includes a Reconfigure Accept option if is willing to # accept Reconfigure messages from the server. # Le cas du send and reply est assez particulier car suivant la # presence d'une option rapid commit dans le solicit, l'attente # s'arrete au premier message de reponse recu ou alors apres un # timeout. De la meme maniere, si un message Advertise arrive avec une # valeur de preference de 255, il arrete l'attente et envoie une # Request. # - The client announces its intention to use DHCP authentication by # including an Authentication option in its solicit message. The # server selects a key for the client based on the client's DUID. The # client and seerver use that key to authenticate all DHCP messages # exchanged during the session # Advertise Message # - sent by server # - Includes a server identifier option # - Includes a client identifier option # - the client identifier option must match the client's DUID # - transaction ID must match # Request Message # - sent by clients # - includes a server identifier option # - the content of Server Identifier option must match server's DUID # - includes a client identifier option # - must include an ORO Option (even with hints) p40 # - can includes a reconfigure Accept option indicating whether or # not the client is willing to accept Reconfigure messages from # the server (p40) # - When the server receives a Request message via unicast from a # client to which the server has not sent a unicast option, the server # discards the Request message and responds with a Reply message # containinig Status Code option with the value UseMulticast, a Server # Identifier Option containing the server's DUID, the client # Identifier option from the client message and no other option. # Confirm Message # - sent by clients # - must include a clien identifier option # - When the server receives a Confirm Message, the server determines # whether the addresses in the Confirm message are appropriate for the # link to which the client is attached. cf p50 # Renew Message # - sent by clients # - must include a server identifier option # - content of server identifier option must match the server's identifier # - must include a client identifier option # - the clients includes any IA assigned to the interface that may # have moved to a new link, along with the addresses associated with # those IAs in its confirm messages # - When the server receives a Renew message that contains an IA # option from a client, it locates the client's binding and verifies # that the information in the IA from the client matches the # information for that client. If the server cannot find a client # entry for the IA the server returns the IA containing no addresses # with a status code option est to NoBinding in the Reply message. cf # p51 pour le reste. # Rebind Message # - sent by clients # - must include a client identifier option # cf p52 # Decline Message # - sent by clients # - must include a client identifier option # - Server identifier option must match server identifier # - The addresses to be declined must be included in the IAs. Any # addresses for the IAs the client wishes to continue to use should # not be in added to the IAs. # - cf p54 # Release Message # - sent by clients # - must include a server identifier option # cf p53 # Reply Message # - sent by servers # - the message must include a server identifier option # - transaction-id field must match the value of original message # The server includes a Rapid Commit option in the Reply message to # indicate that the reply is in response to a solicit message # - if the client receives a reply message with a Status code option # with the value UseMulticast, the client records the receipt of the # message and sends subsequent messages to the server through the # interface on which the message was received using multicast. The # client resends the original message using multicast # - When the client receives a NotOnLink status from the server in # response to a Confirm message, the client performs DHCP server # solicitation as described in section 17 and client-initiated # configuration as descrribed in section 18 (RFC 3315) # - when the client receives a NotOnLink status from the server in # response to a Request, the client can either re-issue the Request # without specifying any addresses or restart the DHCP server # discovery process. # - the server must include a server identifier option containing the # server's DUID in the Reply message # Reconfigure Message # - sent by servers # - must be unicast to the client # - must include a server identifier option # - must include a client identifier option that contains the client DUID # - must contain a Reconfigure Message Option and the message type # must be a valid value # - the server sets the transaction-id to 0 # - The server must use DHCP Authentication in the Reconfigure # message. Autant dire que ca va pas etre le type de message qu'on va # voir le plus souvent. # Information-Request Message # - sent by clients when needs configuration information but no # addresses. # - client should include a client identifier option to identify # itself. If it doesn't the server is not able to return client # specific options or the server can choose to not respond to the # message at all. The client must include a client identifier option # if the message will be authenticated. # - client must include an ORO of option she's interested in receiving # (can include hints) # Relay-Forward Message # - sent by relay agents to servers # If the relay agent relays messages to the All_DHCP_Servers multicast # address or other multicast addresses, it sets the Hop Limit field to # 32. # Relay-Reply Message # - sent by servers to relay agents # - if the solicit message was received in a Relay-Forward message, # the server constructs a relay-reply message with the Advertise # message in the payload of a relay-message. cf page 37/101. Envoie de # ce message en unicast au relay-agent. utilisation de l'adresse ip # presente en ip source du paquet recu ### fin DHCP6 ############################################################################### # MIPv6 and NEMO (RFC 3775 and 3963) ############################################################################### class ICMPv6HAADRequest(_ICMPv6, Packet): name = 'ICMPv6 Home Agent Address Discovery Request' fields_desc = [ ByteEnumField("type", 144, icmp6types), ByteField("code", 0), XShortField("cksum", None), XShortField("id", 0), BitEnumField("R", 1, 1, {1: 'MR' }), XBitField("reserved",0,15), ] def hashret(self): return struct.pack("H",self.id)+self.payload.hashret() # Must use metaclasses ... class ICMPv6HAADReply(_ICMPv6, Packet): name = 'ICMPv6 Home Agent Address Discovery Reply' fields_desc = [ ByteEnumField("type", 145, icmp6types), ByteField("code", 0), XShortField("cksum", None), XShortField("id", 0), BitEnumField("R", 1, 1, { 1: 'MR' }), XBitField("reserved",0,15), IP6ListField('addresses', None) ] def hashret(self): return struct.pack("H",self.id)+self.payload.hashret() def answers(self, other): if not isinstance(other, ICMPv6HAADRequest): return 0 return self.id == other.id class _MobilityHeader(_IPv6OptionHeader): # _IPv6OptionHeader ?!? name = 'Dummy IPv6 Mobility Header' def post_build(self, p): if self.cksum == None: if self.underlayer != None: cksum = in6_chksum(135, self.underlayer, p) else: # May prevent to send a null checksum; actually needed for pad() cksum = 0 p = p[:4]+struct.pack("!H", cksum)+p[6:] return p class _MOGP: # RFC3775: 6.1.1 # # These options include padding # options that can be used to ensure that other options are aligned # properly, and that the total length of the message is divisible by 8. # The encoding and format of defined options are described in Section # 6.2. # # Alignment requirements for the Mobility Header are the same as for # any IPv6 protocol Header. That is, they MUST be aligned on an 8- # octet boundary. # # NOTE: alignement is calculated from the begining of the BU/BACK def default_payload_class(self,p): if len(p) > 1: otype = ord(p[0]) return globals().get(mobopts.get(otype,"Raw"), "Raw") else: return Raw mhtypes = { 0: 'BRR', 1: 'HoTI', 2: 'CoTI', 3: 'HoT', 4: 'CoT', 5: 'BU', 6: 'BACK', 7: 'BE' } class MobilityOptions(StrField): islist=1 def getfield(self, pkt, s): return self._do(pkt, s) def m2i(self, pkt, x): return self._do(pkt,x)[1] def _do(self, pkt, x): # TODO: change this method print "_do x=%s len=%d" % (x, len(x)) opt = [] while x: t = ord(x[0]) print t cls = globals().get(mobopts.get(t,"Raw"), "Raw") print cls if not cls == Raw: f = PacketField('', None, cls) left, val = f.getfield(pkt, x) opt.append((val.name, val)) x = str(val.payload) del(val.payload) continue else: break return x,opt class IPv6MobilityHeader_BU(_MOGP, _MobilityHeader): def __init__(self, _pkt="", _internal=0, autopad=1, **fields): self.autopad = autopad Packet.__init__(self, _pkt, _internal, **fields) name = "IPv6 Mobility Header - Binding Update" fields_desc = [ ByteEnumField("nh", 59, ipv6nh), BitField("len", 3, 8), # unit ? ByteEnumField("mhtype", 5, mhtypes), ByteField("reserved", 0), XShortField("cksum", None), XShortField("seq", 0x4242), # => ShortNonceField FlagsField("flags", 49, 6, "AHLKMR"), XBitField("reserved", 0, 10), XShortField("time", 3) # unit == 4 seconds #MobilityOptions("mopts", []) # with autopad?!? ] overload_fields = { IPv6: { "nh": 135 }} def hashret(self): return str(self.seq) def answers(self, other): return 0 backstatus = { 0: 'Binding Update accepted', 1: 'Accepted but prefix discovery necessary', 128: 'Reason unspecified', 129: 'Administratively prohibited', 130: 'Insufficient resources', 131: 'Home registration not supported', 132: 'Not home subnet', 133: 'Not home agent for this mobile node', 134: 'Duplicate Address Detection failed', 135: 'Sequence number out of window', 136: 'Expired home nonce index', 137: 'Expired care-of nonce index', 138: 'Expired nonces', 139: 'Registration type change disallowed', 140: 'Mobile Router Operation not permitted', 141: 'Invalid Prefix', 142: 'Not Authorized for Prefix', 143: 'Forwarding Setup failed (prefixes missing)' } class IPv6MobilityHeader_BACK(_MOGP, _MobilityHeader): name = "IPv6 Mobility Header - Binding ACK" fields_desc = [ ByteEnumField("nh", 59, ipv6nh), BitField("len", 1, 8), # unit ? ByteEnumField("mhtype", 6, mhtypes), ByteField("reserved", 0), XShortField("cksum", None), ByteEnumField("status", 0, backstatus), FlagsField("flags", 2, 2, "KR"), XBitField("reserved", 0, 6), XShortField("seq", 0), # => ShortNonceField XShortField("time", 0) ] overload_fields = { IPv6: { "nh": 135 }} def hashret(self): return str(self.seq) def answers(self, other): if isinstance(other, IPv6MobilityHeader_BU) and other.mhtype == 5 and self.mhtype == 6: return 1 class IPv6MobilityHeader_BE(_MOGP, _MobilityHeader): name = "IPv6 Mobility Header - Binding Error" fields_desc = [ ByteEnumField("nh", 59, ipv6nh), BitField("len", 1, 8), # unit ? ByteEnumField("mhtype", 7, mhtypes), ByteField("reserved", 0), XShortField("cksum", None), ByteEnumField("status", 0, backstatus), ByteField("reserved", 0), IP6Field("ha", "::")#, #MobilityOptions("mopts", []) # with autopad?!? ] overload_fields = { IPv6: { "nh": 135 }} def hashret(self): return 'coincoin' def answers(self, other): if isinstance(other, ICMPv6MPSol): # Won't work that easily return 1 class Pad1PKT(_MOGP, Packet): # TODO: ugly ... name = 'Pad packet' fields_desc = [ Pad1('pad') ] class PadNPKT(_MOGP, Packet): # TODO: ugly ... name = 'PadN packet' fields_desc = [ PadN('pad', 0) ] class BindingRefreshAdvice(_MOGP, Packet): name = 'MIPv6 - Binding Refresh Advice' fields_desc = [ ByteField('type', 2), ByteField('len', 2), ShortField('rint', 0) ] class AlternateCareofAddress(_MOGP, Packet): name = 'MIPv6 - Alternate Care-of Address' fields_desc = [ ByteField('type', 3), ByteField('len', 16), IP6Field("acoa", "::") ] class NonceIndices(_MOGP, Packet): name = 'MIPv6 - Nonce Indices' fields_desc = [ ByteField('type', 3), ByteField('len', 16), ShortField('hni', 0), ShortField('coni', 0) ] class MobileNetworkPrefixOption(_MOGP, Packet): name = 'NEMO - Mobile Network Prefix Option' fields_desc = [ ByteField("type", 6), ByteField("len", 16), ByteField("Reserved", 0), ByteField("plen", 64), IP6Field("prefix", "::") ] mobopts = { 0: "Pad1PKT", 1: "PadNPKT", 2: "BindingRefreshAdvice", 3: "AlternateCareofAddress", 4: "NonceIndices", #5: BindingAuthorizationData # TO IMPLEMENT 6: "MobileNetworkPrefixOption" } moboptsal = { 2: lambda n: 3*n, 3: lambda n: 8*n+6, 4: lambda n: 2*n, #5:, lambda n: 8*n+2 6: lambda n: 8*n+4 } def find_pad(l, t): i = 1 pad = moboptsal[t](i) - l while pad <= 0: print pad pad = moboptsal[t](i) - l i *= 2 return pad-2 # Error if required padding < 1 # Is padding the end of the packet required ? def pad(lst): bu = lst[0] l = len(str(bu)) for i in lst[1:]: bu /= PadNPKT(pad=find_pad(l, i.type)) bu /= i l = len(str(bu)) return bu # RFC 2460, Destination Options Header # len: 8-octet units; not including the first 8 octets class IPv6OptionHeaderHomeAddress(_IPv6OptionHeader): name = "IPv6 Option Header Home Address" fields_desc = [ ByteEnumField("nh", 59, ipv6nh), ByteField("len", 2), PadN('pad', 2), # To achieve 8n alignement ByteField("opttype", 201), ByteField("optlen", 16), IP6Field("ha", '::') ] overload_fields = { IPv6: { "nh": 60 }} class ICMPv6MPSol(_ICMPv6, Packet): name = 'ICMPv6 Mobile Prefix Solicitation' fields_desc = [ ByteEnumField("type", 146, icmp6types), ByteField("code", 0), XShortField("cksum", None), XShortField("id", 0x4142), XShortField("reserved", 0), ] def _hashret(self): return struct.pack("H",self.id) class ICMPv6MPAdv(_ICMPv6, Packet): name = 'ICMPv6 Mobile Prefix Advertissement' fields_desc = [ ByteEnumField("type", 147, icmp6types), ByteField("code", 0), XShortField("cksum", None), XShortField("id", 0), BitEnumField("flags", 2, 2, {2: 'M', 1:'O'}), XBitField("reserved", 0, 6) ] def hashret(self): return struct.pack("H",self.id) # Revoir de quoi elle herite. Modifie pour demarre le truc # En fait tous les _ICMPv6 etait des ICMPv6. voir quoi faire class ICMPerror6(_ICMPv6): name = "ICMPv6 in ICMPv6" def answers(self, other): if not isinstance(other,_ICMPv6): return 0 if not ((self.type == other.type) and (self.code == other.code)): return 0 if self.code in [3,0,128,129,133,134,135,136,139,140,141,142]: if (self.id == other.id and self.seq == other.seq): return 1 else: return 0 else: return 1 def mysummary(self): return Packet.mysummary(self) try: import sw6 SW6 = 1 except ImportError: log_loading.warning("did not find sw6 module. See http://namabiiru.hongo.wide.ad.jp/scapy6") SW6 = 0 def lookup(add): add = socket.inet_pton(socket.AF_INET6, add) for k in sw6.servers.keys(): l = k.split('/') p = l[0] c = int(l[1]) h = in6_and(add, in6_cidr2mask(c)) if in6_xor(socket.inet_pton(socket.AF_INET6, p), h) == socket.inet_pton(socket.AF_INET6, '::'): return sw6.servers[k] class TracerouteResult6(TracerouteResult): def show(self): return self.make_table(lambda (s,r): (s.sprintf("%-42s,IPv6.dst%:{TCP:tcp%TCP.dport%}{UDP:udp%UDP.dport%}{ICMPv6EchoRequest:IER}"), # TODO: ICMPv6 ! s.hlim, r.sprintf("%-42s,IPv6.src% {TCP:%TCP.flags%}"+ "{ICMPv6DestUnreach:%ir,type%}{ICMPv6PacketTooBig:%ir,type%}"+ "{ICMPv6TimeExceeded:%ir,type%}{ICMPv6ParamProblem:%ir,type%}"+ "{ICMPv6EchoReply:%ir,type%}"))) def world_trace(self): warning('Not implemented for IPv6 !') def _hasinstance(self, p, i): while p: if isinstance(p, i): return True p = p.payload return False def _get_trace_id(self, s, r): if s.haslayer(TCP) or s.haslayer(UDP): trace_id = (s.src,s.dst,s.nh,s.dport) elif self._hasinstance(s, _ICMPv6): trace_id = (s.src,s.dst,s.nh,s.type,s.code) else: trace_id = (s.src,s.dst,s.nh,0) return trace_id def _get_trace(self, s, r, trace_id, trace, ports, ports_done): if not self._hasinstance(r, _ICMPv6) or r.type != 3: if ports_done.has_key(trace_id): return None ports_done[trace_id] = None p = ports.get(r.src,[]) if r.haslayer(TCP): p.append(r.sprintf(" %TCP.sport%: %TCP.flags%")) trace[s.hlim] = r.sprintf('"%IPv6.src%":T%ir,TCP.sport%') elif r.haslayer(UDP): p.append(r.sprintf(" %UDP.sport%")) trace[s.hlim] = r.sprintf('"%IPv6.src%":U%ir,UDP.sport%') elif self._hasinstance(r, _ICMPv6): p.append(r.sprintf(" ICMPv6 %type%")) trace[s.hlim] = r.sprintf('"%IPv6.src%":I%ir,type%') else: p.append(r.sprintf(" IPv6 %IPv6.nh%")) trace[s.hlim] = r.sprintf('"%IPv6.src%":P%ir,IPv6.nh%') ports[r.src] = p else: trace[s.hlim] = r.sprintf('"%IPv6.src%"') return (trace, ports, ports_done) def _get_clusters(self, ips, ASN): ASN_query_list = dict.fromkeys(map(lambda x:x.split("_")[0],ips)).keys() ASNlist = [] if SW6 and ASN == 1: def parse(x): asn,desc = None,"" for l in x.splitlines(): if not asn and l.startswith("inet6num:"): asn = l[9:].strip() elif not asn and l.startswith("CIDR:"): asn = l[5:].strip() if l.startswith("netname:") or l.startswith("NetName:"): if desc: desc += r"\n" desc += l[8:].strip() #if l.startswith("country:") or l.startswith("Country:"): # print '%s %s' % (desc, l[9:].strip()) if asn is not None and desc: break return asn,desc.strip() for ip in ASN_query_list: s = socket.socket(socket.AF_INET, socket.SOCK_STREAM) s.connect((lookup(ip), 43)) s.send("%s\n" % ip) t = s.recv(8192) buff = '' while t != '': buff = buff + t t = s.recv(8192) s.close() asn, desc = parse(buff) ASNlist.append((ip,asn,desc)) else: ASNlist = [] return ASNlist def graph(self, CL=1, padding=0, **kargs): """x.graph(CL=1, other args): CL=0 : no clustering CL=1 : use whois based clustering other args are passed to do_graph()""" self._graph(ASN=CL, padding=padding, **kargs) ############# ## Sockets ## ############# class L3RawSocket6(L3RawSocket): def __init__(self, type = ETH_P_IPV6, filter=None, iface=None, promisc=None, nofilter=0): L3RawSocket.__init__(self, type, filter, iface, promisc) # NOTE: if framentation is needed, it will be done by the kernel (RFC 2292) self.outs = socket.socket(socket.AF_INET6, socket.SOCK_RAW, socket.IPPROTO_RAW) self.ins = socket.socket(socket.AF_PACKET, socket.SOCK_RAW, socket.htons(type)) def IPv6inIP(dst='203.178.135.36', src=None): _IPv6inIP.dst = dst _IPv6inIP.src = src if not conf.L3socket == _IPv6inIP: _IPv6inIP.cls = conf.L3socket else: del(conf.L3socket) return _IPv6inIP class _IPv6inIP(SuperSocket): dst = '127.0.0.1' src = None cls = None def __init__(self, family=socket.AF_INET6, type=socket.SOCK_STREAM, proto=0, **args): SuperSocket.__init__(self, family, type, proto) self.worker = self.cls(**args) def set(self, dst, src=None): _IPv6inIP.src = src _IPv6inIP.dst = dst def nonblock_recv(self): p = self.worker.nonblock_recv() return self._recv(p) def recv(self, x): p = self.worker.recv(x) return self._recv(p, x) def _recv(self, p, x=MTU): if p is None: return p elif isinstance(p, IP): # TODO: verify checksum if p.src == self.dst and p.proto == socket.IPPROTO_IPV6: if isinstance(p.payload, IPv6): return p.payload return p def send(self, x): return self.worker.send(IP(dst=self.dst, src=self.src, proto=socket.IPPROTO_IPV6)/x) ############### ## Functions ## ############### def neighsol(host, **args): """Send an ICMPv6 Neighbor Solicitation message to determine the MAC address of the specified IPv6 address. neighsol(host) -> ans,unans""" ns = ICMPv6ND_NS(tgt=host) ns /= ICMPv6NDOptSrcLLAddr(lladdr=get_if_hwaddr(conf.iface)) r,u = sr(IPv6(hlim=255)/ns, timeout=2) return r,u def traceroute6(target, dport=80, minttl=1, maxttl=30, sport=RandShort(), l4 = None, timeout=2, **kargs): """Instant TCP traceroute with IPv6 traceroute(target, [maxttl=30], [dport=80], [sport=80]) -> None """ if l4 is None: a,b = sr(IPv6(dst=target, hlim=(minttl,maxttl))/TCP(seq=RandInt(),sport=sport, dport=dport), timeout=timeout, filter="icmp6 or tcp", **kargs) else: a,b = sr(IPv6(dst=target, hlim=(minttl,maxttl))/l4, timeout=timeout, **kargs) a = TracerouteResult6(a.res) a.display() return a,b #################### ## Layers bonding ## #################### L3Types[ETH_P_IPV6] = IPv6 LLTypes[31] = IPv6 LLNumTypes[IPv6] = 31 layer_bonds = layer_bonds + [ ( Ether, IPv6, { "type" : 0x86dd } ), ( IPerror6, TCPerror, { "nh" : socket.IPPROTO_TCP } ), ( IPerror6, UDPerror, { "nh" : socket.IPPROTO_UDP } ), ##( IPerror6, ICMPerror6, { "nh" : socket.IPPROTO_ICMPV6 } ), ( IPv6, TCP, { "nh" : socket.IPPROTO_TCP } ), ( IPv6, UDP, { "nh" : socket.IPPROTO_UDP } ), ( IP, IPv6, {"proto": socket.IPPROTO_IPV6} ), ( IPv6, IPv6, {"nh": socket.IPPROTO_IPV6} ) ] for l in layer_bonds: bind_layers(*l) del(l) conf.route6 = Route6() if __name__ == '__main__': interact(mydict=globals(), mybanner="IPv6 enabled") else: import __builtin__ __builtin__.__dict__.update(globals())