Detecting Rogue DHCP Servers with PyDHCPDiscover

It’s been awhile since I’ve released or updated a tool, so I present to you PyDHCPDiscover!

PyDHCPDiscover is a Python script that can send out DHCP requests, and parse the responses from the servers. It will not actually accept these offers, but the tool can detect what IPs the servers are offering up etc.

In this post, I will show how the script can detect rogue DHCP servers on your network.

To start, I will setup dnsmasq as a rogue DHCP server on my network.

root@kali:~# cat /etc/dnsmasq.conf 
domain-needed
bogus-priv
no-resolv
no-poll
no-hosts
expand-hosts
dhcp-range=lan,192.168.1.170,192.168.1.180,1337
dhcp-option=option:router,192.168.1.1
dhcp-option=lan,6,192.168.1.1
root@kali:~# dnsmasq -k

Once that was running, it was time to go over my script.

I based the original code for this on http://code.activestate.com/recipes/577649-dhcp-query/, but I made some modifications, additions, and proper support for DHCP options.

Once I sent out a DHCP request, I needed to parse the offer.

You can find the current code for PyDHCPDiscover below.

import socket
import struct
from uuid import getnode as get_mac
from random import randint

# Based on http://code.activestate.com/recipes/577649-dhcp-query/

def strToIP(input):
    return '.'.join(str(int(x.encode('hex'), 16)) for x in input)

def getMacString():
    mac = str(hex(get_mac())[2:])
    while (len(mac) < 12):
        mac = '0' + mac
    macB = ''
    for i in range(0, 12, 2) :
        m = int(mac[i:i + 2], 16)
        macB += struct.pack('!B', m)
    return macB

def genTransactionID():
    transactionID = ''
    for i in range(4):
        t = randint(0, 255)
        transactionID += struct.pack('!B', t)
    return transactionID

def buildDiscoverPacket(transactionID):
    # en.wikipedia.org/wiki/Dynamic_Host_Configuration_Protocol#DHCP_discovery
    
    packet = b''
    # Message type: Boot Request (1)
    packet += b'\x01'
    # Hardware type: Ethernet
    packet += b'\x01'
    # Hardware address length: 6
    packet += b'\x06'
    # Hops: 0
    packet += b'\x00'
    # Transaction ID
    packet += transactionID
    # Seconds elapsed: 0
    packet += b'\x00\x00'
    # Bootp flags: 0x8000 (Broadcast) + reserved flags
    packet += b'\x80\x00'
    # Client IP address: 0.0.0.0
    packet += b'\x00\x00\x00\x00'
    # Your (client) IP address: 0.0.0.0
    packet += b'\x00\x00\x00\x00'
    # Next server IP address: 0.0.0.0
    packet += b'\x00\x00\x00\x00'
    # Relay agent IP address: 0.0.0.0
    packet += b'\x00\x00\x00\x00'
    # Client MAC address
    packet += getMacString()
    # Client hardware address padding: 00000000000000000000
    packet += b'\x00\x00\x00\x00\x00\x00\x00\x00\x00\x00'
    # Server host name not given
    packet += b'\x00' * 67
    # Boot file name not given
    packet += b'\x00' * 125
    # Magic cookie: DHCP
    packet += '\x63\x82\x53\x63'
    # Option: (t=53,l=1) DHCP Message Type = DHCP Discover
    packet += b'\x35\x01\x01'
    # Option: (t=61,l=6) Client MAC
    packet += b'\x3d\x06' + getMacString()
    # Option: (t=55,l=3) Parameter Request List
    packet += b'\x37\x03\x03\x01\x06'
    # End Option
    packet += b'\xff'
    return packet

def getOption(key, value):
    # en.wikipedia.org/wiki/Dynamic_Host_Configuration_Protocol#DHCP_options

    optName = 'Option not found'
    optValue = 'N/A'

    if key is 1:
        optName = 'Subnet Mask'
        optValue = strToIP(value)
    elif key is 3:
        optName = 'Available Router'
        optValue = strToIP(value)
    elif key is 6:
        optName = 'Domain Name Server(s)'
        optValue = strToIP(value)
    elif key is 28:
        optName = 'Broadcast Address'
        optValue = strToIP(value)
    elif key is 51:
        optName = 'IP address Lease Time'
        optValue = str(struct.unpack('!L', value)[0])
    elif key is 53:
        optName = 'DHCP Message Type'
        if ord(value) is 1:
            optValue = 'DHCP Discover message (DHCPDiscover)'
        elif ord(value) is 2:
            optValue = 'DHCP Offer message (DHCPOffer)'
        elif ord(value) is 3:
            optValue = 'DHCP Request message (DHCPRequest)'
        elif ord(value) is 4:
            optValue = 'DHCP Decline message (DHCPDecline)'
        elif ord(value) is 5:
            optValue = 'DHCP Acknowledgment message (DHCPAck)'
        elif ord(value) is 6:
            optValue = 'DHCP Negative Acknowledgment message (DHCPNak)'
        else:
            optValue = 'Message type not supported'
    elif key is 54:
        optName = 'Server Identifier'
        optValue = strToIP(value)
    elif key is 58:
        optName = 'Renewal (T1) Time Value'
        optValue = str(struct.unpack('!L', value)[0])
    elif key is 59:
        optName = 'Rebinding (T2) Time Value'
        optValue = str(struct.unpack('!L', value)[0])
    return [optName, optValue]

