WIP: add an async dhcp client

pyroute2/compat.py

@@ -0,0 +1,21 @@
+'''Compatibility with older but supported Python versions'''
+
+try:
+    from enum import StrEnum
+except ImportError:
+    # StrEnum appeared in python 3.11
+
+    from enum import Enum
+
+    class StrEnum(str, Enum):
+        '''Same as enum, but members are also strings.'''
+
+
+try:
+    from socket import ETHERTYPE_IP
+except ImportError:
+    # ETHERTYPE_* are new in python 3.12
+    ETHERTYPE_IP = 0x800
+
+
+__all__ = ('StrEnum',)

pyroute2/dhcp/__init__.py

@@ -105,18 +105,7 @@ class array8(option):
 from pyroute2.common import basestring
 from pyroute2.protocols import msg
 
-BOOTREQUEST = 1
-BOOTREPLY = 2
-
-DHCPDISCOVER = 1
-DHCPOFFER = 2
-DHCPREQUEST = 3
-DHCPDECLINE = 4
-DHCPACK = 5
-DHCPNAK = 6
-DHCPRELEASE = 7
-DHCPINFORM = 8
-
+from . import enums
 
 if not hasattr(array, 'tobytes'):
     # Python2 and Python3 versions of array differ,
@@ -262,8 +251,8 @@ def encode(self):
         self._register_options()
         # put message type
         options = self.get('options') or {
-            'message_type': DHCPDISCOVER,
-            'parameter_list': [1, 3, 6, 12, 15, 28],
+            'message_type': enums.dhcp.MessageType.DISCOVER,
+            'parameter_list': [1, 3, 6, 12, 15, 28],  # FIXME
         }
 
         self.buf += (
@@ -321,3 +310,6 @@ class array8(option):
 
     class client_id(option):
         fields = (('type', 'uint8'), ('key', 'l2addr'))
+
+    class message_type(option):
+        policy = {'format': 'B', 'decode': lambda x: enums.dhcp.MessageType(x)}

pyroute2/dhcp/cli.py

@@ -0,0 +1,106 @@
+import asyncio
+import logging
+from argparse import ArgumentDefaultsHelpFormatter, ArgumentParser
+from importlib import import_module
+from typing import Any
+
+from pyroute2.dhcp.client import AsyncDHCPClient
+from pyroute2.dhcp.fsm import State
+from pyroute2.dhcp.hooks import ConfigureIP, Hook
+from pyroute2.dhcp.leases import Lease
+
+
+def importable(name: str) -> Any:
+    '''Imports anything by name. Used by the argument parser.'''
+    module_name, obj_name = name.rsplit('.', 1)
+    module = import_module(module_name)
+    return getattr(module, obj_name)
+
+
+def get_psr() -> ArgumentParser:
+    psr = ArgumentParser(
+        description='pyroute2 DHCP client',
+        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=importable,
+        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.',
+        nargs='+',
+        type=importable,
+        default=[ConfigureIP],
+        metavar='dotted.name',
+    )
+    psr.add_argument(
+        '-x',
+        '--exit-on-lease',
+        help='Exit as soon as getting a lease.',
+        default=False,
+        action='store_true',
+    )
+    psr.add_argument(
+        '--log-level',
+        help='Logging level to use.',
+        choices=('DEBUG', 'INFO', 'WARNING', 'ERROR'),
+        default='INFO',
+    )
+    return psr
+
+
+async def main():
+    psr = get_psr()
+    args = psr.parse_args()
+    logging.basicConfig(
+        level=args.log_level,
+        format='%(asctime)s %(levelname)s [%(name)s:%(funcName)s] %(message)s',
+    )
+
+    if not issubclass(args.lease_type, Lease):
+        psr.error(f'{args.lease_type!r} must be a Lease subclass')
+
+    # Check hooks are subclasses of Hook
+    for i in args.hook:
+        if not issubclass(i, Hook):
+            psr.error(f'{i!r} must be a Hook subclass')
+
+    acli = AsyncDHCPClient(
+        interface=args.interface,
+        lease_type=args.lease_type,
+        # Instantiate hooks
+        hooks=[i() for i in args.hook],
+    )
+
+    # Open the socket, read existing lease, etc
+    async with acli:
+        # Bootstrap the client by sending a DISCOVER or a REQUEST
+        await acli.bootstrap()
+        if args.exit_on_lease:
+            # Wait until we're bound once, then exit
+            await acli.wait_for_state(State.BOUND)
+        else:
+            # Wait until the client is stopped otherwise
+            await acli.wait_for_state(None)
+
+
+def run():
+    # for the setup.cfg entrypoint
+    try:
+        asyncio.run(main())
+    except KeyboardInterrupt:
+        pass
+
+
+if __name__ == '__main__':
+    run()