我是Python新手，正在学习一些网络编程，我希望通过我的tap接口将DHCP数据包发送到我的DHCP服务器，然后等待它的响应。我尝试了几种数据包构建技术，例如结构体和 ctypes，最终使用了
scapyimport socket
from scapy.all import *
def main():
if len(sys.argv)<3:
print " fewer arguments."
sys.exit(1)
else:
tap_interface = sys.argv[1]
src_mac_address = sys.argv[2]
ethernet = Ether(dst='ff:ff:ff:ff:ff:ff',src=src_mac_address,type=0x800)
ip = IP(src ='0.0.0.0',dst='255.255.255.255')
udp =UDP (sport=68,dport=67)
fam,hw = get_if_raw_hwaddr(tap_interface)
bootp = BOOTP(chaddr = hw, ciaddr = '0.0.0.0',xid = 0x01020304,flags= 1)
dhcp = DHCP(options=[("message-type","discover"),"end"])
packet = ethernet / ip / udp / bootp / dhcp
fd = open('/dev/net/tun','r+')
TUNSETIFF = 0x400454ca
IFF_TAP = 0x0002
IFF_NO_PI = 0x1000
mode = IFF_TAP | IFF_NO_PI
ifr = struct.pack('16sH', tap_interface, IFF_TAP | IFF_NO_PI)
fcntl.ioctl(fd,TUNSETIFF,ifr)
while True:
sendp(packet, iface = tap_interface)
time.sleep(10)
if __name__ == '__main__':
main()
还有其他方法可以实现这一目标吗？如果是这样，也请提及它们。 提前致谢。
解决了！我也遇到了同样的问题，
我认为问题出在 srp() 函数上，它无法在端口 68 上接收数据包，但我创建了一个新函数，其中包含一个新线程，用于嗅探 BOOTP 消息并显示数据包字段。 你可以模拟它：
嗅探（iface=myiface，filter =“端口68和端口67”）
然后使用 srp() 或 sendp() 函数发送数据包:)
注意： 我使用了多线程机制，因为我的程序会发送消息并嗅探网络上是否有恶意 DHCP 服务器
我不确定这是否符合答案，但我们使用 scapy 来模拟 DHCP 服务器/客户端交换，以下内容为我们完成了这项工作：
discover = Ether(dst='ff:ff:ff:ff:ff:ff', src=cliMAC, type=0x0800) / IP(src='0.0.0.0', dst='255.255.255.255') / UDP(dport=67,sport=68) / BOOTP(op=1, chaddr=cliMACchaddr) / DHCP(options=[('message-type','discover'), ('end')])
我的代码和你的代码之间的主要区别似乎是 BOOTP 标头的定义方式。也许你可以尝试我的数据包定义，看看它是否有效？
这是我所做的一个示例，它获取 dhcp 地址并将其分配给 ip 接口：
我在为我的项目创建代码时的粗略 POC：
#!/usr/bin/python
from scapy.all import Ether,IP,UDP,DHCP,BOOTP,get_if_raw_hwaddr,get_if_hwaddr,conf,sniff,sendp
from pyroute2 import IPDB
from Queue import Empty
from multiprocessing import Process, Queue, Manager
from wpa_supplicant.core import WpaSupplicantDriver
from twisted.internet.selectreactor import SelectReactor
import threading
import time
import errno
import sys
import types
import netifaces
import dbus
import json
import re
class PythonDHCPScanner:
def change_ip(self,ipObject,netInterface):
ipdb = IPDB()
ips= ipdb.interfaces[self.get_interface(netInterface)]
ipAddrs = ips.ipaddr.ipv4[0]
ips.del_ip(ipAddrs['address'],ipAddrs['prefixlen'])
ips.add_ip(ipObject['ipAddr'],24)
ipdb.commit()
ipdb.routes.add(dst="default",gateway=ipObject['router'])
ipdb.commit()
def queue_get_all(self):
items = []
maxItems = 50
for numOfItemsRetrieved in range(0, maxItems):
try:
items.append(self.q.get_nowait())
except Empty, e:
break
return items
def __init__(self):
self.net_iface = netifaces.interfaces()
def dhcp_print(self,pkt):
self.q.put(str(pkt))
def get_interface(self,number):
return str(self.net_iface[number].decode())
def get_interfaces(self):
return self.net_iface
def get_dhcp_object(self,interfaceNumber):
self.q = Manager().Queue()
c = Process(target=self.callSniffer,args=(interfaceNumber,)).start()
time.sleep(0.1)
p = Process(target=self.callPacket(interfaceNumber)).start()
time.sleep(5)
if c is not None:
c.join()
dhcp = {}
for strPkt in self.queue_get_all():
try:
pkt = Ether(strPkt)
pkt.show()
if pkt[Ether].dst == get_if_hwaddr(self.get_interface(interfaceNumber)):
if pkt[DHCP]:
if pkt.getlayer(DHCP).fields['options'][0][1] == 2:
if pkt[IP]:
dhcp['ipAddr'] = pkt[IP].dst
for option in pkt.getlayer(DHCP).fields['options']:
if option == 'end':
break
dhcp[option[0]] = option[1]
print dhcp['router']
print dhcp['subnet_mask']
break
except:
continue
return dhcp
def callSniffer(self,interfaceNumber):
inter = self.get_interface(interfaceNumber)
conf.iface = inter
print inter
sniff(iface=inter,filter="udp",prn=self.dhcp_print, timeout=10)
def callPacket(self,interfaceNumber):
inter = self.get_interface(interfaceNumber)
print inter
fam,hw = get_if_raw_hwaddr(inter)
macaddress= get_if_hwaddr(inter)
conf.iface = inter
ethernet = Ether(dst="ff:ff:ff:ff:ff:ff",src=macaddress,type=0x800)
ip = IP(src="0.0.0.0",dst="255.255.255.255")
udp = UDP(sport=68,dport=67)
bootp = BOOTP(chaddr =hw,xid=0x10000000)
dhcp = DHCP(options=[("message-type","discover"),("end")])
packet=ethernet/ip/udp/bootp/dhcp
sendp(packet,iface=inter)
# get dhcp object
dave = PythonDHCPScanner()
dhcpObject = dave.get_dhcp_object(3)
# Pick interface number 3 on my box
time.sleep(1)
for dhcpKey in dhcpObject.keys():
print str(dhcpKey) + ":" + str(dhcpObject[dhcpKey])
time.sleep(1)
dave.change_ip(dhcpObject,3)