Skip to content
Snippets Groups Projects
Commit 8329256d authored by gpotter2's avatar gpotter2
Browse files

Full rework of IGMP(v1/v2) + tests

parent ea91ebad
No related branches found
No related tags found
No related merge requests found
Hide whitespace changes
Inline Side-by-side
scapy/contrib/igmp.py
+ 73
− 141
View file @ 8329256d
...@@ -17,33 +17,28 @@ ...@@ -17,33 +17,28 @@
# scapy.contrib.description = IGMP/IGMPv2 # scapy.contrib.description = IGMP/IGMPv2
# scapy.contrib.status = loads # scapy.contrib.status = loads
# TODO: scapy 2 has function getmacbyip, maybe it can replace igmpize
# at least from the MAC layer
from __future__ import print_function from __future__ import print_function
from scapy.packet import * from scapy.packet import *
from scapy.fields import * from scapy.fields import *
from scapy.layers.inet import * from scapy.layers.inet import *
from scapy.layers.l2 import DestMACField, getmacbyip
from scapy.error import warning
#--------------------------------------------------------------------------
def isValidMCAddr(ip): def isValidMCAddr(ip):
"""convert dotted quad string to long and check the first octet""" """convert dotted quad string to long and check the first octet"""
FirstOct=atol(ip)>>24 & 0xFF FirstOct=atol(ip)>>24 & 0xFF
return (FirstOct >= 224) and (FirstOct <= 239) return (FirstOct >= 224) and (FirstOct <= 239)
#--------------------------------------------------------------------------
class IGMP(Packet): class IGMP(Packet):
"""IGMP Message Class for v1 and v2. """IGMP Message Class for v1 and v2.
This class is derived from class Packet. You need to "igmpize" This class is derived from class Packet. You need call "igmpize()"
the IP and Ethernet layers before a full packet is sent. so the packet is transformed according the RFC when sent.
a=Ether(src="00:01:02:03:04:05") a=Ether(src="00:01:02:03:04:05")
b=IP(src="1.2.3.4") b=IP(src="1.2.3.4")
c=IGMP(type=0x12, gaddr="224.2.3.4") c=IGMP(type=0x12, gaddr="224.2.3.4")
c.igmpize(b, a) x = a/b/c
print "Joining IP " + c.gaddr + " MAC " + a.dst x[IGMP].igmpize()
sendp(a/b/c, iface="en0") sendp(a/b/c, iface="en0")
Parameters: Parameters:
...@@ -55,134 +50,71 @@ See RFC2236, Section 2. Introduction for definitions of proper ...@@ -55,134 +50,71 @@ See RFC2236, Section 2. Introduction for definitions of proper
IGMPv2 message format http://www.faqs.org/rfcs/rfc2236.html IGMPv2 message format http://www.faqs.org/rfcs/rfc2236.html
""" """
name = "IGMP" name = "IGMP"
igmptypes = { 0x11 : "Group Membership Query", igmptypes = { 0x11 : "Group Membership Query",
0x12 : "Version 1 - Membership Report", 0x12 : "Version 1 - Membership Report",
0x16 : "Version 2 - Membership Report", 0x16 : "Version 2 - Membership Report",
0x17 : "Leave Group"} 0x17 : "Leave Group"}
fields_desc = [ ByteEnumField("type", 0x11, igmptypes), fields_desc = [ ByteEnumField("type", 0x11, igmptypes),
ByteField("mrtime",20), ByteField("mrtime", 20),
XShortField("chksum", None), XShortField("chksum", None),
IPField("gaddr", "0.0.0.0")] IPField("gaddr", "0.0.0.0")]
#-------------------------------------------------------------------------- def post_build(self, p, pay):
def post_build(self, p, pay): """Called implicitly before a packet is sent to compute and place IGMP checksum.
"""Called implicitly before a packet is sent to compute and place IGMP checksum.
Parameters:
Parameters: self The instantiation of an IGMP class
self The instantiation of an IGMP class p The IGMP message in hex in network byte order
p The IGMP message in hex in network byte order pay Additional payload for the IGMP message
pay Additional payload for the IGMP message """
""" p += pay
p += pay if self.chksum is None:
if self.chksum is None: ck = checksum(p)
ck = checksum(p) p = p[:2]+chr(ck>>8)+chr(ck&0xff)+p[4:]
p = p[:2]+chr(ck>>8)+chr(ck&0xff)+p[4:] return p
return p
def igmpize(self):
#-------------------------------------------------------------------------- """Applies IGMP rules to the packet"""
def mysummary(self): gaddr = self.gaddr if self.gaddr else "0.0.0.0"
"""Display a summary of the IGMP object.""" underlayer = self.underlayer
# The rules are:
if isinstance(self.underlayer, IP): # 1. the Max Response time is meaningful only in Membership Queries and should be zero
return self.underlayer.sprintf("IGMP: %IP.src% > %IP.dst% %IGMP.type% %IGMP.gaddr%") # otherwise (RFC 2236, section 2.2)
else: if self.type != 0x11: # Rule 1
return self.sprintf("IGMP %IGMP.type% %IGMP.gaddr%") self.mrtime = 0
if isinstance(underlayer, IP):
#-------------------------------------------------------------------------- if (self.type == 0x11):
def igmpize(self, ip=None, ether=None): if (gaddr == "0.0.0.0"):
"""Called to explicitly fixup associated IP and Ethernet headers underlayer.dst = "224.0.0.1" # IP rule 1
elif isValidMCAddr(gaddr):
Parameters: underlayer.dst = gaddr # IP rule 3a
self The instantiation of an IGMP class. else:
ip The instantiation of the associated IP class. warning("Invalid IGMP Group Address detected !")
ether The instantiation of the associated Ethernet. return False
elif ((self.type == 0x17) and isValidMCAddr(gaddr)):
Returns: underlayer.dst = "224.0.0.2" # IP rule 2
True The tuple ether/ip/self passed all check and represents elif ((self.type == 0x12) or (self.type == 0x16)) and (isValidMCAddr(gaddr)):
a proper IGMP packet. underlayer.dst = gaddr # IP rule 3b
False One of more validation checks failed and no fields else:
were adjusted. warning("Invalid IGMP Type detected !")
return False
The function will examine the IGMP message to assure proper format. _root = self.firstlayer()
Corrections will be attempted if possible. The IP header is then properly if _root.haslayer(Ether):
adjusted to ensure correct formatting and assignment. The Ethernet header # Force recalculate Ether dst
is then adjusted to the proper IGMP packet format. _root[Ether].dst = getmacbyip(underlayer.dst)
""" return True
# The rules are: def mysummary(self):
# 1. the Max Response time is meaningful only in Membership Queries and should be zero """Display a summary of the IGMP object."""
# otherwise (RFC 2236, section 2.2) if isinstance(self.underlayer, IP):
return self.underlayer.sprintf("IGMP: %IP.src% > %IP.dst% %IGMP.type% %IGMP.gaddr%")
if (self.type != 0x11): #rule 1
self.mrtime = 0
if (self.adjust_ip(ip) == True):
if (self.adjust_ether(ip, ether) == True): return True
return False
#--------------------------------------------------------------------------
def adjust_ether (self, ip=None, ether=None):
"""Called to explicitly fixup an associated Ethernet header
The function adjusts the ethernet header destination MAC address based on
the destination IP address.
"""
# The rules are:
# 1. send to the group mac address address corresponding to the IP.dst
if ip != None and ip.haslayer(IP) and ether != None and ether.haslayer(Ether):
iplong = atol(ip.dst)
ether.dst = "01:00:5e:%02x:%02x:%02x" % ( (iplong>>16)&0x7F, (iplong>>8)&0xFF, (iplong)&0xFF )
# print "igmpize ip " + ip.dst + " as mac " + ether.dst
return True
else:
return False
#--------------------------------------------------------------------------
def adjust_ip (self, ip=None):
"""Called to explicitly fixup an associated IP header
The function adjusts the IP header based on conformance rules
and the group address encoded in the IGMP message.
The rules are:
1. Send General Group Query to 224.0.0.1 (all systems)
2. Send Leave Group to 224.0.0.2 (all routers)
3a.Otherwise send the packet to the group address
3b.Send reports/joins to the group address
4. ttl = 1 (RFC 2236, section 2)
5. send the packet with the router alert IP option (RFC 2236, section 2)
"""
if ip != None and ip.haslayer(IP):
if (self.type == 0x11):
if (self.gaddr == "0.0.0.0"):
ip.dst = "224.0.0.1" # IP rule 1
retCode = True
elif isValidMCAddr(self.gaddr):
ip.dst = self.gaddr # IP rule 3a
retCode = True
else: else:
print("Warning: Using invalid Group Address") return self.sprintf("IGMP %IGMP.type% %IGMP.gaddr%")
retCode = False
elif ((self.type == 0x17) and isValidMCAddr(self.gaddr)):
ip.dst = "224.0.0.2" # IP rule 2
retCode = True
elif ((self.type == 0x12) or (self.type == 0x16)) and (isValidMCAddr(self.gaddr)):
ip.dst = self.gaddr # IP rule 3b
retCode = True
else:
print("Warning: Using invalid IGMP Type")
retCode = False
else:
print("Warning: No IGMP Group Address set")
retCode = False
if retCode == True:
ip.ttl=1 # IP Rule 4
ip.options=[IPOption_Router_Alert()] # IP rule 5
return retCode
bind_layers( IP, IGMP, frag=0, proto=2)
bind_layers( IP, IGMP, frag=0,
proto=2,
ttl=1,
options=[IPOption_Router_Alert()])
scapy/contrib/igmp.uts 0 → 100644
+ 73
− 0
View file @ 8329256d
############
% IGMP tests
############
+ Basic IGMP tests
= Build IGMP - Basic
a=Ether(src="00:01:02:03:04:05")
b=IP(src="1.2.3.4")
c=IGMP(gaddr="0.0.0.0")
x = a/b/c
x[IGMP].igmpize()
assert x.mrtime == 20
assert x[IP].dst == "224.0.0.1"
= Build IGMP - Custom membership
a=Ether(src="00:01:02:03:04:05")
b=IP(src="1.2.3.4")
c=IGMP(gaddr="224.0.1.2")
x = a/b/c
x[IGMP].igmpize()
assert x.mrtime == 20
assert x[IP].dst == "224.0.1.2"
= Build IGMP - LG
a=Ether(src="00:01:02:03:04:05")
b=IP(src="1.2.3.4")
c=IGMP(type=0x17, gaddr="224.2.3.4")
x = a/b/c
x[IGMP].igmpize()
assert x.dst == "01:00:5e:00:00:02"
assert x.mrtime == 0
assert x[IP].dst == "224.0.0.2"
= Change IGMP params
x = Ether(src="00:01:02:03:04:05")/IP()/IGMP()
x[IGMP].igmpize()
assert x.mrtime == 20
assert x[IP].dst == "224.0.0.1"
x = Ether(src="00:01:02:03:04:05")/IP()/IGMP(gaddr="224.2.3.4", type=0x12)
x.mrtime = 1
x[IGMP].igmpize()
x = Ether(str(x))
assert x.mrtime == 0
x.gaddr = "224.3.2.4"
x[IGMP].igmpize()
assert x.dst == "01:00:5e:03:02:04"
= Test mysummary
x = Ether(src="00:01:02:03:04:05")/IP(src="192.168.0.1")/IGMP(gaddr="224.0.0.2", type=0x17)
x[IGMP].igmpize()
assert x[IGMP].mysummary() == "IGMP: 192.168.0.1 > 224.0.0.2 Leave Group 224.0.0.2"
assert IGMP().mysummary() == "IGMP Group Membership Query 0.0.0.0"
= IGMP - misc
~ netaccess
x = Ether(src="00:01:02:03:04:05")/IP(dst="192.168.0.1")/IGMP(gaddr="www.google.fr", type=0x11)
x = Ether(str(x))
x[IGMP].igmpize()
assert x[IP].dst == "192.168.0.1"
x = Ether(src="00:01:02:03:04:05")/IP(dst="192.168.0.1")/IGMP(gaddr="124.0.2.1", type=0x00)
x[IGMP].igmpize()
assert x[IP].dst == "192.168.0.1"
\ No newline at end of file
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment