使用 python scapy 发送 DHCP Discover

Question

我是 python 新手，正在学习一些网络编程，我希望通过我的 tap 接口向我的 DHCP 服务器发送一个 DHCP 数据包，并期待它的一些响应。我尝试了几种数据包构建技术，例如结构和 ctypes，最终使用了 scapy。在这里，我能够发送 DHCP 数据包，但无法从 DHCP 服务器获得任何响应（使用 Wireshark 和 tcpdump 分析）..我的数据包看起来与原始 DHCP 数据包相同，但未能获得响应。这是我的代码

import 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()

还有其他方法可以实现吗？如果是这样，请务必提及它们。 提前致谢。

Answer 1

解决了！我遇到了同样的问题，

我认为问题出在 srp() 函数上，它无法在端口 68 上接收数据包，但我创建了一个新函数，其中包含一个新线程，用于嗅探 BOOTP 消息并显示数据包字段。 你可以模拟它：

sniff(iface=myiface, filter="port 68 and port 67")

然后使用 srp() 或 sendp() func 发送数据包 :)

注意： 我使用了多线程机制，因为如果网络上有恶意 DHCP 服务器，我的程序会发送消息和嗅探

Answer 2

我不确定这是否可以作为答案，但我们使用 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 标头的定义方式。也许你可以试试我的数据包定义，看看它是否有效？

Answer 3

这是我做的一个例子，它获取一个 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)