用python编写两个代码,用于模拟DHCP协议的动态分配IP的过程,因为之前遇到一个课程实现,不下心选择了DHCP这个题目,网上能找到的资料不多,为此自己研究一晚上,写了这个不全但是还能勉强糊弄过得去的代码,因为第二天就要验收,所以没来得及做更多优化,也没有做窗口,反正能过就行。本文仅模拟客户端第一次请求,即客户端发送discover广播报文,服务器响应offer报文,客户端发送request报文,服务器响应ACK确认报文。客户端代码可复制多份,形成多客户端,复制后的每个客户端的mac地址记得改一下,不改也能运行。代码中存在一些多余冗余,主要是本人调式用的,后续IP地址租用等等自己改代码,这里只模拟了四个报文,服务器一直监听端口,对响应报文给予响应,但是没有对request的IP地址进行验证,需要的同学自己该代码。通过抓包软件可以抓到DHCP报文,这里用到python第三方库scapy,和Pycharm软件,自己去下载,服务器单独运行在一个pycharm项目,客户端单独运行pycharm项目。需要用窗口显示的可以自己弄一个显示窗口,运行结果如下:
服务器端:

客户端10:

客户端20:

服务器代码:
 

from scapy.all import *
import random
import time

from scapy.layers.dhcp import DHCP, BOOTP
from scapy.layers.inet import IP, UDP
from scapy.layers.l2 import Ether

def build_dns_option(dns_servers):
    """
    手动构建 DNS 服务器 DHCP 选项的字节串
    """
    option_code = 6  # domain_name_server 的 DHCP 选项代码
    option_data = b''.join([socket.inet_aton(server) for server in dns_servers.split(',')])
    option_length = len(option_data) + 1  # 长度包括选项代码本身
    return bytes([option_code]) + bytes([option_length]) + option_data

def respond_to_dhcp_discover(pkt):
    if DHCP in pkt and pkt[DHCP].options[0][1] == 1:
        client_mac = pkt[Ether].src
        print(f"\n发现discover请求,mac地址为:{client_mac}")
        client_ip = "192.168.1." + str(random.randint(100, 200)) #随机分配ip给客户端
        global clien_t_ip  #定义全局变量存client的ip
        clien_t_ip = client_ip
        dns_option = build_dns_option("8.8.8.8,8.8.4.4") #构建 DNS 服务器 DHCP 选项的字节串

        dhcp_offer = Ether(src=get_if_hwaddr(conf.iface), dst=client_mac) / \
                     IP(src="192.168.1.1", dst="255.255.255.255") / \
                     UDP(sport=67, dport=68) / \
                     BOOTP(op=2, chaddr=client_mac, yiaddr=client_ip) / \
                     DHCP(options=[
                         ("message-type", "offer"),
                         ("server_id", "192.168.1.1"),
                         ("lease_time", 43200),
                         ("subnet_mask", "255.255.255.0"),
                         ("router", "192.168.1.1"),
                         (b'\x06' + bytes([len(dns_option)]) + dns_option),  # 添加 DNS 服务器选项
                         ("end")
                     ])

        sendp(dhcp_offer, verbose=0)
        print(f"发送DHCP Offer给:客户端mac: {client_mac} 随机分配客户端IP为: {client_ip}")
        print(" 为DHCP 客户端可以提供 IP 地址中……")
def respond_to_dhcp_request(pkt):
    if DHCP in pkt and pkt[DHCP].options[0][1] == 3:
        print("发现request请求报文")
        requested_ip = pkt[DHCP].options[2][1]  #
        client_mac = pkt[Ether].src
        print(f"对方mac地址为:{client_mac}")
        server_id = pkt[DHCP].options[2][1]

        # print(f"requested_ip{requested_ip}\n")
        print(f"确认是本服务器id:{server_id}")

        if server_id == "192.168.1.1":
            dns_option = build_dns_option("8.8.8.8,8.8.4.4")  # 构建 DNS 服务器 DHCP 选项的字节串
            # print("22222222222")
            dhcp_ack = Ether(src=get_if_hwaddr(conf.iface), dst=client_mac) / \
                        IP(src="192.168.1.1", dst="255.255.255.255") / \
                        UDP(sport=67, dport=68) / \
                        BOOTP(op=2, chaddr=client_mac, yiaddr=clien_t_ip) / \
                        DHCP(options=[
                            ("message-type", "ack"),
                            ("server_id", server_id),
                            ("lease_time", 43200),
                            ("subnet_mask", "255.255.255.0"),
                            ("router", "192.168.1.1"),
                            (b'\x06' + bytes([len(dns_option)]) + dns_option),  # 手动添加 DNS 服务器选项
                            ("end")
                        ])
            # print("33333333333")
            sendp(dhcp_ack, verbose=0)
            print(f"发送DHCP ACK 确认 to {client_mac} for IP {clien_t_ip}")
            print("分配IP成功!\n\n")