def unpackOfferPacket(data, transactionID):
    # en.wikipedia.org/wiki/Dynamic_Host_Configuration_Protocol#DHCP_offer

    #print ':'.join(x.encode('hex') for x in data)
    if (data[4:8] == transactionID):
        print '\nDHCP SERVER FOUND!\n-------------------'
        
        offerIP = strToIP(data[16:20])
        nextServerIP = strToIP(data[20:24])
        dhcpOptions = data[240:]
        optionsDict = {}
        optionsOut = []
        toPrint = {}
        nextOption = dhcpOptions[0]
        while ord(nextOption) is not 255:
            optionKey = ord(nextOption)
            optionLen = ord(dhcpOptions[1])
            optionVal = dhcpOptions[2:2+optionLen]
            optionsDict[optionKey] = optionVal
            dhcpOptions = dhcpOptions[2+optionLen:]
            nextOption = dhcpOptions[0]

        for key in optionsDict:
            optionsOut.append(getOption(key, optionsDict[key]))

        #print optionsOut

        # Current iteration may not properly support more than one DNS server
        """
        DNS = []
        dnsNB = ord(data[268])/4
        for i in range(0, 4 * dnsNB, 4):
            DNS.append(strToIP(data[269 + i :269 + i + 4]))
        print('{0:20s}'.format('DNS Servers') + ' : ')
        if DNS:
            print('     {0:15s}'.format(DNS[0]))
        if len(DNS) > 1:
            for i in range(1, len(DNS)): 
                print('     {0:22s} {1:15s}'.format(' ', DNS[i]))
        """
            
        for i in range(len(optionsOut)):
            print '{0:25s} : {1:15s}'.format(optionsOut[i][0], optionsOut[i][1])

        print '{0:25s} : {1:15s}'.format('Offered IP Address', offerIP)
        print '{0:25s} : {1:15s}'.format('Gateway IP Address', nextServerIP)
        print ''

dhcpSrv = socket.socket(socket.AF_INET, socket.SOCK_DGRAM)
dhcpSrv.setsockopt(socket.SOL_SOCKET, socket.SO_BROADCAST, 1)
dhcpSrv.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)

try:
    dhcpSrv.bind(('192.168.5.100', 68))
except Exception as ex:
    print 'There was an exception with the bind: ' + str(ex)
    dhcpSrv.close()
    #exit()

transactionID = genTransactionID()

dhcpSrv.sendto(buildDiscoverPacket(transactionID), ('', 67))

print '\nDHCP Discover sent, waiting for reply\n'

dhcpSrv.settimeout(3)
try:
    while (1):
        data = dhcpSrv.recv(2048)
        #print str(data)
        unpackOfferPacket(data, transactionID)
except Exception as ex:
    if 'timed out' not in ex:
        print 'There was an exception with the offer: ' + str(ex)

dhcpSrv.close()

After I had my dnsmasq running, I ran the tool, and it discovered two DHCP servers! The 192.168.5.1 being my router, and the other being dnsmasq running on Kali. Additionally, you can see that it parses out all the options and displays them as well.

While this is definitely a more blue-team oriented tool, I could also see it being used to make sure that no drop boxes or other malicious devices are still running or trying to give out DHCP address.

Some of my next steps will be to add more DHCP options, sort the output so that it is a bit more useful, and maybe looking into sending/responding to other DHCP requests.

Finally, you can find the code and updates in my GitHub repository.

Ray Doyle

Ray Doyle is an avid pentester/security enthusiast/beer connoisseur who has worked in IT for almost 16 years now. From building machines and the software on them, to breaking into them and tearing it all down; he’s done it all. To show for it, he has obtained an OSCE, OSCP, eCPPT, GXPN, eWPT, eWPTX, SLAE, eMAPT, Security+, ICAgile CP, ITIL v3 Foundation, and even a sabermetrics certification!

He currently serves as a Senior Staff Adversarial Engineer for Avalara, and his previous position was a Principal Penetration Testing Consultant for Secureworks.

This page contains links to products that I may receive compensation from at no additional cost to you. View my Affiliate Disclosure page here. As an Amazon Associate, I earn from qualifying purchases.