blob: 653b1e5408d601437ca92993d55def44f0770e7a [file] [log] [blame]
#
# Copyright 2016-present Ciena Corporation
#
# Licensed under the Apache License, Version 2.0 (the "License");
# you may not use this file except in compliance with the License.
# You may obtain a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.
#
from scapy.all import *
conf.verb = 0 # Disable Scapy verbosity
conf.checkIPaddr = 0 # Don't check response packets for matching destination IPs
class DHCPTest:
def __init__(self, seed_ip = '192.168.1.1', iface = 'veth0'):
self.seed_ip = seed_ip
self.seed_mac = self.ipToMac(self.seed_ip)
self.iface = iface
self.mac_map = {}
self.mac_inverse_map = {}
self.bootpmac = None
self.dhcpresp = None
def is_mcast(self, ip):
mcast_octet = (atol(ip) >> 24) & 0xff
return True if mcast_octet >= 224 and mcast_octet <= 239 else False
def discover(self, mac = None, update_seed = False):
'''Send a DHCP discover/offer'''
if mac is None:
mac = self.seed_mac
if update_seed:
self.seed_ip = self.incIP(self.seed_ip)
self.seed_mac = self.ipToMac(self.seed_ip)
mac = self.seed_mac
chmac = self.macToChaddr(mac)
self.bootpmac = chmac
L2 = Ether(dst="ff:ff:ff:ff:ff:ff", src=mac)
L3 = IP(src="0.0.0.0", dst="255.255.255.255")
L4 = UDP(sport=68, dport=67)
L5 = BOOTP(chaddr=chmac)
L6 = DHCP(options=[("message-type","discover"),"end"])
resp = srp1(L2/L3/L4/L5/L6, filter="udp and port 68", timeout=10, iface=self.iface)
self.dhcpresp = resp
try:
srcIP = resp.yiaddr
serverIP = resp.siaddr
except AttributeError:
print("Failed to acquire IP via DHCP for %s on interface %s" %(mac, self.iface))
return (None, None)
subnet_mask = "0.0.0.0"
for x in resp.lastlayer().options:
if(x == 'end'):
break
op,val = x
if(op == "subnet_mask"):
subnet_mask = val
elif(op == 'server_id'):
server_id = val
L5 = BOOTP(chaddr=chmac, yiaddr=srcIP)
L6 = DHCP(options=[("message-type","request"), ("server_id",server_id),
("subnet_mask",subnet_mask), ("requested_addr",srcIP), "end"])
srp(L2/L3/L4/L5/L6, filter="udp and port 68", timeout=10, iface=self.iface)
self.mac_map[mac] = (srcIP, serverIP)
self.mac_inverse_map[srcIP] = (mac, serverIP)
return (srcIP, serverIP)
def only_discover(self, mac = None, desired = False, lease_time = False):
'''Send a DHCP discover'''
if mac is None:
mac = self.seed_mac
chmac = self.macToChaddr(mac)
self.bootpmac = chmac
L2 = Ether(dst="ff:ff:ff:ff:ff:ff", src=mac)
L3 = IP(src="0.0.0.0", dst="255.255.255.255")
L4 = UDP(sport=68, dport=67)
L5 = BOOTP(chaddr=chmac)
if desired:
L6 = DHCP(options=[("message-type","discover"),("requested_addr",self.seed_ip),"end"])
elif time:
L6 = DHCP(options=[("message-type","discover"),("lease_time",700),"end"])
elif lease_time:
L6 = DHCP(options=[("message-type","discover"),"end"])
resp = srp1(L2/L3/L4/L5/L6, filter="udp and port 68", timeout=10, iface=self.iface)
if resp == None:
return (None, None, None)
self.dhcpresp = resp
for x in resp.lastlayer().options:
if(x == 'end'):
break
op,val = x
if(op == "message-type"):
if(val == 2):
try:
srcIP = resp.yiaddr
serverIP = resp.siaddr
except AttributeError:
print "In Attribute error."
print("Failed to acquire IP via DHCP for %s on interface %s" %(mac, self.iface))
return (None, None, None)
if lease_time == True:
for x in resp.lastlayer().options:
if(x == 'end'):
break
op,val = x
if(op == "lease_time"):
return (srcIP, serverIP, mac, val)
else:
return (srcIP, serverIP, mac)
elif(val == 6):
return (None, None, mac)
def only_request(self, cip, mac):
'''Send a DHCP offer'''
subnet_mask = "0.0.0.0"
for x in self.dhcpresp.lastlayer().options:
if(x == 'end'):
break
op,val = x
if(op == "subnet_mask"):
subnet_mask = val
elif(op == 'server_id'):
server_id = val
L2 = Ether(dst="ff:ff:ff:ff:ff:ff", src=mac)
L3 = IP(src="0.0.0.0", dst="255.255.255.255")
L4 = UDP(sport=68, dport=67)
L5 = BOOTP(chaddr=self.bootpmac, yiaddr=cip)
L6 = DHCP(options=[("message-type","request"), ("server_id",server_id),
("subnet_mask",subnet_mask), ("requested_addr",cip), "end"])
resp=srp1(L2/L3/L4/L5/L6, filter="udp and port 68", timeout=10, iface=self.iface)
if resp == None:
return (None, None)
for x in resp.lastlayer().options:
if(x == 'end'):
break
op,val = x
if(op == "message-type"):
if(val == 5):
try:
srcIP = resp.yiaddr
serverIP = resp.siaddr
except AttributeError:
print "In Attribute error."
print("Failed to acquire IP via DHCP for %s on interface %s" %(mac, self.iface))
return (None, None)
self.mac_map[mac] = (srcIP, serverIP)
self.mac_inverse_map[srcIP] = (mac, serverIP)
return (srcIP, serverIP)
elif(val == 6):
return (None, None)
def discover_next(self):
'''Send next dhcp discover/request with updated mac'''
return self.discover(update_seed = True)
def release(self, ip):
'''Send a DHCP discover/offer'''
if ip is None:
return False
if not self.mac_inverse_map.has_key(ip):
return False
mac, server_ip = self.mac_inverse_map[ip]
chmac = self.macToChaddr(mac)
L2 = Ether(dst="ff:ff:ff:ff:ff:ff", src=mac)
L3 = IP(src="0.0.0.0", dst="255.255.255.255")
L4 = UDP(sport=68, dport=67)
L5 = BOOTP(chaddr=chmac, ciaddr = ip)
L6 = DHCP(options=[("message-type","release"), ("server_id", server_ip), "end"])
sendp(L2/L3/L4/L5/L6, iface = self.iface)
del self.mac_map[mac]
del self.mac_inverse_map[ip]
return True
def macToChaddr(self, mac):
rv = []
mac = mac.split(":")
for x in mac:
rv.append(chr(int(x, 16)))
return reduce(lambda x,y: x + y, rv)
def get_ip(self, mac):
if self.mac_map.has_key(mac):
return self.mac_map[mac]
return (None, None)
def get_mac(self, ip):
if self.mac_inverse_map.has_key(ip):
return self.mac_inverse_map[ip]
return (None, None)
def ipToMac(self, ip):
'''Generate a mac from a ip'''
mcast = self.is_mcast(ip)
mac = "01:00:5e" if mcast == True else "00:00:00"
octets = ip.split(".")
for x in range(1,4):
num = str(hex(int(octets[x])))
num = num.split("x")[1]
if len(num) < 2:
num = "0" + str(num)
mac += ":" + num
return mac
def incIP(self, ip, n=1):
'''Increment an IP'''
if n < 1:
return ip
o = ip.split(".")
for ii in range(3,-1,-1):
if int(o[ii]) < 255:
o[ii] = str(int(o[ii]) + 1)
break
else:
o[ii] = str(0)
n -= 1
return self.incIP(".".join(o), n)