Wala shell

Advanced PHP File Manager

Server IP: 127.0.0.1 | PHP Version: 8.3.30 | Domains Hosted: 54

Name	Size	Last Modified	Actions
📁 ..	-	2025-05-25 12:26
📄 __init__.py ''' DHCP protocol ============= The DHCP implementation here is far from complete, but already provides some basic functionality. Later it will be extended with IPv6 support and more DHCP options will be added. Right now it can be interesting mostly to developers, but not users and/or system administrators. So, the development hints first. The packet structure description is intentionally implemented as for netlink packets. Later these two parsers, netlink and generic, can be merged, so the syntax is more or less compatible. Packet fields ------------- There are two big groups of items within any DHCP packet. First, there are BOOTP/DHCP packet fields, they're defined with the `fields` attribute:: class dhcp4msg(msg): fields = ((name, format, policy), (name, format, policy), ... (name, format, policy)) The `name` can be any literal. Format should be specified as for the struct module, like `B` for `uint8`, or `i` for `int32`, or `>Q` for big-endian uint64. There are also aliases defined, so one can write `uint8` or `be16`, or like that. Possible aliases can be seen in the `pyroute2.protocols` module. The `policy` is a bit complicated. It can be a number or literal, and it will mean that it is a default value, that should be encoded if no other value is given. But when the `policy` is a dictionary, it can contain keys as follows:: 'l2addr': {'format': '6B', 'decode': ..., 'encode': ...} Keys `encode` and `decode` should contain filters to be used in decoding and encoding procedures. The encoding filter should accept the value from user's definition and should return a value that can be packed using `format`. The decoding filter should accept a value, decoded according to `format`, and should return value that can be used by a user. The `struct` module can not decode IP addresses etc, so they should be decoded as `4s`, e.g. Further transformation from 4 bytes string to a string like '10.0.0.1' performs the filter. DHCP options ------------ DHCP options are described in a similar way:: options = ((code, name, format), (code, name, format), ... (code, name, format)) Code is a `uint8` value, name can be any string literal. Format is a string, that must have a corresponding class, inherited from `pyroute2.dhcp.option`. One can find these classes in `pyroute2.dhcp` (more generic) or in `pyroute2.dhcp.dhcp4msg` (IPv4-specific). The option class must reside within dhcp message class. Every option class can be decoded in two ways. If it has fixed width fields, it can be decoded with ordinary `msg` routines, and in this case it can look like that:: class client_id(option): fields = (('type', 'uint8'), ('key', 'l2addr')) If it must be decoded by some custom rules, one can define the policy just like for the fields above:: class array8(option): policy = {'format': 'string', 'encode': lambda x: array('B', x).tobytes(), 'decode': lambda x: array('B', x).tolist()} In the corresponding modules, like in `pyroute2.dhcp.dhcp4msg`, one can define as many custom DHCP options, as one need. Just be sure, that they are compatible with the DHCP server and all fit into 1..254 (`uint8`) -- the 0 code is used for padding and the code 255 is the end of options code. ''' import logging import struct from array import array from typing import ClassVar, NamedTuple, Optional, TypedDict, TypeVar, Union from pyroute2.common import basestring from pyroute2.protocols import Policy, _decode_mac, _encode_mac, msg from .enums.dhcp import MessageType, Option LOG = logging.getLogger(__name__) # hack because Self is not supported in py39 _dhcpmsgSelf = TypeVar('_dhcpmsgSelf', bound='dhcpmsg') _optionSelf = TypeVar('_optionSelf', bound='option') class ClientId(TypedDict): '''A dict with 'type' and 'key' keys. The types stores the client id type, and the key stores the value. See RFC for their meaning. ''' type: int key: Union[bytes, str] def _decode_client_id(value: bytes) -> ClientId: '''Decode a raw client id option into a dict with type and key. If the type is 1, the key is decoded as a mac address, otherwise it's just the raw bytes. ''' type_ = value[0] key: Union[bytes, str] = value[1:] if type_ == 1: # ethernet assert isinstance(key, bytes) key = _decode_mac(value=key) return ClientId(type=type_, key=key) def _encode_client_id(value: ClientId) -> bytes: '''Encode a client_id dict into bytes.''' type_ = value['type'] key = value['key'] if type_ == 1: assert isinstance(key, str), 'client_id must be a mac str when type=1' key = bytes(_encode_mac(key)) assert isinstance(key, bytes), 'client_id must be bytes' return struct.pack('B', type_) + key class option(msg): policy: ClassVar[Optional[Policy]] = None def __init__(self, content=None, buf=b'', offset=0, value=None, code=0): msg.__init__( self, content=content, buf=buf, offset=offset, value=value ) self.code = code @property def length(self) -> Optional[int]: if self.data_length is None: return None if self.data_length == 0: return 1 else: return self.data_length + 2 def encode(self: _optionSelf) -> _optionSelf: # pack code self.buf += struct.pack('B', self.code) if self.code in (0, 255): return self # save buf save = self.buf self.buf = b'' # pack data into the new buf if self.policy is not None: value = self.policy.encode(self.value) if self.policy.format == 'string': fmt = '%is' % len(value) else: fmt = self.policy.format if isinstance(value, str): value = value.encode('utf-8') self.buf = struct.pack(fmt, value) else: msg.encode(self) # get the length data = self.buf self.buf = save self.buf += struct.pack('B', len(data)) # attach the packed data self.buf += data return self def decode(self: _optionSelf) -> _optionSelf: try: self.data_length = struct.unpack( 'B', self.buf[self.offset + 1 : self.offset + 2] )[0] except struct.error as err: raise ValueError( f'Cannot decode length for DHCP option {self.code}: {err}' ) if self.policy is not None: if self.policy.format == 'string': fmt = '%is' % self.data_length else: fmt = self.policy.format try: value = struct.unpack( fmt, self.buf[ self.offset + 2 : self.offset + 2 + self.data_length ], ) except struct.error as err: raise ValueError( f'Cannot decode option {self.code} ' f'as {self.policy.format}: {err}' ) if len(value) == 1: value = value[0] value = self.policy.decode(value) if ( isinstance(value, basestring) and self.policy.format == 'string' ): value = value.lstrip(b"\x00") self.value = value else: # remember current offset as msg.decode() will advance it offset = self.offset # move past the code and option length bytes so that msg.decode() # starts parsing at the right location self.offset += 2 msg.decode(self) # restore offset so that dhcpmsg.decode() advances it correctly self.offset = offset return self class CodeMapping(NamedTuple): name: str code: int format: str class dhcpmsg(msg): options: ClassVar[tuple[tuple[Option, str], ...]] = () def __init__(self, content=None, buf=b'', offset=0, value=None) -> None: super().__init__(content, buf, offset, value) self._encode_map: dict[str, CodeMapping] = {} self._decode_map: dict[int, CodeMapping] = {} self._register_options() def _register_options(self) -> None: for code, fmt in self.options: name = code.name.lower() self._decode_map[code] = self._encode_map[name] = CodeMapping( name=name, code=code, format=fmt ) def decode(self: _dhcpmsgSelf) -> _dhcpmsgSelf: msg.decode(self) self['options'] = {} while self.offset < len(self.buf): code: int = struct.unpack( 'B', self.buf[self.offset : self.offset + 1] )[0] if code == Option.PAD: self.offset += 1 continue if code == Option.END: return self if code in self._decode_map: # use the known decoded & name option_class = getattr(self, self._decode_map[code].format) optname = self._decode_map[code].name else: # code is unknown # if we know this option number, get the value as a list # even if we don't know how to parse it option_class = self.array8 try: # if we know this option name, use it code = Option(code) optname = code.name.lower() except ValueError: optname = f"option{code}" option = option_class(buf=self.buf, offset=self.offset, code=code) try: option.decode() except ValueError as err: # FIXME: maybe we would like the raw option data # if we can't decode it ? but that would complicate typing LOG.error("%s", err) break self.offset += option.length if option.value is not None: value = option.value else: value = option self['options'][optname] = value return self def encode(self: _dhcpmsgSelf) -> _dhcpmsgSelf: msg.encode(self) # put message type options = self.get('options') or {'message_type': MessageType.DISCOVER} self.buf += ( self.uint8(code=Option.MESSAGE_TYPE, value=options['message_type']) .encode() .buf ) for name, value in options.items(): if name == 'message_type': continue if (code_mapping := self._encode_map.get(name)) is None: continue # name is known, ok option_class = getattr(self, code_mapping.format) if isinstance(value, dict): option = option_class(value, code=code_mapping.code) else: option = option_class(code=code_mapping.code, value=value) self.buf += option.encode().buf self.buf += self.none(code=Option.END).encode().buf return self class none(option): pass class be16(option): policy = Policy(format='>H') class be32(option): policy = Policy(format='>I') class sbe32(option): policy = Policy(format='>i') class uint8(option): policy = Policy(format='B') class string(option): policy = Policy(format='string') class array8(option): policy = Policy( format='string', encode=lambda x: array('B', x).tobytes(), decode=lambda x: array('B', x).tolist(), ) class client_id(option): policy = Policy( format='string', encode=_encode_client_id, decode=_decode_client_id ) def __init__( self, content=None, buf=b'', offset=0, value=None, code=0 ): # FIXME: we have to override value w/ content here, because # when trying to encode an option, if it's a dict, its value is # set to None. super().__init__(content, buf, offset, value=content, code=code) class message_type(option): policy = Policy(format='B', decode=MessageType) class bool(option): policy = Policy(format='B', encode=int, decode=bool)	12.37 KB	2025-05-25 12:26
📁 __pycache__	-	2025-05-25 12:26
📄 cli.py import asyncio import logging from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser from importlib import import_module from typing import Any, Optional from pyroute2.dhcp.client import AsyncDHCPClient, ClientConfig from pyroute2.dhcp.fsm import State from pyroute2.dhcp.hooks import Hook from pyroute2.dhcp.iface_status import InterfaceNotFound, InterfaceStateWatcher from pyroute2.dhcp.leases import Lease LOG = logging.getLogger(__name__) def import_dotted_name(name: str) -> Any: '''Import anything by name. Return None if the import wasn't successful.''' try: module_name, obj_name = name.rsplit('.', 1) module = import_module(module_name) return getattr(module, obj_name) except (ValueError, ImportError, AttributeError): return None def get_psr() -> ArgumentParser: psr = ArgumentParser( description='A DHCP client based on pyroute2. ' 'Tries to obtain & keep a lease on an interface, running ' 'configurable hooks to assign the obtained IP address and gw to it.', epilog='Send a SIGUSR1 to renew the current lease, SIGUSR2 to rebind ' 'it and SIGHUP to reset & get a new lease.', formatter_class=ArgumentDefaultsHelpFormatter, ) psr.add_argument( 'interface', help='The interface to request an address for.' ) psr.add_argument( '--lease-type', help='Class to use for leases. ' 'Must be a subclass of `pyroute2.dhcp.leases.Lease`.', type=str, default='pyroute2.dhcp.leases.JSONFileLease', metavar='dotted.name', ) psr.add_argument( '--hook', help='Hooks to load. ' 'These are used to run async python code when, ' 'for example, renewing or expiring a lease. ' 'Defaults to adding & removing ip & gateway.', action='append', type=str, metavar='dotted.name', ) psr.add_argument( '--disable-hooks', help='Disable all hooks.', default=False, action='store_true', ) psr.add_argument( '-x', '--exit-on-timeout', metavar='N', help='Wait for max N seconds for a lease, ' 'exit if none could be obtained.', type=float, ) psr.add_argument( '--log-level', help='Logging level to use.', choices=('DEBUG', 'INFO', 'WARNING', 'ERROR'), default='INFO', ) psr.add_argument( '-p', '--write-pidfile', default=False, action='store_true', help='Write a pid file in the working directory. ', ) psr.add_argument( '-R', '--no-release', default=False, action='store_true', help='Do not send a DHCPRELEASE on exit.', ) # TODO: add options for parameters, retransmission, timeouts... return psr async def run_client( cfg: ClientConfig, exit_timeout: Optional[float] = None ) -> None: '''Run the client until interrupted, or a timeout occurs. The optional `exit_timeout` controls 2 things when provided: - How long to wait for the interface to be up - How long to wait for the client to be bound when starting up ''' acli = AsyncDHCPClient(cfg) async with InterfaceStateWatcher(cfg.interface) as iface_watcher: while True: # Open the socket, read existing lease, etc if iface_watcher.state != 'up': LOG.info('Waiting for %s to go up...', cfg.interface) await asyncio.wait_for( iface_watcher.up.wait(), timeout=exit_timeout ) async with acli: # Bootstrap the client by sending a DISCOVER or a REQUEST await acli.bootstrap() if exit_timeout: # Wait a bit for a lease, and raise if we have none await acli.wait_for_state( State.BOUND, timeout=exit_timeout ) break await iface_watcher.down.wait() LOG.warning('%s went down', cfg.interface) async def main() -> None: psr = get_psr() args = psr.parse_args() logging.basicConfig( format='%(asctime)s %(levelname)s [%(name)s:%(funcName)s] %(message)s' ) logging.getLogger('pyroute2.dhcp').setLevel(args.log_level) LOG.setLevel(args.log_level) # parse lease type lease_type = import_dotted_name(args.lease_type) if not (isinstance(lease_type, type) and issubclass(lease_type, Lease)): psr.error(f'{args.lease_type!r} must point to a Lease subclass.') # parse hooks hooks: list[Hook] = [] if not args.disable_hooks: if not args.hook: args.hook = [ 'pyroute2.dhcp.hooks.configure_ip', 'pyroute2.dhcp.hooks.add_default_gw', 'pyroute2.dhcp.hooks.remove_default_gw', 'pyroute2.dhcp.hooks.remove_ip', ] LOG.debug('Configured hooks:') for dotted_hook_name in args.hook: hook = import_dotted_name(dotted_hook_name) if not isinstance(hook, Hook): psr.error(f'{dotted_hook_name!r} must point to a valid hook.') hooks.append(hook) LOG.debug("- %s", hook.name) # Create configuration cfg = ClientConfig( interface=args.interface, lease_type=lease_type, hooks=hooks, write_pidfile=args.write_pidfile, release=not args.no_release, handle_signals=True, ) try: await run_client(cfg, exit_timeout=args.exit_on_timeout) except InterfaceNotFound as err: psr.error(f"Interface not found: {err}") def run(): # for the setup.cfg entrypoint try: asyncio.run(main()) except KeyboardInterrupt: pass if __name__ == '__main__': # pragma: no cover run()	5.8 KB	2025-05-25 12:26
📄 client.py import asyncio import errno import os import random from dataclasses import dataclass, field from logging import getLogger from math import floor from pathlib import Path from signal import Signals from socket import gethostname from time import time from typing import Callable, DefaultDict, Iterable, Iterator, Optional, Union from pyroute2.dhcp import fsm, messages from pyroute2.dhcp.dhcp4socket import AsyncDHCP4Socket from pyroute2.dhcp.enums import dhcp from pyroute2.dhcp.hooks import Hook, Trigger, run_hooks from pyroute2.dhcp.leases import JSONFileLease, Lease from pyroute2.dhcp.timers import LeaseTimers from pyroute2.dhcp.xids import Xid LOG = getLogger(__name__) # TODO: maybe move retransmission stuff to its own file Retransmission = Callable[[], Union[Iterator[int], Iterator[float]]] def randomized_increasing_backoff( wait_first: float = 4.0, wait_max: float = 32.0, factor: float = 2.0 ) -> Iterator[float]: '''Yields seconds to wait until the next retry, forever.''' delay = wait_first while True: yield delay if delay <= wait_max: delay = min(random.uniform(delay, delay * factor), wait_max) @dataclass class ClientConfig: '''Stores configuration option for the DHCP client.''' # FIXME: this is probably not enough if we want the client to work with # WLAN interfaces, as we would have to handle the SSID somehow ? # The interface to bind to and obtain a lease for. interface: str # The lease type to use, allows flexibility on where and how to store it. lease_type: type[Lease] = JSONFileLease # A list of hooks that will be called when the required triggers are met. hooks: Iterable[Hook] = () # The DHCP parameters requested by the client. requested_parameters: Iterable[dhcp.Option] = ( dhcp.Option.SUBNET_MASK, dhcp.Option.ROUTER, dhcp.Option.NAME_SERVER, dhcp.Option.DOMAIN_NAME, dhcp.Option.LEASE_TIME, dhcp.Option.RENEWAL_TIME, dhcp.Option.REBINDING_TIME, ) # Timeouts for various client states. # If the client reaches the specified amount of seconds in one of the # configured states, it will reset the lease process and start anew. timeouts: dict[fsm.State, int] = field( default_factory=lambda: { # No use staying too long in REBOOTING state if nobody is answering fsm.State.REBOOTING: 10, # When we get an OFFER, how long should we wait for an ACK ? fsm.State.REQUESTING: 30, } ) # FIXME: we send too many retries according to the RFC # In both RENEWING and REBINDING states, if the client receives no # response to its DHCPREQUEST message, the client SHOULD wait one-half # of the remaining time until T2 (in RENEWING state) and one-half of # the remaining lease time (in REBINDING state), down to a minimum of # 60 seconds, before retransmitting the DHCPREQUEST message. retransmission: Retransmission = randomized_increasing_backoff # Whether to write a pidfile in the working directory write_pidfile: bool = False # Send a DHCPRELEASE on client exit release: bool = True # Maximum execution duration for a single hook hook_timeout: Optional[float] = 2.0 # Custom client id. The mac is used as the client id if not provided. client_id: Optional[bytes] = None # Whether to handle USR1 (renew), USR2 (rebind) and HUP (reset) handle_signals: bool = True # optional vendor_id & hostname options included in requests vendor_id: Optional[str] = 'pyroute2' host_name: Optional[str] = field(default_factory=gethostname) @property def pidfile_path(self) -> Path: '''Where to write the pid file. It's named after the interface.''' return ( Path.cwd() .joinpath(self.interface) .with_name(f"{self.interface}.pid") ) class AsyncDHCPClient: '''A DHCP client based on pyroute2. Implemented as an async context manager running a finite state machine. The client will try to acquire and keep a lease as long as it's running. It can run hooks specified in its configuration to perform various actions such as adding an IP address to an interface, configuring routing, when the specified events are triggered (see the `hook()` decorator.) Exiting the context manager causes the lease to be released and relevant hooks to be run. Example usage:: from pyroute2.dhcp.client import AsyncDHCPClient, ClientConfig from pyroute2.dhcp.fsm import State cfg = ClientConfig(interface='eth3') async with AsyncDHCPClient(cfg) as client: # Bootstrap the client by sending a DISCOVER or a REQUEST await client.bootstrap() # Wait 20s until the client gets a lease await client.wait_for_state(State.BOUND, timeout=20.0) ''' def __init__(self, config: ClientConfig) -> None: self.config = config # The raw socket used to send and receive packets self._sock: AsyncDHCP4Socket = AsyncDHCP4Socket(self.config.interface) # Current client state self._state: fsm.State = fsm.State.OFF # Current lease, read from persistent storage or sent by a server self._lease: Optional[Lease] = None # dhcp messages put in this queue are sent by _send_forever self._sendq: asyncio.Queue[Optional[messages.SentDHCPMessage]] = ( asyncio.Queue() ) # Handle to run _send_forever for the context manager's lifetime self._sender_task: Optional[asyncio.Task] = None # Handle to run _recv_forever for the context manager's lifetime self._receiver_task: Optional[asyncio.Task] = None # Timers to run callbacks on lease timeouts expiration self.lease_timers = LeaseTimers() # Calls reset() after a timeout in some states to avoid getting stuck self._state_watchdog: asyncio.Task \| None = None # Allows to easily track the state when running the client from python self._states: DefaultDict[fsm.State, asyncio.Event] = DefaultDict( asyncio.Event ) # Used to compute lease times, taking into account the request time self.last_state_change: float = time() self._xid: Optional[Xid] = None # 'public api' async def wait_for_state( self, state: fsm.State, timeout: Optional[float] = None ) -> None: '''Waits until the client is in the target state.''' try: await asyncio.wait_for(self._states[state].wait(), timeout=timeout) except asyncio.exceptions.TimeoutError as err: raise asyncio.exceptions.TimeoutError( f'Timed out waiting for the {state.name} state. ' f'Current state: {self.state.name}' ) from err @fsm.state_guard(fsm.State.INIT, fsm.State.INIT_REBOOT) async def bootstrap(self) -> None: '''Send a `DISCOVER` or a `REQUEST`, depending on whether we're initializing or rebooting. Use this to get a lease when running the client from Python code. ''' if self.state is fsm.State.INIT: # send discover await self.transition( to=fsm.State.SELECTING, send=messages.discover( parameter_list=self.config.requested_parameters ), ) elif self.state is fsm.State.INIT_REBOOT: assert self.lease, 'cannot init_reboot without a lease' # send request for lease await self.transition( to=fsm.State.REBOOTING, send=messages.request_for_lease( parameter_list=self.config.requested_parameters, lease=self.lease, state=fsm.State.REBOOTING, ), ) # the decorator prevents the needs for an else # properties @property def lease(self) -> Optional[Lease]: '''The current lease, if we have one.''' return self._lease @lease.setter def lease(self, value: Lease) -> None: '''Set a fresh lease; only call this when a server grants one.''' self._lease = value self.lease_timers.arm( lease=self._lease, renewal=self._renew, rebinding=self._rebind, expiration=self._expire_lease, ) # whatever error might happen when writing a lease, # it should never make the client crash try: self._lease.dump() except Exception as exc: LOG.error('Could not dump lease: %s', exc) @property def state(self) -> fsm.State: '''The current client state.''' return self._state @state.setter def state(self, value: fsm.State) -> None: '''Check the client can transition to the state, and set it. Only supposed to be called by the client internally. ''' old_state = self.state if value and value not in fsm.TRANSITIONS[old_state]: raise ValueError( f'Cannot transition from {self._state.name} to {value.name}' ) LOG.info('%s -> %s', old_state.name, value.name) if self._state_watchdog: self._state_watchdog.cancel() self._state_watchdog = None if old_state in self._states: self._states[old_state].clear() self._state = value self.last_state_change = time() self._states[value].set() if state_timeout := self.config.timeouts.get(value): self._state_watchdog = asyncio.Task( self.reset(delay=state_timeout) ) @property def xid(self) -> Xid: '''The xid currently in use by the client. Only available when the context manager is running. ''' if self._xid is None: raise AttributeError('xid is not defined') return self._xid # Timer callbacks @fsm.state_guard(fsm.State.BOUND) async def _renew(self, signal: Optional[Signals] = None) -> None: '''Called when the renewal time defined in the lease expires.''' assert self.lease, 'cannot renew without an existing lease' LOG.info( '%s, renewing lease', f'{signal.name} received' if signal else 'T1 expired', ) self.lease_timers._reset_timer('renewal') # FIXME should be automatic await self.transition( to=fsm.State.RENEWING, send=messages.request_for_lease( parameter_list=self.config.requested_parameters, lease=self.lease, state=fsm.State.RENEWING, ), ) @fsm.state_guard(fsm.State.RENEWING, fsm.State.BOUND) async def _rebind(self, signal: Optional[Signals] = None) -> None: '''Called when the rebinding time defined in the lease expires.''' assert self.lease, 'cannot rebind without an existing lease' LOG.info( '%s, rebinding lease', f'{signal.name} received' if signal else 'T2 expired', ) # if the user asks for a rebind before the renewal timer expires, # better cancel the renewal timer too self.lease_timers._reset_timer('renewal') self.lease_timers._reset_timer('rebinding') # FIXME await self.transition( to=fsm.State.REBINDING, send=messages.request_for_lease( parameter_list=self.config.requested_parameters, lease=self.lease, state=fsm.State.REBINDING, ), ) async def _expire_lease(self) -> None: '''Called when the expiration time defined in the lease expires.''' LOG.info('Lease expired') self.lease_timers._reset_timer('expiration') await self._run_hooks(Trigger.EXPIRED) await self.reset() # DHCP packet sending & receving coroutines async def _send_message(self, msg: messages.SentDHCPMessage) -> None: '''Set secs, xid & client id on the message, and send it.''' # Set secs to the time elapsed since the last state change # (max 16 bits) msg.dhcp['secs'] = min(floor(time() - self.last_state_change), 0xFFFF) msg.dhcp['xid'] = self.xid.for_state(self.state) # set the client id, rfc says: # A hardware type of 0 (zero) should be used when the value field # contains an identifier other than a hardware address (e.g. a fully # qualified domain name). if self.config.client_id: client_id = {'type': 0, 'key': self.config.client_id} else: # default behavior, use hw type & hw addr as client id client_id = {'type': 1, 'key': self._sock.l2addr} msg.dhcp['options']['client_id'] = client_id # set hostname & vendor id if configured to do so if self.config.host_name: msg.dhcp['options']['host_name'] = self.config.host_name if self.config.vendor_id: msg.dhcp['options']['vendor_id'] = self.config.vendor_id LOG.info('Sending %s', msg) await self._sock.put(msg) async def _send_forever(self) -> None: '''Send packets from `_sendq` until the client is in `State.OFF`.''' msg_to_send: Optional[messages.SentDHCPMessage] = None # Called to get the interval value below interval_factory: Optional[Union[Iterator[int], Iterator[float]]] = ( None ) # How long to sleep before retrying interval: Union[int, float] = 1 # this is triggered by __aexit__ wait_til_off = asyncio.Task(self.wait_for_state(fsm.State.OFF)) while not (wait_til_off.done() and self._sendq.empty()): # FIXME: interval handling is pretty awkward and convoluted if interval_factory: interval = next(interval_factory) else: interval = 9999999 wait_for_msg_to_send = asyncio.Task( self._sendq.get(), name='wait for packet to send' ) if msg_to_send: LOG.debug('%.1f seconds until retransmission', interval) done, pending = await asyncio.wait( (wait_til_off, wait_for_msg_to_send), return_when=asyncio.FIRST_COMPLETED, timeout=interval, ) if wait_for_msg_to_send in done: if msg_to_send := wait_for_msg_to_send.result(): # There is a new message to send, reset the interval interval_factory = self.config.retransmission() else: # No need to retry anything interval_factory = None elif wait_for_msg_to_send in pending: wait_for_msg_to_send.cancel() if msg_to_send: if ( msg_to_send.message_type != dhcp.MessageType.RELEASE and wait_til_off.done() ): LOG.debug( 'Not sending %s, client is shutting down', msg_to_send.message_type.name, ) continue try: await self._send_message(msg_to_send) except OSError as err: # That happens when the interface goes down. # In theses cases, the client is supposed to be restarted if err.errno == errno.ENETDOWN: LOG.error('Could not send, network is down') return raise async def _recv_forever(self) -> None: '''Receive & process DHCP messages until the client stops.''' # TODO: is there a better way to wait for the client to stop ? wait_til_off = asyncio.Task(self.wait_for_state(fsm.State.OFF)) while not wait_til_off.done(): wait_for_received_msg = asyncio.Task( coro=self._sock.get(), name=f'wait for DHCP messages on {self.config.interface}', ) # sleep until a new message is received, or the client is OFF done, pending = await asyncio.wait( (wait_til_off, wait_for_received_msg), return_when=asyncio.FIRST_COMPLETED, ) if wait_for_received_msg in done: try: received_msg = wait_for_received_msg.result() except OSError as err: # That happens when the interface goes down. # In theses cases, the client is supposed to be restarted if err.errno == errno.ENETDOWN: LOG.error('Could not recv, network is down') return raise await self._process_msg(received_msg) elif wait_for_received_msg in pending: wait_for_received_msg.cancel() # Callbacks for received DHCP messages @fsm.state_guard( fsm.State.REQUESTING, fsm.State.REBOOTING, fsm.State.REBINDING, fsm.State.RENEWING, ) async def ack_received(self, msg: messages.ReceivedDHCPMessage) -> None: '''Called when an ACK is received. Stores the lease and puts the client in the BOUND state. ''' # FIXME: according to the RFC: # When the client receives a DHCPACK from the server, the client # computes the lease expiration time as the sum of the time at which # the client sent the DHCPREQUEST message and the duration of the lease # in the DHCPACK message. # The state the client was in when sending the message request_state = msg.xid.request_state if request_state in (fsm.State.REQUESTING, fsm.State.REBOOTING): trigger = Trigger.BOUND elif request_state == fsm.State.RENEWING: trigger = Trigger.RENEWED elif request_state == fsm.State.REBINDING: trigger = Trigger.REBOUND else: LOG.warning('Invalid request state for xid %s', msg.xid) return if 'lease_time' not in msg.dhcp['options']: # that would not make sense at all, but still... LOG.warning('Server did not define a lease time, ignoring ACK.') return self.lease = self.config.lease_type( ack=msg.dhcp, interface=self.config.interface, server_mac=msg.eth_src, ) LOG.info( 'Got lease for %s from %s (%s)', self.lease.ip, self.lease.server_id, msg.eth_src, ) await self.transition(to=fsm.State.BOUND) await self._run_hooks(trigger) @fsm.state_guard( fsm.State.REQUESTING, fsm.State.REBOOTING, fsm.State.RENEWING, fsm.State.REBINDING, ) async def nak_received(self, msg: messages.ReceivedDHCPMessage) -> None: '''Called when a NAK is received. Resets the client and starts looking for a new IP. ''' await self.reset() @fsm.state_guard(fsm.State.SELECTING) async def offer_received(self, msg: messages.ReceivedDHCPMessage) -> None: '''Called when an OFFER is received. Sends a REQUEST for the offered IP address. ''' # FIXME: we should probably validate the offer some more, # like checking if the offered options are satisfying await self.transition( to=fsm.State.REQUESTING, send=messages.request_for_offer( parameter_list=self.config.requested_parameters, offer=msg ), ) # Async context manager methods async def __aenter__(self) -> 'AsyncDHCPClient': '''Set up the client so it's ready to obtain an IP. Tries to load a lease for the client's interface, opens the socket, starts the sender & receiver tasks and allocates a request ID. ''' if self.config.handle_signals: self._register_signal_handlers() self._xid = Xid() if self.config.write_pidfile: self.config.pidfile_path.write_text(str(os.getpid())) LOG.debug('Wrote pidfile to %s', self.config.pidfile_path) if loaded_lease := self.config.lease_type.load(self.config.interface): self._lease = loaded_lease self.state = fsm.State.INIT_REBOOT else: LOG.debug('No current lease') self.state = fsm.State.INIT await self._sock.__aenter__() self._receiver_task = asyncio.Task( self._recv_forever(), name=f'Listen for DHCP packets on {self.config.interface}', ) self._sender_task = asyncio.Task( self._send_forever(), name=f'Send outgoing DHCP packets on {self.config.interface}', ) return self async def __aexit__(self, *_) -> None: '''Shut down the client. If there's an active lease, send a RELEASE for it first. ''' if self.config.handle_signals: self._remove_signal_handlers() self.lease_timers.cancel() if self.lease and self.config.release: await self._run_hooks(Trigger.UNBOUND) if not self.lease.expired: await self._sendq.put(messages.release(lease=self.lease)) self.state = fsm.State.OFF if self._sender_task: await self._sender_task if self._receiver_task: await self._receiver_task await self._sock.__aexit__() self._xid = None if self.config.write_pidfile: self.config.pidfile_path.unlink(missing_ok=True) LOG.debug('Removed pidfile at %s', self.config.pidfile_path) # internal methods async def transition( self, to: fsm.State, send: Optional[messages.SentDHCPMessage] = None ) -> None: '''Change the client's state, and start sending a message repeatedly. If the message is None, any current message will stop being sent. ''' self.state = to await self._sendq.put(send) async def reset( self, delay: float = 0.0, signal: Optional[Signals] = None ) -> None: '''Called internally to restart the lease acquisition process. - When the client receives a NAK; - When it spends too much time in a configured state (see `ClientConfig.timeouts`) - When the current lease expires. - Upon receiving a SIGHUP Erases the lease, cancel timers, resets the xid, and sends DISCOVERs to get a new lease. ''' if delay: await asyncio.sleep(delay=delay) LOG.warning('Resetting after %.1f seconds', delay) if signal: LOG.info('%s received, resetting', signal.name) await self.transition(to=fsm.State.INIT) # Reset lease & timers and start again self._lease = None self.lease_timers.cancel() self._xid = Xid() await self.bootstrap() async def _run_hooks(self, trigger: Trigger) -> None: '''Run hooks for the given trigger. Hooks are awaited and not run in the background; that means they could cause issues if they run for too long. Hence, each hook is limited to the `hook_timeout` config setting. ''' assert self.lease, 'tried to run hooks without a lease' await run_hooks( hooks=self.config.hooks, lease=self.lease, trigger=trigger, timeout=self.config.hook_timeout, ) async def _process_msg(self, msg: messages.ReceivedDHCPMessage) -> None: '''Processes a received DHCP message. The messages's xid is checked against the client's xid, and the appropriate async handler (`self.<msg type>_received`) is then called if it exists. ''' msg_type = dhcp.MessageType(msg.dhcp['options']['message_type']) LOG.info('Received %s', msg) if not self.xid.matches(msg.xid): LOG.error( 'Incorrect xid %s (expected %sX), discarding', msg.xid, hex(self.xid.random_part)[:-1], ) return handler_name = f'{msg_type.name.lower()}_received' handler = getattr(self, handler_name, None) if not handler: LOG.debug('DHCP %s messages are not handled', msg_type.name) else: await handler(msg) def _add_signal_handler( self, loop: asyncio.AbstractEventLoop, signal: Signals, callback: fsm.AsyncCallback, ): '''Arranges for `callback` to be called when receiving `signal`. `callback` must take a `signal` keyword argument. ''' loop.add_signal_handler( sig=signal, callback=lambda: asyncio.create_task(callback(signal=signal)), ) def _register_signal_handlers(self): '''Add signal handlers to catch USR1, USR2 and HUP. Called by the context manager when `handle_signals` is True. - SIGUSR1 causes a renew, - SIGUSR2 causes a rebind, - SIGHUP causes a reset. ''' LOG.info('Registering signal handlers for USR1, USR2 and HUP') loop = asyncio.get_running_loop() self._add_signal_handler(loop, Signals.SIGUSR1, self._renew) self._add_signal_handler(loop, Signals.SIGUSR2, self._rebind) self._add_signal_handler(loop, Signals.SIGHUP, self.reset) def _remove_signal_handlers(self): remove_handler = asyncio.get_running_loop().remove_signal_handler LOG.info('Removing signal handlers for USR1, USR2 and HUP') remove_handler(Signals.SIGUSR1) remove_handler(Signals.SIGUSR2) remove_handler(Signals.SIGHUP)	25.62 KB	2025-05-25 12:26
📄 dhcp4msg.py from socket import AF_INET, inet_ntop, inet_pton from pyroute2.dhcp import Policy, dhcpmsg, option from pyroute2.dhcp.enums.dhcp import Option class dhcp4msg(dhcpmsg): # # https://www.ietf.org/rfc/rfc2131.txt # fields = ( ('op', 'uint8', 1), # request ('htype', 'uint8', 1), # ethernet ('hlen', 'uint8', 6), # ethernet addr len ('hops', 'uint8'), ('xid', 'be32'), ('secs', 'be16'), ('flags', 'be16'), ('ciaddr', 'ip4addr'), ('yiaddr', 'ip4addr'), ('siaddr', 'ip4addr'), ('giaddr', 'ip4addr'), ('chaddr', 'l2paddr'), ('sname', '64s'), ('file', '128s'), ('cookie', '4s', b'c\x82Sc'), ) # # https://www.ietf.org/rfc/rfc2132.txt # options = ( # TODO: add & test more options (Option.PAD, 'none'), (Option.SUBNET_MASK, 'ip4addr'), (Option.TIME_OFFSET, 'be32'), (Option.ROUTER, 'ip4list'), (Option.TIME_SERVER, 'ip4list'), (Option.IEN_NAME_SERVER, 'ip4list'), (Option.NAME_SERVER, 'ip4list'), (Option.LOG_SERVER, 'ip4list'), (Option.COOKIE_SERVER, 'ip4list'), (Option.LPR_SERVER, 'ip4list'), (Option.IMPRESS_SERVER, 'ip4list'), (Option.RESOURCE_LOCATION_SERVER, 'ip4list'), (Option.HOST_NAME, 'string'), # in multiples of 512 bytes (Option.BOOT_FILE_SIZE, 'be16'), (Option.MERIT_DUMP_FILE, 'string'), (Option.DOMAIN_NAME, 'string'), (Option.SWAP_SERVER, 'ip4addr'), (Option.ROOT_PATH, 'string'), (Option.EXTENSIONS_PATH, 'string'), (Option.IP_FORWARDING, 'bool'), (Option.NON_LOCAL_SOURCE_ROUTING, 'bool'), # TODO: dict of ip/mask pairs # (Option.POLICY_FILTER, '?'), # minimum legal value is 567 according to rfc (Option.MAX_DATAGRAM_REASSEMBLY, 'be16'), (Option.DEFAULT_TTL, 'uint8'), (Option.PATH_MTU_AGING_TIMEOUT, 'be32'), # TODO: list of be16 # (Option.PATH_MTU_PLATEAU_TABLE, '?'), (Option.INTERFACE_MTU, 'be16'), (Option.ALL_SUBNETS_LOCAL, 'bool'), (Option.BROADCAST_ADDRESS, 'ip4addr'), (Option.PERFORM_MASK_DISCOVERY, 'bool'), (Option.MASK_SUPPLIER, 'bool'), (Option.PERFORM_ROUTER_DISCOVERY, 'bool'), (Option.ROUTER_SOLICITATION_ADDRESS, 'ip4addr'), # TODO: dict of ip addrs # (Option.STATIC_ROUTE, '?') (Option.TRAILER_ENCAPSULATION, 'bool'), (Option.ARP_CACHE_TIMEOUT, 'be32'), (Option.ETHERNET_ENCAPSULATION, 'bool'), (Option.TCP_DEFAULT_TTL, 'uint8'), (Option.TCP_KEEPALIVE_INTERVAL, 'be32'), (Option.TCP_KEEPALIVE_GARBAGE, 'bool'), (Option.NIS_DOMAIN, 'string'), (Option.NDS_SERVERS, 'ip4list'), (Option.NTP_SERVERS, 'ip4list'), (Option.VENDOR_SPECIFIC_INFORMATION, 'string'), (Option.NETBIOS_NAME_SERVER, 'ip4list'), (Option.NETBIOS_DDG_SERVER, 'ip4list'), # 1=B, 2=P, 4=M, 8=H, probably obsolete & unused (Option.NETBIOS_NODE_TYPE, 'uint8'), (Option.NETBIOS_SCOPE, 'string'), (Option.X_WINDOW_FONT_SERVER, 'ip4list'), (Option.X_WINDOW_DISPLAY_MANAGER, 'ip4list'), (Option.REQUESTED_IP, 'ip4addr'), (Option.LEASE_TIME, 'sbe32'), # 1: options in file, 2: options in sname, 4, both (Option.OPTION_OVERLOAD, 'uint8'), (Option.TFTP_SERVER_NAME, 'string'), (Option.BOOTFILE_NAME, 'string'), (Option.MESSAGE_TYPE, 'message_type'), (Option.SERVER_ID, 'ip4addr'), (Option.PARAMETER_LIST, 'array8'), (Option.MESSAGE, 'string'), # minimum value: 576 bytes (Option.MAX_MSG_SIZE, 'be16'), (Option.RENEWAL_TIME, 'sbe32'), (Option.REBINDING_TIME, 'sbe32'), (Option.VENDOR_ID, 'string'), (Option.CLIENT_ID, 'client_id'), (Option.END, 'none'), ) class ip4addr(option): policy = Policy( format='4s', encode=lambda x: inet_pton(AF_INET, x), decode=lambda x: inet_ntop(AF_INET, x), ) class ip4list(option): policy = Policy( format='string', encode=lambda x: b''.join([inet_pton(AF_INET, i) for i in x]), decode=lambda x: [ inet_ntop(AF_INET, x[i * 4 : i * 4 + 4]) for i in range(len(x) // 4) ], )	4.4 KB	2025-05-25 12:26
📄 dhcp4socket.py ''' IPv4 DHCP socket ================ ''' import asyncio import logging import socket from typing import Optional from pyroute2.compat import ETHERTYPE_IP from pyroute2.dhcp.dhcp4msg import dhcp4msg from pyroute2.dhcp.messages import ReceivedDHCPMessage, SentDHCPMessage from pyroute2.ext.bpf import BPF from pyroute2.ext.rawsocket import AsyncRawSocket from pyroute2.protocols import ethmsg, ip4msg, udp4_pseudo_header, udpmsg LOG = logging.getLogger(__name__) UDP_HEADER_SIZE = 8 IPV4_HEADER_SIZE = 20 SKF_AD_OFF = -0x1000 SKF_AD_VLAN_TAG_PRESENT = 48 def listen_udp_port(port: int = 68) -> list[list[int]]: '''BPF filter that matches Ethernet + IPv4 + UDP on the given port. Packets tagged on a vlan are also dropped, see https://lore.kernel.org/netdev/51FB6A9D.2050002@redhat.com/T/ ''' bpf_code = [ # Load vlan presence indicator [BPF.LD + BPF.B + BPF.ABS, 0, 0, SKF_AD_OFF + SKF_AD_VLAN_TAG_PRESENT], # bail out immediately if there is one and we don't want it [BPF.JMP + BPF.JEQ + BPF.K, 0, 10, 0], # Load eth type at offset 12 and check it's IPv4 [BPF.LD + BPF.H + BPF.ABS, 0, 0, 12], [BPF.JMP + BPF.JEQ + BPF.K, 0, 8, 0x0800], # Load IP proto at offset 23 and check it's UDP [BPF.LD + BPF.B + BPF.ABS, 0, 0, 23], [BPF.JMP + BPF.JEQ + BPF.K, 0, 6, socket.IPPROTO_UDP], # load frag offset at offset 20 [BPF.LD + BPF.H + BPF.ABS, 0, 0, 20], # Check mask & drop fragmented packets [BPF.JMP + BPF.JSET + BPF.K, 4, 0, 8191], # load ip header length at offset 14 [BPF.LDX + BPF.B + BPF.MSH, 0, 0, 14], # load udp dport from that offset + 16 and check it [BPF.LD + BPF.H + BPF.IND, 0, 0, 16], [BPF.JMP + BPF.JEQ + BPF.K, 0, 1, port], # allow packet [BPF.RET + BPF.K, 0, 0, 65535], # drop packet [BPF.RET + BPF.K, 0, 0, 0], ] return bpf_code class AsyncDHCP4Socket(AsyncRawSocket): ''' Parameters: * ifname -- interface name to work on * port -- UDP port to listen on This raw socket binds to an interface and installs a BPF filter to receive (non-VLAN) messages only on the specified UDP port. It can be used in poll/select and implements the async context manager protocol, so can be used in `aysnc with` statements. ''' def __init__(self, ifname: str, port: int = 68): AsyncRawSocket.__init__(self, ifname, listen_udp_port(port)) self.port = port self._loop = asyncio.get_running_loop() @property def loop(self) -> asyncio.AbstractEventLoop: # We define this as a property because it's easier to patch in tests return self._loop async def put(self, msg: SentDHCPMessage) -> SentDHCPMessage: '''Send a DHCP message. This encapsulates the `SentDHCPMessage` into Ethernet, IPv4, and UDP. If not provided, both the Ethernet source and the DHCP chaddr are set to the MAC address of the underlying socket's interface. Example:: msg = SentDHCPMessage( dhcp=dhcp4msg({ 'op': bootp.MessageType.BOOTREQUEST, 'flags': bootp.Flag.BROADCAST, 'options': { 'message_type': dhcp.MessageType.DISCOVER, 'parameter_list': [ dhcp.Option.SUBNET_MASK, dhcp.Option.ROUTER, dhcp.Option.NAME_SERVER, dhcp.Option.LEASE_TIME, ], }, }), ) await sock.put(msg) ''' if msg.sport != self.port: raise ValueError( f"Client source port is set to {self.port}, " f"cannot send message from port {msg.sport}." ) if not msg.eth_src: msg.eth_src = self.l2addr # DHCP layer dhcp = msg.dhcp # auto add src addr if dhcp['chaddr'] is None: dhcp['chaddr'] = msg.eth_src data = dhcp.encode().buf dhcp_payload_size = len(data) # UDP layer udp = udpmsg( { 'sport': self.port, 'dport': msg.dport, 'len': UDP_HEADER_SIZE + dhcp_payload_size, } ) # Pseudo UDP header, only for checksum purposes udph = udp4_pseudo_header( { 'src': msg.ip_src, 'dst': msg.ip_dst, 'len': UDP_HEADER_SIZE + dhcp_payload_size, } ) udp['csum'] = self.csum(udph.encode().buf + udp.encode().buf + data) udp.reset() # IPv4 layer ip4 = ip4msg( { 'len': IPV4_HEADER_SIZE + UDP_HEADER_SIZE + dhcp_payload_size, 'proto': socket.IPPROTO_UDP, 'dst': msg.ip_dst, 'src': msg.ip_src, } ) ip4['csum'] = self.csum(ip4.encode().buf) ip4.reset() # MAC layer eth = ethmsg( {'dst': msg.eth_dst, 'src': msg.eth_src, 'type': ETHERTYPE_IP} ) data = eth.encode().buf + ip4.encode().buf + udp.encode().buf + data await self.loop.sock_sendall(self, data) dhcp.reset() return msg async def get(self) -> ReceivedDHCPMessage: ''' Get the next incoming packet from the socket and try to decode it as IPv4 DHCP. Packets that cannot be decoded are logged & discarded. Invalid/truncated packets raise `ValueError`. Example:: # Send a DISCOVER and read an OFFER disco = messages.discover(parameter_list=[ dhcp.Option.SUBNET_MASK, dhcp.Option.ROUTER, dhcp.Option.NAME_SERVER, dhcp.Option.LEASE_TIME, ]) await sock.put(disco) offer = await sock.get() print('received', offer.message_type, 'from', offer.eth_src, 'lease time', offer.dhcp['options']['lease_time'], ) ''' msg: Optional[ReceivedDHCPMessage] = None while not msg: raw = await self.loop.sock_recv(self, 4096) try: msg = self._decode_msg(raw) except ValueError as err: LOG.error('%s', err) return msg @classmethod def _decode_msg(cls, data: bytes) -> ReceivedDHCPMessage: '''Decode an IPv4 DHCP packet from bytes. No analysis is done here. The MAC/IPv4/UDP headers are stripped out, the relevant values are stored in the `ReceivedDHCPMessage` metadata, and the rest is interpreted as DHCP. ''' eth = ethmsg(buf=data).decode() ip4 = ip4msg(buf=data, offset=eth.offset).decode() udp = udpmsg(buf=data, offset=ip4.offset).decode() dhcp = dhcp4msg(buf=data, offset=udp.offset).decode() return ReceivedDHCPMessage( dhcp=dhcp, eth_src=eth['src'], eth_dst=eth['dst'], ip_src=ip4['src'], ip_dst=ip4['dst'], sport=udp['sport'], dport=udp['dport'], )	7.21 KB	2025-05-25 12:26
📁 enums	-	2025-05-25 12:26
📄 fsm.py '''DHCP client state machine helpers.''' import functools from enum import IntEnum, auto from logging import getLogger from typing import Any, Callable, Final, Protocol LOG = getLogger(__name__) class State(IntEnum): '''DHCP client states. see http://www.tcpipguide.com/free/t_DHCPGeneralOperationandClientFiniteStateMachine.htm ''' OFF = 0 INIT = auto() INIT_REBOOT = auto() REBOOTING = auto() REQUESTING = auto() SELECTING = auto() BOUND = auto() RENEWING = auto() REBINDING = auto() # allowed transitions between states TRANSITIONS: Final[dict[State, set[State]]] = { State.OFF: {State.INIT, State.INIT_REBOOT}, State.INIT_REBOOT: {State.REBOOTING}, State.REBOOTING: {State.INIT, State.BOUND}, State.INIT: {State.SELECTING}, State.SELECTING: {State.REQUESTING, State.INIT}, State.REQUESTING: {State.BOUND, State.INIT}, State.BOUND: {State.INIT, State.RENEWING, State.REBINDING}, State.RENEWING: {State.BOUND, State.INIT, State.REBINDING}, State.REBINDING: {State.BOUND, State.INIT}, } class AsyncCallback(Protocol): __name__: str async def __call__(self, a, kwds: Any) -> Any: '''Protocol for async methods that makes mypy happy.''' def state_guard(states: State) -> Callable[[AsyncCallback], AsyncCallback]: '''Decorator that prevents a method from running if the associated instance is not in one of the given States.''' def decorator(meth: AsyncCallback) -> AsyncCallback: @functools.wraps(meth) async def wrapper(self, args: Any, kwargs: Any) -> None: if self.state not in states: LOG.debug( 'Ignoring call to %r in %s state', meth.__name__, self.state.name, ) return await meth(self, args, **kwargs) return wrapper return decorator	1.9 KB	2025-05-25 12:26
📄 hooks.py '''Hooks called by the DHCP client when bound, a leases expires, etc.''' import asyncio import errno from enum import auto from logging import getLogger from typing import Callable, Iterable, NamedTuple, Optional, Protocol from pyroute2.compat import StrEnum from pyroute2.dhcp.leases import Lease, MissingOptionError from pyroute2.iproute.linux import AsyncIPRoute from pyroute2.netlink.exceptions import NetlinkError LOG = getLogger(__name__) class Trigger(StrEnum): '''Events that can trigger hooks in the client.''' # The client has obtained a new lease BOUND = auto() # The client has voluntarily relinquished its lease UNBOUND = auto() # The client has renewed its lease after the renewal timer expired RENEWED = auto() # The client has rebound its leas after the rebinding timer expired REBOUND = auto() # The lease has expired (the client will restart the lease process) EXPIRED = auto() class HookFunc(Protocol): '''Signature for functions that can be passed to the hook decorator.''' __name__: str async def __call__(self, lease: Lease) -> None: pass # pragma: no cover class Hook(NamedTuple): '''Stores a hook function and its triggers. Returned by the `hook()` decorator; no need to subclass or instantiate. ''' func: HookFunc triggers: set[Trigger] async def __call__(self, lease: Lease) -> None: '''Call the hook function.''' await self.func(lease=lease) @property def name(self) -> str: '''Shortcut for the function name.''' return self.func.__name__ async def run_hooks( hooks: Iterable[Hook], lease: Lease, trigger: Trigger, timeout: Optional[float] = None, ): '''Called by the client to run the hooks registered for the given trigger. The optional `timeout` arguments causes individual hooks to timeout if they exceed it. Exceptions are handled and printed, but don't prevent the other hooks from running. .. warning:: The timeout is async, which means that hooks that block on non-async code can ignore it and freeze the whole DHCP client ! ''' if hooks := list(hooks): LOG.debug('Running %s hooks', trigger) for i in filter(lambda y: trigger in y.triggers, hooks): try: await asyncio.wait_for(i(lease), timeout=timeout) except asyncio.exceptions.TimeoutError: LOG.error('Hook %r timed out', i.name) except Exception as exc: LOG.error('Hook %s failed: %r', i.name, exc) def hook(*triggers: Trigger) -> Callable[[HookFunc], Hook]: '''Decorator for dhcp client hooks. A hook is an async function that takes a lease. Hooks set in `ClientConfig.hooks` will be called in order by the client when one of the triggers passed to the decorator happens. For example:: @hook(Trigger.RENEWED) async def lease_was_renewed(lease: Lease): print(lease.server_mac, 'renewed our lease !') .. warning:: - blocking non-async code in hooks might freeze the client - long-running async hooks might be canceled after a timeout The decorator returns a `Hook` instance, a utility class storing the hook function and its triggers. .. warning:: The hooks API might still change. ''' def decorator(hook_func: HookFunc) -> Hook: return Hook(func=hook_func, triggers=set(triggers)) return decorator @hook(Trigger.BOUND) async def configure_ip(lease: Lease): '''Add the IP allocated in the lease to its interface. Use the `remove_ip` hook in addition to this one for cleanup. The DHCP server must have set the subnet mask and broadcast address. ''' LOG.info( 'Adding %s/%s to %s', lease.ip, lease.subnet_mask, lease.interface ) try: bcast: Optional[str] = lease.broadcast_address except MissingOptionError as err: LOG.debug("%s", err) bcast = None async with AsyncIPRoute(ext_ack=True, strict_check=True) as ipr: await ipr.addr( 'replace', index=await ipr.link_lookup(ifname=lease.interface), address=lease.ip, prefixlen=lease.subnet_mask, broadcast=bcast, ) @hook(Trigger.UNBOUND, Trigger.EXPIRED) async def remove_ip(lease: Lease): '''Remove the IP in the lease from its interface.''' LOG.info( 'Removing %s/%s from %s', lease.ip, lease.subnet_mask, lease.interface ) async with AsyncIPRoute(ext_ack=True, strict_check=True) as ipr: await ipr.addr( 'del', index=await ipr.link_lookup(ifname=lease.interface), address=lease.ip, prefixlen=lease.subnet_mask, ) @hook(Trigger.BOUND) async def add_default_gw(lease: Lease): '''Configures the default gateway set in the lease. Use in addition to `remove_default_gw` for cleanup. ''' LOG.info( 'Adding %s as default route through %s', lease.default_gateway, lease.interface, ) async with AsyncIPRoute(ext_ack=True, strict_check=True) as ipr: ifindex = (await ipr.link_lookup(ifname=lease.interface),) await ipr.route( 'replace', dst='0.0.0.0/0', gateway=lease.default_gateway, oif=ifindex, ) @hook(Trigger.UNBOUND, Trigger.EXPIRED) async def remove_default_gw(lease: Lease): '''Removes the default gateway set in the lease.''' LOG.info('Removing %s as default route', lease.default_gateway) async with AsyncIPRoute(ext_ack=True, strict_check=True) as ipr: ifindex = await ipr.link_lookup(ifname=lease.interface) try: await ipr.route( 'del', dst='0.0.0.0/0', gateway=lease.default_gateway, oif=ifindex, ) except NetlinkError as err: if err.code == errno.ESRCH: LOG.info( 'Default route was already removed by another process' ) else: LOG.error('Got a netlink error: %s', err)	6.07 KB	2025-05-25 12:26
📄 iface_status.py import asyncio import sys from typing import Literal, Optional from pyroute2.iproute.linux import AsyncIPRoute from pyroute2.netlink.exceptions import NetlinkError from pyroute2.netlink.rtnl import RTMGRP_LINK IfaceState = Literal['up', 'down'] class InterfaceNotFound(LookupError): '''Raised when an interface is not found.''' class InterfaceStateWatcher: '''Async context manager, fires events when an interface changes state. Used by the dhcp client to restart itself in such cases. ''' def __init__(self, interface: str) -> None: self.interface = interface self.up = asyncio.Event() self.down = asyncio.Event() self._state: Optional[IfaceState] = None self._watcher: Optional[asyncio.Task] = None self._ipr: Optional[AsyncIPRoute] = None @property def ipr(self) -> AsyncIPRoute: '''The async iproute context. Only useable in the context manager.''' assert self._ipr, 'need to use as a context manager' return self._ipr @property def state(self) -> IfaceState: '''The current state of the watched interface.''' assert self._state, "not yet started" return self._state @state.setter def state(self, value: IfaceState) -> None: '''Set the state & trigger the relevant event.''' self._state = value if value == 'up': self.up.set() self.down.clear() elif value == 'down': self.down.set() self.up.clear() async def _fetch_current_state(self) -> IfaceState: '''Get the initial state before we're notified of changes.''' lookup_results = await self.ipr.link_lookup(ifname=self.interface) if not lookup_results: raise InterfaceNotFound(self.interface) get_results = await self.ipr.link("get", index=lookup_results[0]) return get_results[0].get('state') async def _watch_changes(self) -> None: '''Updates `state` in real time forever.''' while True: # TODO: svinota says we should call .link('dump') here # to empty the buffer before starting async for msg in self.ipr.get(): if msg.get('IFLA_IFNAME') == self.interface: self.state = msg.get('state') async def __aenter__(self) -> 'InterfaceStateWatcher': self._ipr = await AsyncIPRoute().__aenter__() await self.ipr.bind(RTMGRP_LINK) self.state = await self._fetch_current_state() self._watcher = asyncio.create_task(self._watch_changes()) return self async def __aexit__(self, *_) -> None: self.ipr.close() if self._watcher: try: await self._watcher except NetlinkError as exc: if exc.code == 104: # we called close() so that is expected pass else: raise self._watcher = None self._ipr = None async def main(iface: str, state: str) -> None: '''Very basic entrypoint for commandline testing.''' assert state in ('up', 'down') async with InterfaceStateWatcher(iface) as watcher: print('Will exit when', iface, 'is', state) await getattr(watcher, state).wait() if __name__ == '__main__': try: asyncio.run(main(sys.argv[1], sys.argv[2])) except IndexError: print('Usage:', sys.argv[0], '<interface> <status>')	3.42 KB	2025-05-25 12:26
📄 leases.py '''Lease classes used by the dhcp client.''' import abc import json import time from dataclasses import asdict, dataclass, field from logging import getLogger from pathlib import Path from secrets import SystemRandom from typing import Generic, Optional, TypeVar from pyroute2.common import dqn2int from pyroute2.dhcp.dhcp4msg import dhcp4msg from pyroute2.dhcp.enums.dhcp import Option LOG = getLogger(__name__) random = SystemRandom() def _now() -> float: '''The current timestamp.''' return time.time() class MissingOptionError(LookupError): '''Raised when trying to access a missing option in a lease.''' def __init__(self, opt: Option): super().__init__(f"Lease does not set {opt!r}") LeaseOptionT = TypeVar('LeaseOptionT') class LeaseOption(Generic[LeaseOptionT]): '''Descriptor to factorize properties on leases that read DHCP options.''' def __init__(self, opt: Option) -> None: '''Create a new instance that looks up the given option.''' self.opt = opt def __get__( self, obj: object, objtype: Optional[type] = None ) -> LeaseOptionT: '''Gets the option matching self.opt in the lease.''' assert isinstance(obj, Lease) opt_name = self.opt.name.lower() if opt_name not in obj.ack['options']: raise MissingOptionError(self.opt) opt_value = obj.ack['options'][opt_name] return opt_value @dataclass class Lease(abc.ABC): '''Represents a lease obtained through DHCP.''' # The DHCP ack sent by the server which allocated this lease ack: dhcp4msg # Name of the interface for which this lease was requested interface: str # MAC address of the server that allocated the lease server_mac: str # Timestamp of when this lease was obtained obtained: float = field(default_factory=_now) def _seconds_til_timer(self, value: int) -> Optional[float]: '''Number of seconds to wait until the given timer value expires. If the value is -1, that means it is infinite and None is returned. ''' if value == -1: # infinite timer return None return self.obtained + value - _now() @property def expired(self) -> bool: '''Whether this lease has expired (its expiration is in the past). When loading a persisted lease, this won't be correct if the clock has been adjusted since the lease was written. However the worst case scenario is that we send a REQUEST for it, get a NAK and restart from scratch. ''' return self.expiration_in is not None and self.expiration_in <= 0 @property def expiration_in(self) -> Optional[float]: '''The amount of seconds before the lease expires. Computed from the `lease_time` option. Can be negative if it's past due, None if infinite, or raise MissingOptionError if the server did not give a lease time. ''' return self._seconds_til_timer(self.lease_time) @property def renewal_in(self) -> Optional[float]: '''The amount of seconds before we have to renew the lease. Computed from the `renewal_time` option, defaults to ~.5 * lease exp. Can be negative if it's past due, or None if infinite. ''' try: return self._seconds_til_timer(self.renewal_time) except MissingOptionError: # RFC section 4.4.5 says we need a fuzzy value around 0.5 if self.expiration_in: return self.expiration_in * random.uniform(0.4, 0.6) return None @property def rebinding_in(self) -> Optional[float]: '''The amount of seconds before we have to rebind the lease. Computed from the `rebinding_time` option, defaults to ~.8 * lease exp. Can be negative if it's past due, or None if infinite. ''' try: return self._seconds_til_timer(self.rebinding_time) except MissingOptionError: if self.expiration_in: return self.expiration_in * random.uniform(0.75, 0.90) return None @property def ip(self) -> str: '''The IP address assigned to the client.''' return self.ack['yiaddr'] @property def prefixlen(self) -> int: '''The length of the subnet mask assigned to the client.''' return dqn2int(self.subnet_mask) @property def default_gateway(self) -> str: '''The default gateway for this interface. As mentioned by the RFC, the first router is the most prioritary. ''' # TODO: unit test to make this crash return self.routers[0] # timers lease_time = LeaseOption[int](Option.LEASE_TIME) renewal_time = LeaseOption[int](Option.RENEWAL_TIME) rebinding_time = LeaseOption[int](Option.REBINDING_TIME) # The subnet mask assigned to the client. subnet_mask = LeaseOption[str](Option.SUBNET_MASK) routers = LeaseOption[list[str]](Option.ROUTER) # The broadcast address for this network. broadcast_address = LeaseOption[str](Option.BROADCAST_ADDRESS) # The MTU for this interface. mtu = LeaseOption[int](Option.INTERFACE_MTU) name_servers = LeaseOption[list[str]](Option.NAME_SERVER) # The IP address of the server which allocated this lease. server_id = LeaseOption[str](Option.SERVER_ID) domain_name = LeaseOption[str](Option.DOMAIN_NAME) domain_search = LeaseOption[list[str]](Option.DOMAIN_SEARCH) @abc.abstractmethod def dump(self) -> None: '''Write a lease, i.e. to disk or to stdout.''' @classmethod @abc.abstractmethod def load(cls, interface: str) -> 'Optional[Lease]': '''Load an existing lease for an interface, if it exists. The lease is not checked for freshness, and will be None if no lease could be loaded. ''' class JSONStdoutLease(Lease): '''Just prints the lease to stdout when the client gets a new one.''' def dump(self) -> None: '''Writes the lease as json to stdout.''' print(json.dumps(asdict(self), indent=2)) @classmethod def load(cls, interface: str) -> None: '''Does not do anything.''' return None class JSONFileLease(Lease): '''Write and load the lease from a JSON file in the working directory.''' @classmethod def _get_lease_dir(cls) -> Path: '''Where to store the lease file, i.e. the working directory.''' return Path.cwd() @classmethod def _get_path(cls, interface: str) -> Path: '''The lease file, named after the interface.''' return ( cls._get_lease_dir() .joinpath(interface) .with_name(f'{interface}.lease.json') ) def dump(self) -> None: '''Dump the lease to a file. The lease file is named after the interface and written in the working directory. ''' lease_path = self._get_path(self.interface) LOG.info('Writing lease for %s to %s', self.interface, lease_path) with lease_path.open('wt') as lf: json.dump(asdict(self), lf, indent=2) @classmethod def load(cls, interface: str) -> 'Optional[JSONFileLease]': '''Load the lease from a file. The lease file is named after the interface and read from the working directory. ''' lease_path = cls._get_path(interface) try: with lease_path.open('rt') as lf: LOG.info('Loading lease for %s from %s', interface, lease_path) return cls(**json.load(lf)) except FileNotFoundError: LOG.info('No existing lease at %s for %s', lease_path, interface) return None except (TypeError, ValueError) as err: LOG.warning('Error loading lease: %s', err) return None	7.75 KB	2025-05-25 12:26
📄 messages.py """Helper functions to build dhcp client messages.""" from dataclasses import dataclass from typing import Iterable, Literal, Optional from pyroute2.dhcp import enums from pyroute2.dhcp.dhcp4msg import dhcp4msg from pyroute2.dhcp.fsm import State from pyroute2.dhcp.leases import Lease from pyroute2.dhcp.xids import Xid Parameters = Iterable[enums.dhcp.Option] @dataclass class _DHCPMessage: '''A DHCP message with some extra info from other layers.''' dhcp: dhcp4msg eth_src: str eth_dst: str = 'ff:ff:ff:ff:ff:ff' ip_src: str = '0.0.0.0' ip_dst: str = '255.255.255.255' sport: int = 68 dport: int = 67 @property def message_type(self) -> enums.dhcp.MessageType: '''The DHCP message type (DISCOVER, REQUEST, ACK...)''' return self.dhcp['options']['message_type'] @property def xid(self) -> Xid: return Xid(self.dhcp['xid']) @dataclass class SentDHCPMessage(_DHCPMessage): '''A DHCP message to be sent to a server or broadcast.''' eth_src: Optional[str] = None # type: ignore[assignment] def __str__(self) -> str: type_name = self.message_type.name return ( f'{type_name} to {self.eth_dst}/{self.ip_dst}:{self.dport} ' f'(xid {self.xid})' ) class ReceivedDHCPMessage(_DHCPMessage): '''A DHCP message received by the client.''' def __str__(self) -> str: type_name = self.dhcp['options']['message_type'].name return ( f'{type_name} from {self.eth_src}/{self.ip_src}:{self.sport} ' f'(xid {self.xid})' ) def discover(parameter_list: Parameters) -> SentDHCPMessage: '''Make a broadcast DISCOVER message for the given parameters.''' # Default for SentDHCPMessage is broadcast which is what we want here return SentDHCPMessage( dhcp=dhcp4msg( { 'op': enums.bootp.MessageType.BOOTREQUEST, 'flags': enums.bootp.Flag.BROADCAST, 'options': { 'message_type': enums.dhcp.MessageType.DISCOVER, 'parameter_list': list(parameter_list), }, } ) ) def request_for_offer( parameter_list: Parameters, offer: ReceivedDHCPMessage ) -> SentDHCPMessage: '''Make a REQUEST message for a given OFFER. Since we don't have an IP yet, the message is always broadcast. When requesting an offer in the Selecting state, the server_id DHCP option is always set as opposed to when a REQUEST is sent in other states. See RFC 2131 section 4.3.2. ''' return SentDHCPMessage( dhcp=dhcp4msg( { 'op': enums.bootp.MessageType.BOOTREQUEST, 'flags': enums.bootp.Flag.BROADCAST, 'options': { 'message_type': enums.dhcp.MessageType.REQUEST, 'requested_ip': offer.dhcp['yiaddr'], 'server_id': offer.dhcp['options']['server_id'], 'parameter_list': list(parameter_list), }, } ) ) def request_for_lease( parameter_list: Parameters, lease: Lease, state: Literal[State.RENEWING, State.REBINDING, State.REBOOTING], ) -> SentDHCPMessage: '''Make a REQUEST for an existing lease. This differs from REQUESTs in response to an OFFER in that the server_id option is never set. When rebooting, the message is broadcast, and the requested_ip option is set to the IP in the stored lease. The bootp client IP is left blank. When renewing, (i.e. T1 expires) the message is for the server that granted the lease. The lease's IP is expected to be assigned to the client's interface at this point. When rebinding (T2), the message is broadcast on the network. In both cases, the bootp client IP (ciaddr) is set to the lease's IP. See RFC 2131 section 4.3.6. ''' dhcp_msg = dhcp4msg( { 'op': enums.bootp.MessageType.BOOTREQUEST, 'flags': enums.bootp.Flag.BROADCAST, 'options': { 'message_type': enums.dhcp.MessageType.REQUEST, 'parameter_list': list(parameter_list), }, } ) if state == State.REBOOTING: dhcp_msg['options']['requested_ip'] = lease.ip else: dhcp_msg['ciaddr'] = lease.ip if state == State.RENEWING: # T1 timer expired, send a request directly to the known server dhcp_msg['flags'] = enums.bootp.Flag.UNICAST return SentDHCPMessage( dhcp=dhcp_msg, eth_dst=lease.server_mac, # default ip_dst to broadcast, it should not be necessary # but it'll avoid a crash if the server_id is missing ip_dst=lease.server_id or '255.255.255.255', ip_src=lease.ip, ) # Reboot or rebind, broadcast the request return SentDHCPMessage(dhcp=dhcp_msg) def release(lease: Lease) -> SentDHCPMessage: '''Make a RELEASE for an existing & active lease.''' # RELEASE messages have nearly no allowed options, # and the released IP address must be set in ciaddr return SentDHCPMessage( dhcp=dhcp4msg( { 'op': enums.bootp.MessageType.BOOTREQUEST, 'flags': enums.bootp.Flag.UNICAST, 'ciaddr': lease.ip, 'options': { 'message_type': enums.dhcp.MessageType.RELEASE, 'server_id': lease.server_id, }, } ), # RELEASEs are unicast (see rfc section 4.4.4) eth_dst=lease.server_mac, # default ip_dst to broadcast, it should not be necessary # but it'll avoid a crash if the server_id is missing ip_dst=lease.server_id or '255.255.255.255', ip_src=lease.ip, )	5.8 KB	2025-05-25 12:26
📄 server_detector.py import argparse import asyncio import dataclasses import json import logging import time from asyncio.exceptions import CancelledError, TimeoutError from secrets import SystemRandom from typing import AsyncGenerator, Iterable from pyroute2.dhcp import messages from pyroute2.dhcp.dhcp4socket import AsyncDHCP4Socket from pyroute2.dhcp.enums.dhcp import Option DHCPResponse = tuple[str, messages.ReceivedDHCPMessage] LOG = logging.getLogger('dhcp-server-detector') DEFAULT_PARAMETERS = ( Option.SUBNET_MASK, Option.ROUTER, Option.BROADCAST_ADDRESS, Option.NAME_SERVER, ) class DHCPServerDetector: '''Sends `DISCOVER`s on interfaces and listens for responses.''' def __init__( self, interfaces: str, duration: float = 25.0, interval: float = 4.0, requested_parameters: messages.Parameters = DEFAULT_PARAMETERS, sport: int = 68, ): self.interfaces = interfaces self.interval = interval self.duration = duration self.sport = sport # The DISCOVERs that will be sent repeatedly per interface self.discover_messages = self._make_discover_msgs( interfaces=self.interfaces, params=requested_parameters ) # All received responses are put here by the dedicated tasks, # along with the name of the interface they were received on. self._responses_queue: asyncio.Queue[DHCPResponse] = asyncio.Queue() @classmethod def _make_discover_msgs( cls, interfaces: Iterable[str], params: messages.Parameters ) -> dict[str, messages.SentDHCPMessage]: '''Generate DISCOVERs with a different xid for each interface .''' msgs: dict[str, messages.SentDHCPMessage] = {} rand = SystemRandom().randint for i in interfaces: msgs[i] = messages.discover(parameter_list=params) # use a different xid per interface msgs[i].dhcp['xid'] = rand(0xFF, 0xFFFFFFFF) return msgs async def _send_forever(self, sock: AsyncDHCP4Socket): '''Send the `DISCOVER` message at `interval` until cancelled.''' msg = self.discover_messages[sock.ifname] while True: LOG.info('[%s] -> %s', sock.ifname, msg) await sock.put(msg) await asyncio.sleep(self.interval) async def _get_offers(self, interface: str): '''Send DISCOVERs and wait for responses on an interface.''' expected_xids = { ifname: msg.dhcp['xid'] for ifname, msg in self.discover_messages.items() } async with AsyncDHCP4Socket(ifname=interface, port=self.sport) as sock: send_task = asyncio.create_task( self._send_forever(sock), name=f'Send DISCOVERS on {interface}' ) # if send_task returns, that means _send_forever crashed while not send_task.done(): try: get_next_msg = asyncio.Task( sock.get(), name=f'Get message from {interface}' ) # wait for a message and put it in the queue _, pending = await asyncio.wait( [send_task, get_next_msg], return_when=asyncio.FIRST_COMPLETED, ) # if get is still pending, that means send_task is over if get_next_msg in pending: get_next_msg.cancel() continue # we have a new message next_msg = await get_next_msg if next_msg.dhcp['xid'] != expected_xids[sock.ifname]: LOG.debug( '[%s] Got %s with xid mismatch, ignoring', interface, next_msg.message_type.name, ) continue LOG.info('[%s] <- %s', interface, next_msg) await self._responses_queue.put((interface, next_msg)) except asyncio.CancelledError: LOG.debug('[%s] stop discovery', interface) send_task.cancel() break else: await send_task async def detect_servers(self) -> AsyncGenerator[DHCPResponse, None]: '''Detect DHCP servers on `interfaces` for `duration`. Yields tuples of (interface name, response). ''' discover_tasks = [ asyncio.create_task(self._get_offers(i), name=i) for i in self.interfaces ] started = time.time() remaining = self.duration while remaining > 0 and discover_tasks: get_response = asyncio.create_task(self._responses_queue.get()) try: done, _ = await asyncio.wait( [get_response, discover_tasks], timeout=remaining, return_when=asyncio.FIRST_COMPLETED, ) if get_response in done: yield await get_response done.remove(get_response) else: get_response.cancel() for i in done: if task_exc := i.exception(): LOG.error("%r: %s", i.get_name(), task_exc) discover_tasks.remove(i) remaining -= time.time() - started except (TimeoutError, CancelledError): break for i in discover_tasks: i.cancel() def get_argparser() -> argparse.ArgumentParser: psr = argparse.ArgumentParser( description='Send DHCP DISCOVER messages on the given interface(s) ' 'and collect all responses. ' 'Responses and their metadata are printed as JSON to stdout.', epilog='Exits successfully if at least one response was received.', formatter_class=argparse.ArgumentDefaultsHelpFormatter, ) psr.add_argument( 'interface', nargs='+', help='Interface(s) to DISCOVER on.' ) psr.add_argument( '-d', '--duration', type=float, default=30.0, help='Number of seconds spent collecting responses.', metavar='SEC', ) psr.add_argument( '-i', '--interval', type=float, default=4.0, help='Interval in seconds between DISCOVERs.', metavar='SEC', ) psr.add_argument( '-s', '--source-port', type=int, default=68, help='Source port to bind to.', metavar='PORT', ) psr.add_argument( '-1', '--exit-on-first-offer', action='store_true', default=False, help="Exit as soon as a response is received.", ) psr.add_argument( '-l', '--log-level', default='WARNING', help="Log level. Set to INFO to log sent & received messages.", metavar='LEVEL', ) return psr async def main() -> int: '''Commandline entrypoint. Returns the number of received responses.''' args = get_argparser().parse_args() LOG.setLevel(args.log_level) detector = DHCPServerDetector( *args.interface, interval=args.interval, duration=args.duration, sport=args.source_port, ) response_count: int = 0 async for interface, msg in detector.detect_servers(): response_count += 1 print( json.dumps( {'interface': interface, 'message': dataclasses.asdict(msg)}, indent=2, ) ) if args.exit_on_first_offer: break return response_count def run(): logging.basicConfig( format='%(asctime)s %(levelname)s %(name)s %(message)s' ) # Exit on failure if there were no received responses exit(asyncio.run(main()) == 0) if __name__ == '__main__': run()	7.86 KB	2025-05-25 12:26
📄 timers.py '''Timers to manage lease rebinding, renewal & expiration.''' import asyncio import dataclasses from logging import getLogger from typing import Optional from pyroute2.dhcp.fsm import AsyncCallback from pyroute2.dhcp.leases import Lease LOG = getLogger(__name__) @dataclasses.dataclass class LeaseTimers: '''Manage callbacks associated with DHCP leases.''' renewal: Optional[asyncio.TimerHandle] = None rebinding: Optional[asyncio.TimerHandle] = None expiration: Optional[asyncio.TimerHandle] = None def cancel(self) -> None: '''Cancel all current timers.''' for timer_name in ('renewal', 'rebinding', 'expiration'): self._reset_timer(timer_name) def _reset_timer(self, timer_name: str) -> None: '''Cancel a timer and set it to None.''' if timer_name not in ('renewal', 'rebinding', 'expiration'): raise ValueError(f'Unknown timer {timer_name!r}') if timer := getattr(self, timer_name): assert isinstance(timer, asyncio.TimerHandle) if not timer.cancelled(): # FIXME: how do we know a timer wasn't cancelled ? # this causes spurious logs LOG.debug('Canceling %s timer', timer_name) timer.cancel() setattr(self, timer_name, None) def arm(self, lease: Lease, **callbacks: AsyncCallback) -> None: '''Reset & arm timers from a `Lease`. `callbacks` must be async callables with no arguments that will be called when the associated timer expires. ''' self.cancel() loop = asyncio.get_running_loop() for timer_name, async_callback in callbacks.items(): self._reset_timer(timer_name) lease_time = getattr(lease, f'{timer_name}_in') if not lease_time: LOG.debug('%s time is infinite', timer_name) continue if lease_time < 0.0: LOG.debug('%s is in the past', timer_name) continue LOG.info('Scheduling %s in %.2fs', timer_name, lease_time) timer = loop.call_later( lease_time, self._create_timer_task, timer_name, async_callback ) setattr(self, timer_name, timer) def _create_timer_task( self, timer_name: str, async_callback: AsyncCallback ) -> None: ''' 'Internal callback for loop.call_later. Creates a Task that runs the async_callback. ''' asyncio.create_task( async_callback(), name=f"{timer_name} timer callback" )	2.55 KB	2025-05-25 12:26
📄 xids.py from secrets import SystemRandom from typing import Any, Optional, Union from pyroute2.dhcp.fsm import State random = SystemRandom() def random_xid_prefix() -> int: '''A (max) 32 bit random int with its lowest nibble set to 0. These last 4 bits are used by the client to store its state, in the form of the associated `State` value, when sending a DHCP message. ''' return random.randint(0x00000010, 0xFFFFFFF0) class Xid: '''Transaction IDs used to identify responses to DHCP requests. We use the last nibble to store the state the message was sent in. The most significant bits store a random value we use to match requests and responses, since the RFC specifies that the server must send back the same value when answering. (see RFC section 4.1) ''' def __init__(self, value: Optional[Union[int, 'Xid']] = None): if isinstance(value, int): int_value = value elif isinstance(value, Xid): int_value = int(value) elif value is None: int_value = random_xid_prefix() else: raise TypeError(f'{value!r} is not an xid') assert int_value < 0xFFFFFFFF # we have 32 bits self._value = int_value @property def random_part(self) -> int: '''The random part of the transaction id.''' return self._value & 0xFFFFFFF0 @property def request_state(self) -> Optional[State]: '''The state in which the request was sent. Since servers answer with the same transaction ID as the request, we can use this to know what client state does a response answer to. ''' try: return State(self._value & 0xF) except ValueError: return None def for_state(self, state: State) -> 'Xid': '''A new Xid built from the random part + the state.''' return Xid(self.random_part \| state) def __index__(self) -> int: '''Allows xids to be used as int.''' return self._value def matches(self, received_xid: 'Xid'): '''Loose match, whether the random part of both XIDs match. This can be used to check if a message is indeed in response to a request we sent. ''' return self.random_part == received_xid.random_part def __eq__(self, other: Any) -> bool: '''Xids compare to other xids or ints.''' if isinstance(other, Xid): return self._value == other._value elif isinstance(other, int): return self._value == other return False def __str__(self) -> str: return hex(self._value) def __repr__(self) -> str: return f"Xid({self})"	2.65 KB	2025-05-25 12:26

Terminal