+++ /dev/null
-# Anemon Dhcp
-# Copyright (C) 2005 Mathieu Ignacio -- mignacio@april.org
-#
-# This program is free software; you can redistribute it and/or modify
-# it under the terms of the GNU General Public License as published by
-# the Free Software Foundation; either version 2 of the License, or
-# (at your option) any later version.
-#
-# 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.
-#
-# You should have received a copy of the GNU General Public License
-# along with this program; if not, write to the Free Software
-# Foundation, Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301, USA
-
-import operator
-from struct import unpack
-from struct import pack
-from dhcp_basic_packet import *
-from dhcp_constants import *
-from type_ipv4 import ipv4
-from type_strlist import strlist
-class DhcpPacket(DhcpBasicPacket):
-
-
- # Useful function for debugging
- def PrintHeaders(self):
- print "# Header fields\n"
- print "readable_dhcp_headers = {"
- for opt in ['op','htype','hlen','hops','xid','secs','flags',
- 'ciaddr','yiaddr','siaddr','giaddr','chaddr','sname','file'] :
- begin = DhcpFields[opt][0]
- end = DhcpFields[opt][0]+DhcpFields[opt][1]
- data = self.packet_data[begin:end]
- if DhcpFieldsTypes[opt] == "int" : result = str(data[0])
- if DhcpFieldsTypes[opt] == "int2" : result = str(data[0]*256+data[0])
- if DhcpFieldsTypes[opt] == "int4" : result = str(ipv4(data).int())
- if DhcpFieldsTypes[opt] == "str" : result = strlist(data).str()
- if DhcpFieldsTypes[opt] == "ipv4" : result = ipv4(data).str()
- if DhcpFieldsTypes[opt] == "hwmac" : result = "".join(map(chr,data))
-
- line = "\t'"+opt+"':"+str(data)+",\t# "+result
- print line
- print "\t'end':'true'}"
-
- # Useful function for debugging
- def PrintOptions(self):
- print "# Options fields"
- print "readable_dhcp_options = {"
- for opt in self.options_data.keys():
- data = self.options_data[opt]
- result = ""
- optnum = DhcpOptions[opt]
- if DhcpOptionsTypes[optnum] == "char" : result = str(data[0])
- if DhcpOptionsTypes[optnum] == "16-bits" : result = str(data[0]*256+data[0])
- if DhcpOptionsTypes[optnum] == "32bits" : result = str(ipv4(data).int())
- if DhcpOptionsTypes[optnum] == "string" : result = strlist(data).str()
- if DhcpOptionsTypes[optnum] == "ipv4" : result = ipv4(data).str()
- if DhcpOptionsTypes[optnum] == "ipv4+" :
- for i in range(0,len(data),4) :
- if len(data[i:i+4]) == 4 :
- result += ipv4(data[i:i+4]).str() + " - "
- line = "\t'"+opt+"':"+str(data)+",\t# "+result
- print line
- print "\t'end':'true'}"
-
-
-
- # FIXME: This is called from IsDhcpSomethingPacket, but is this really
- # needed? Or maybe this testing should be done in
- # DhcpBasicPacket.DecodePacket().
-
- # Test Packet Type
- def IsDhcpSomethingPacket(self,type):
- if self.IsDhcpPacket() == False : return False
- if self.IsOption("dhcp_message_type") == False : return False
- if self.GetOption("dhcp_message_type") != type : return False
- return True
-
- def IsDhcpDiscoverPacket(self):
- return self.IsDhcpSomethingPacket([1])
-
- def IsDhcpOfferPacket(self):
- return self.IsDhcpSomethingPacket([2])
-
- def IsDhcpRequestPacket(self):
- return self.IsDhcpSomethingPacket([3])
-
- def IsDhcpDeclinePacket(self):
- return self.IsDhcpSomethingPacket([4])
-
- def IsDhcpAckPacket(self):
- return self.IsDhcpSomethingPacket([5])
-
- def IsDhcpNackPacket(self):
- return self.IsDhcpSomethingPacket([6])
-
- def IsDhcpReleasePacket(self):
- return self.IsDhcpSomethingPacket([7])
-
- def IsDhcpInformPacket(self):
- return self.IsDhcpSomethingPacket([8])
-
-
- def GetMultipleOptions(self,options=()):
- result = {}
- for each in options:
- result[each] = self.GetOption(each)
- return result
-
- def SetMultipleOptions(self,options={}):
- for each in options.keys():
- self.SetOption(each,options[each])
-
-
-
-
-
-
- # Creating Response Packet
-
- # Server-side functions
- # From RFC 2132 page 28/29
- def CreateDhcpOfferPacketFrom(self,src): # src = discover packet
- self.SetOption("htype",src.GetOption("htype"))
- self.SetOption("xid",src.GetOption("xid"))
- self.SetOption("flags",src.GetOption("flags"))
- self.SetOption("giaddr",src.GetOption("giaddr"))
- self.SetOption("chaddr",src.GetOption("chaddr"))
- self.SetOption("ip_address_lease_time",src.GetOption("ip_address_lease_time"))
- self.TransformToDhcpOfferPacket()
-
- def TransformToDhcpOfferPacket(self):
- self.SetOption("dhcp_message_type",[2])
- self.SetOption("op",[2])
- self.SetOption("hlen",[6])
-
- self.DeleteOption("secs")
- self.DeleteOption("ciaddr")
- self.DeleteOption("request_ip_address")
- self.DeleteOption("parameter_request_list")
- self.DeleteOption("client_identifier")
- self.DeleteOption("maximum_message_size")
-
-
-
-
-
- """ Dhcp ACK packet creation """
- def CreateDhcpAckPacketFrom(self,src): # src = request or inform packet
- self.SetOption("htype",src.GetOption("htype"))
- self.SetOption("xid",src.GetOption("xid"))
- self.SetOption("ciaddr",src.GetOption("ciaddr"))
- self.SetOption("flags",src.GetOption("flags"))
- self.SetOption("giaddr",src.GetOption("giaddr"))
- self.SetOption("chaddr",src.GetOption("chaddr"))
- self.SetOption("ip_address_lease_time_option",src.GetOption("ip_address_lease_time_option"))
- self.TransformToDhcpAckPacket()
-
- def TransformToDhcpAckPacket(self): # src = request or inform packet
- self.SetOption("op",[2])
- self.SetOption("hlen",[6])
- self.SetOption("dhcp_message_type",[5])
-
- self.DeleteOption("secs")
- self.DeleteOption("request_ip_address")
- self.DeleteOption("parameter_request_list")
- self.DeleteOption("client_identifier")
- self.DeleteOption("maximum_message_size")
-
-
- """ Dhcp NACK packet creation """
- def CreateDhcpNackPacketFrom(self,src): # src = request or inform packet
-
- self.SetOption("htype",src.GetOption("htype"))
- self.SetOption("xid",src.GetOption("xid"))
- self.SetOption("flags",src.GetOption("flags"))
- self.SetOption("giaddr",src.GetOption("giaddr"))
- self.SetOption("chaddr",src.GetOption("chaddr"))
- self.TransformToDhcpNackPacket()
-
- def TransformToDhcpNackPacket(self):
- self.SetOption("op",[2])
- self.SetOption("hlen",[6])
- self.DeleteOption("secs")
- self.DeleteOption("ciaddr")
- self.DeleteOption("yiaddr")
- self.DeleteOption("siaddr")
- self.DeleteOption("sname")
- self.DeleteOption("file")
- self.DeleteOption("request_ip_address")
- self.DeleteOption("ip_address_lease_time_option")
- self.DeleteOption("parameter_request_list")
- self.DeleteOption("client_identifier")
- self.DeleteOption("maximum_message_size")
- self.SetOption("dhcp_message_type",[6])
-
-
-
-
-
-
-
- """ GetClientIdentifier """
-
- def GetClientIdentifier(self) :
- if self.IsOption("client_identifier") :
- return self.GetOption("client_identifier")
- return []
-
- def GetGiaddr(self) :
- return self.GetOption("giaddr")
-
- def GetHardwareAddress(self) :
- length = self.GetOption("hlen")[0]
- full_hw = self.GetOption("chaddr")
- if length!=[] and length<len(full_hw) : return full_hw[0:length]
- return full_hw
-
projects / invirt/packages/invirt-dhcp.git / blobdiff