# 监听并同时响应DISCOVER和REQUEST报文
while(1):
    filter_str = "udp and (port 67 or port 68)"
    sniff(prn=lambda x: respond_to_dhcp_discover(x) or respond_to_dhcp_request(x),
          filter=filter_str, store=0, iface="WLAN") #WLAN是网卡

客户端10代码:

from scapy.all import *
import random
import time
import binascii

from scapy.layers.dhcp import BOOTP, DHCP
from scapy.layers.inet import IP, UDP
from scapy.layers.l2 import Ether


def send_dhcp_discover(interface="WLAN"):
    """
    发送DHCP DISCOVER报文
    """
    # 使用 binascii.hexlify 转换MAC地址字符串为原始字节
    mac_bytes = binascii.unhexlify("00:11:22:33:44:55".replace(':', ''))#这里修改客户端mac地址
    #global clien_t_mac  #定义全局变量用来存client_mac地址
    #clien_t_mac = mac_bytes
    #print(f"本地客户端mac为:{mac_bytes}")
    dhcp_discover = Ether(src=mac_bytes, dst="ff:ff:ff:ff:ff:ff") \
                    / IP(src="0.0.0.0", dst="255.255.255.255") \
                    / UDP(sport=68, dport=67) \
                    / BOOTP(chaddr=mac_bytes) \
                    / DHCP(options=[("message-type", "discover"), ("end")])
    sendp(dhcp_discover, iface=interface, verbose=0)
    print("客户端已发送discover广播请求:src=\"0.0.0.0\", dst=\"255.255.255.255\"")

def respond_to_dhcp_offer(pkt):
    """
    响应接收到的DHCP OFFER报文
    """
    if DHCP in pkt and pkt[DHCP].options[0][1] == 2:  # 检查是否为DHCP OFFER报文
        server_id = pkt[DHCP].options[1][1]  # 服务器ID
        offered_ip = pkt[BOOTP].yiaddr  # 提供的IP地址
        client_mac = pkt[Ether].src  # 客户端MAC地址
        print(f"确认与本机mac地址相同:{client_mac}")
        print(f"客户端接收到offer响应报文:服务器ID:{server_id} offer提供的ip:{offered_ip}")
        # 使用正确的方法处理MAC地址
        mac_bytes = binascii.unhexlify("00:11:22:33:44:55".replace(':', ''))

        # 构造DHCP REQUEST报文
        dhcp_request = Ether(src=mac_bytes, dst="ff:ff:ff:ff:ff:ff") / \
                       IP(src="0.0.0.0", dst="255.255.255.255") / \
                       UDP(sport=68, dport=67) / \
                       BOOTP(chaddr=mac_bytes, yiaddr=offered_ip) / \
                       DHCP(options=[
                           ("message-type", "request"),
                           ("requested_addr", offered_ip),
                           ("server_id", server_id),
                           ("end")
                       ])


        sendp(dhcp_request, verbose=0)  #发送DHCP REQUEST报文
        print(f"向服务器发送广播(request)请求ip地址: {offered_ip}  服务器ip: {server_id}")

def ack_ack(pkt):
        if DHCP in pkt and pkt[DHCP].options[0][1] == 5:
            client_ip = pkt[BOOTP].yiaddr  # 分配给客户端的IP地址
            subnet_mask = next((opt[1] for opt in pkt[DHCP].options if opt[0] == 'subnet_mask'), None)  # 子网掩码
            router_ip = next((opt[1] for opt in pkt[DHCP].options if opt[0] == 'router'), None)  # 默认网关
            dns_servers = [opt[1] for opt in pkt[DHCP].options if opt[0] == 6]  # DNS服务器列表

            print(f"客户端接收到DHCP ACK包:")
            print(f" - 分配的IP地址: {client_ip}")
            if subnet_mask:
                print(f" - 子网掩码: {subnet_mask}")
            if router_ip:
                print(f" - 默认网关: {router_ip}")




filter_str = "udp and (port 67 or port 68)"  # 设置过滤器,只捕获DHCP流量

send_dhcp_discover()  # 先发送一个DHCP DISCOVER报文

