lvfrazao / dhcppython Goto Github PK

        # R
        # dnsmasq wants option field DHCP Requested Address filled in
        option_list = option_list if option_list else options.OptionList()
        option_list.insert(0, options.options.short_value_to_object(50, offer.yiaddr.packed))
        request = packet.DHCPPacket.Request(
            mac_addr,
            int(default_timer() - start),
            tx_id,
            use_broadcast=broadcast,
            option_list=option_list,
            #client_ip=offer.yiaddr,
            relay=relay,
        )

    def __del__(self):
        for sock in self.listening_sockets:
            sock.close()
        for sock in self.writing_sockets:
            sock.close()
        for sock in self.except_sockets:
            sock.close()

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

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