dhcppython icon indicating copy to clipboard operation
dhcppython copied to clipboard

struct.error: unpack requires a buffer of 7 bytes

Open pujianto opened this issue 3 years ago • 0 comments

Hi, Currently, I am using dhcppython, but I encounter an error when trying to run packet.view_packet() on an incoming packet (for this case, it's from dhcpcd dhcp client).

Traceback (most recent call last):
  File "/usr/local/lib/python3.10/asyncio/events.py", line 80, in _run
    self._context.run(self._callback, *self._args)
  File "/usr/local/lib/python3.10/asyncio/selector_events.py", line 1026, in _read_ready
    self._protocol.datagram_received(data, addr)
  File "/app/tmp.py", line 25, in datagram_received
    raise e
  File "/app/tmp.py", line 21, in datagram_received
    logging.debug(packet.view_packet())
  File "/usr/local/lib/python3.10/site-packages/dhcppython/packet.py", line 261, in view_packet
    f"{opt.key} {opt.value[opt.key]} ({opt.length})", opt.length
  File "/usr/local/lib/python3.10/site-packages/dhcppython/options.py", line 1617, in value
    hwtype, hwaddr = struct.unpack(">B6s", self.data)
struct.error: unpack requires a buffer of 7 bytes

To replicate this issue, run this small DHCP listener, and run dhcpcd from another PC / VM

import asyncio
import logging
import signal
import socket
import sys
from asyncio import Future, transports

from dhcppython.packet import DHCPPacket


class DHCPProtocol(asyncio.DatagramProtocol):

    def connection_made(self, transport: transports.DatagramTransport) -> None:
        self.transport = transport

    def datagram_received(self, data: bytes, addr: str) -> None:
        try:
            packet = DHCPPacket.from_bytes(bytes(data))
            logging.debug(f'received dhcp packet:')
            logging.debug(packet)
            logging.debug(packet.view_packet())

        except Exception as e:
            logging.error(e)
            raise e


class DHCPServer:

    def __init__(self, interface: bytes) -> Future:
        self.loop = asyncio.get_running_loop()
        self.interface = interface
        self.awaiter = asyncio.Future()

    async def start(self) -> None:
        self.transport, protocol = await self.loop.create_datagram_endpoint(
            lambda: DHCPProtocol(),
            local_addr=('0.0.0.0', 67),
            allow_broadcast=True)

        self.transport.get_extra_info('socket').setsockopt(
            socket.SOL_SOCKET, 25, self.interface)
        logging.info('Starting DHCP service')
        return await self.awaiter

    async def stop(self) -> None:
        logging.info('Stopping DHCP service')
        self.transport.close()
        self.awaiter.set_result(None)


async def main() -> None:
    logging.basicConfig(level=logging.DEBUG, stream=sys.stdout)

    # change the interface name according to your network interface name
    interface = b'eth0'

    server = DHCPServer(interface=interface)
    loop = asyncio.get_running_loop()
    for sig in (signal.SIGINT, signal.SIGTERM):
        loop.add_signal_handler(sig,
                                lambda: asyncio.create_task(server.stop()))
    await server.start()


if __name__ == '__main__':
    asyncio.run(main())

It seems the error is on options.ClientIdentifier class.

Any help would be appreciated. Thanks,

pujianto avatar May 14 '22 05:05 pujianto