#sniff(prn=respond_to_dhcp_offer, filter=filter_str, store=0, iface="WLAN") # 然后开始监听并响应OFFER报文

sniff(prn=lambda x: respond_to_dhcp_offer(x) or ack_ack(x),
        filter=filter_str, store=0, iface="WLAN")


客户端20代码:
 

from scapy.all import *
import random
import time
import binascii

from scapy.layers.dhcp import BOOTP, DHCP
from scapy.layers.inet import IP, UDP
from scapy.layers.l2 import Ether


def send_dhcp_discover(interface="WLAN"):
    """
    发送DHCP DISCOVER报文
    """
    # 使用 binascii.hexlify 转换MAC地址字符串为原始字节
    mac_bytes = binascii.unhexlify("00:11:22:33:44:66".replace(':', ''))
    global clien_t_mac  #定义全局变量用来存client_mac地址
    clien_t_mac = mac_bytes
    # print(f"客户端mac为:{clien_t_mac}")
    dhcp_discover = Ether(src=mac_bytes, dst="ff:ff:ff:ff:ff:ff") \
                    / IP(src="0.0.0.0", dst="255.255.255.255") \
                    / UDP(sport=68, dport=67) \
                    / BOOTP(chaddr=mac_bytes) \
                    / DHCP(options=[("message-type", "discover"), ("end")])
    sendp(dhcp_discover, iface=interface, verbose=0)
    print("客户端已发送discover广播请求:src=\"0.0.0.0\", dst=\"255.255.255.255\"")

def respond_to_dhcp_offer(pkt):
    """
    响应接收到的DHCP OFFER报文
    """
    if DHCP in pkt and pkt[DHCP].options[0][1] == 2:  # 检查是否为DHCP OFFER报文
        server_id = pkt[DHCP].options[1][1]  # 服务器ID
        offered_ip = pkt[BOOTP].yiaddr  # 提供的IP地址
        client_mac = pkt[Ether].src  # 客户端MAC地址
        #  print(f"确认与本机mac地址相同:{client_mac}")
        print(f"客户端接收到offer响应报文:服务器ID:{server_id} offer提供的ip:{offered_ip}")
        # 使用正确的方法处理MAC地址
        mac_bytes = binascii.unhexlify("00:11:22:33:44:66".replace(':', ''))

        # 构造DHCP REQUEST报文
        dhcp_request = Ether(src=mac_bytes, dst="ff:ff:ff:ff:ff:ff") / \
                       IP(src="0.0.0.0", dst="255.255.255.255") / \
                       UDP(sport=68, dport=67) / \
                       BOOTP(chaddr=mac_bytes, yiaddr=offered_ip) / \
                       DHCP(options=[
                           ("message-type", "request"),
                           ("requested_addr", offered_ip),
                           ("server_id", server_id),
                           ("end")
                       ])


        sendp(dhcp_request, verbose=0)  #发送DHCP REQUEST报文
        print(f"向服务器发送广播(request)请求ip地址: {offered_ip}  服务器ip: {server_id}")

def ack_ack(pkt):
        if DHCP in pkt and pkt[DHCP].options[0][1] == 5:
            client_ip = pkt[BOOTP].yiaddr  # 分配给客户端的IP地址
            subnet_mask = next((opt[1] for opt in pkt[DHCP].options if opt[0] == 'subnet_mask'), None)  # 子网掩码
            router_ip = next((opt[1] for opt in pkt[DHCP].options if opt[0] == 'router'), None)  # 默认网关
            dns_servers = [opt[1] for opt in pkt[DHCP].options if opt[0] == 6]  # DNS服务器列表

            print(f"客户端接收到DHCP ACK包:")
            print(f" - 分配的IP地址: {client_ip}")
            if subnet_mask:
                print(f" - 子网掩码: {subnet_mask}")
            if router_ip:
                print(f" - 默认网关: {router_ip}")




filter_str = "udp and (port 67 or port 68)"  # 设置过滤器,只捕获DHCP流量

send_dhcp_discover()  # 先发送一个DHCP DISCOVER报文

#sniff(prn=respond_to_dhcp_offer, filter=filter_str, store=0, iface="WLAN") # 然后开始监听并响应OFFER报文

sniff(prn=lambda x: respond_to_dhcp_offer(x) or ack_ack(x),
        filter=filter_str, store=0, iface="WLAN")


备注:大学的同学拿回去记得改一下代码啊!!!

Logo

北京人形旗下天工造物具身智能开源社区,聚焦具身天工与慧思开物两大平台

更多推荐