lib.dhcp

  1import base64
  2import json
  3import re
  4from dataclasses import dataclass, field
  5from ipaddress import IPv4Address, IPv4Network
  6from pathlib import Path
  7from typing import Any
  8
  9from SRE.lib_sre import Grade0, NetScheme0
 10
 11
 12@dataclass
 13class DhcpSubnet:
 14    """Configuration for a single DHCP subnet declaration."""
 15    subnet: IPv4Network
 16    range_start: IPv4Address
 17    range_end: IPv4Address
 18    routers: list[IPv4Address]            = field(default_factory=list)
 19    dns_servers: list[IPv4Address]        = field(default_factory=list)
 20    domain_name: str | None               = None
 21    broadcast_address: IPv4Address | None = None
 22    default_lease_time: int | None        = None  # subnet-level override; None = use global
 23    max_lease_time: int | None            = None  # subnet-level override; None = use global
 24    fixed_addresses: dict[str, str]       = field(default_factory=dict)  # MAC → IP string
 25
 26
 27@dataclass
 28class DhcpParameters:
 29    """All parameters required to configure an ISC DHCP server on a Debian machine.
 30
 31    Covers both /etc/default/isc-dhcp-server (``interfaces_v4``, ``interfaces_v6``) and
 32    /etc/dhcp/dhcpd.conf (everything else).
 33    """
 34    interfaces_v4: list[str]            # written to INTERFACESv4 in /etc/default/isc-dhcp-server
 35    interfaces_v6: list[str]            = field(default_factory=list)  # written to INTERFACESv6
 36    subnets: list[DhcpSubnet]           = field(default_factory=list)
 37    authoritative: bool                 = True
 38    default_lease_time: int | None      = None  # omitted from global section when None
 39    max_lease_time: int | None          = None  # omitted from global section when None
 40    ddns_update_style: str              = 'none'
 41
 42
 43def set_dhcp_server(net_scheme: NetScheme0, machine: str,
 44                    dhcp_params: DhcpParameters, step: int = 1) -> None:
 45    """Write DHCP server configuration files and (re)start the service on *machine_name*.
 46
 47    Writes:
 48    - /etc/default/isc-dhcp-server  (INTERFACESv4)
 49    - /etc/dhcp/dhcpd.conf          (generated from dhcp_params)
 50
 51    Then runs ``systemctl enable`` and ``systemctl restart`` for isc-dhcp-server.
 52    """
 53    # /etc/default/isc-dhcp-server
 54    ifaces_v4 = ' '.join(dhcp_params.interfaces_v4)
 55    ifaces_v6 = ' '.join(dhcp_params.interfaces_v6)
 56    default_content = f'INTERFACESv4="{ifaces_v4}"\nINTERFACESv6="{ifaces_v6}"\n'
 57    net_scheme.file(machine, '/etc/default/isc-dhcp-server', default_content, step=step)
 58
 59    # /etc/dhcp/dhcpd.conf
 60    lines = []
 61    auth = 'authoritative' if dhcp_params.authoritative else 'not authoritative'
 62    lines += [
 63        f'ddns-update-style {dhcp_params.ddns_update_style};',
 64        f'{auth};',
 65    ]
 66    if dhcp_params.default_lease_time is not None:
 67        lines.append(f'default-lease-time {dhcp_params.default_lease_time};')
 68    if dhcp_params.max_lease_time is not None:
 69        lines.append(f'max-lease-time {dhcp_params.max_lease_time};')
 70    for s in dhcp_params.subnets:
 71        lines.append('')
 72        lines.append(f'subnet {s.subnet.network_address} netmask {s.subnet.netmask} {{')
 73        lines.append(f'    range {s.range_start} {s.range_end};')
 74        if s.routers:
 75            lines.append(f'    option routers {", ".join(str(r) for r in s.routers)};')
 76        if s.dns_servers:
 77            lines.append(f'    option domain-name-servers {", ".join(str(d) for d in s.dns_servers)};')
 78        if s.domain_name is not None:
 79            lines.append(f'    option domain-name "{s.domain_name}";')
 80        if s.broadcast_address is not None:
 81            lines.append(f'    option broadcast-address {s.broadcast_address};')
 82        if s.default_lease_time is not None:
 83            lines.append(f'    default-lease-time {s.default_lease_time};')
 84        if s.max_lease_time is not None:
 85            lines.append(f'    max-lease-time {s.max_lease_time};')
 86        for mac, ip in s.fixed_addresses.items():
 87            safe = mac.replace(':', '-')
 88            lines += [
 89                f'    host {safe} {{',
 90                f'        hardware ethernet {mac};',
 91                f'        fixed-address {ip};',
 92                f'    }}',
 93            ]
 94        lines.append('}')
 95    net_scheme.file(machine, '/etc/dhcp/dhcpd.conf', '\n'.join(lines) + '\n', step=step)
 96
 97    net_scheme.cmd(machine, 'systemctl enable isc-dhcp-server', step=step)
 98    net_scheme.cmd(machine, 'systemctl restart isc-dhcp-server', step=step)
 99
100
101def get_dhcp_server(grade: Grade0, machine: str,
102                    step: int = 1) -> tuple[DhcpParameters | None, int]:
103    """Read and parse the DHCP server configuration on *machine_name*.
104
105    Returns ``(params, errors)`` where *params* is a :class:`DhcpParameters`
106    instance (or ``None`` if /etc/default/isc-dhcp-server is absent/unreadable)
107    and *errors* is the number of parse errors found in dhcpd.conf.
108    """
109    interfaces_v4 = parse_ipv4_interfaces_in_default_dhcp_server_file(grade, machine, step)
110    if interfaces_v4 is None:
111        return None, 1
112    interfaces_v6 = parse_ipv6_interfaces_in_default_dhcp_server_file(grade, machine, step) or []
113
114    parsed = _parse_dhcpd_config(grade, machine, step)
115    errors: int = parsed['errors']
116    gp = parsed['global_parameters']
117
118    authoritative     = bool(gp.get('authoritative', False))
119    _dlt = gp.get('default-lease-time', None)
120    _mlt = gp.get('max-lease-time', None)
121    default_lease_time: int | None = int(_dlt) if _dlt is not None else None
122    max_lease_time: int | None     = int(_mlt) if _mlt is not None else None
123    ddns_update_style = str(gp.get('ddns-update-style', 'none'))
124
125    subnets: list[DhcpSubnet] = []
126    for net_addr, sp in parsed['subnets'].items():
127        # Reconstruct IPv4Network
128        try:
129            subnet_net = IPv4Network(f'{net_addr}/{sp["netmask"]}', strict=False)
130        except (KeyError, ValueError):
131            errors += 1
132            continue
133
134        # Parse pool range — stored as "start end" or "dynamic-bootp start end"
135        range_val = sp.get('range', '')
136        range_parts = range_val.split()
137        if range_parts and range_parts[0].lower() == 'dynamic-bootp':
138            range_parts = range_parts[1:]
139        try:
140            range_start = IPv4Address(range_parts[0])
141            range_end   = IPv4Address(range_parts[1] if len(range_parts) > 1 else range_parts[0])
142        except (IndexError, ValueError):
143            errors += 1
144            continue
145
146        def _parse_addr_list(val: str) -> list[IPv4Address]:
147            result = []
148            for part in re.split(r'[,\s]+', val.strip()):
149                if part:
150                    try:
151                        result.append(IPv4Address(part))
152                    except ValueError:
153                        pass
154            return result
155
156        routers = _parse_addr_list(sp.get('option routers', ''))
157        dns_servers = _parse_addr_list(sp.get('option domain-name-servers', ''))
158
159        domain_name = sp.get('option domain-name', None)
160        if domain_name is not None:
161            domain_name = domain_name.strip('"')
162
163        bcast_str = sp.get('option broadcast-address', None)
164        try:
165            broadcast_address = IPv4Address(bcast_str) if bcast_str else None
166        except ValueError:
167            broadcast_address = None
168
169        # Only set per-subnet overrides when they differ from the global values
170        sub_dlt = sp.get('default-lease-time', None)
171        sub_mlt = sp.get('max-lease-time', None)
172        try:
173            sub_dlt = int(sub_dlt) if sub_dlt is not None else None
174            sub_mlt = int(sub_mlt) if sub_mlt is not None else None
175        except ValueError:
176            sub_dlt = sub_mlt = None
177        subnet_default_lease = sub_dlt if sub_dlt != default_lease_time else None
178        subnet_max_lease     = sub_mlt if sub_mlt != max_lease_time else None
179
180        subnets.append(DhcpSubnet(
181            subnet=subnet_net,
182            range_start=range_start,
183            range_end=range_end,
184            routers=routers,
185            dns_servers=dns_servers,
186            domain_name=domain_name,
187            broadcast_address=broadcast_address,
188            default_lease_time=subnet_default_lease,
189            max_lease_time=subnet_max_lease,
190            fixed_addresses={str(k): str(v) for k, v in sp.get('fixed-addresses', {}).items()},
191        ))
192
193    params = DhcpParameters(
194        interfaces_v4=interfaces_v4,
195        interfaces_v6=interfaces_v6,
196        subnets=subnets,
197        authoritative=authoritative,
198        default_lease_time=default_lease_time,
199        max_lease_time=max_lease_time,
200        ddns_update_style=ddns_update_style,
201    )
202    return params, errors
203
204
205def _parse_dhcpd_interfaces(cmdline_output: str) -> list[str]:
206    """Extract interface names from dhcpd cmdline tokens (one token per line).
207
208    Returns the list of interface names passed to dhcpd, or ``["*"]`` if none
209    were specified (meaning dhcpd listens on all interfaces).
210    """
211    # Flags that consume the next token as their value argument.
212    VALUE_FLAGS = {
213        '-cf', '-lf', '-pf', '-tf', '-sf', '-hpf',
214        '-user', '-group', '-chroot', '-port', '-relay',
215    }
216    tokens = [t for t in cmdline_output.splitlines() if t.strip()]
217    interfaces = []
218    skip_next = False
219    for i, tok in enumerate(tokens):
220        if i == 0:
221            continue  # executable path (e.g. /usr/sbin/dhcpd)
222        if skip_next:
223            skip_next = False
224            continue
225        if tok in VALUE_FLAGS:
226            skip_next = True
227            continue
228        if tok.startswith('-'):
229            continue
230        interfaces.append(tok)
231    return interfaces if interfaces else ['*']
232
233
234def check_running_dhcp_server(grade: Grade0, machine: str) -> tuple[bool, list[str]]:
235    """Check whether ISC DHCP server is running on *machine*.
236
237    Returns a tuple ``(running, interfaces)`` where:
238
239    - *running*: ``True`` if ``isc-dhcp-server`` is currently active.
240    - *interfaces*: list of interface names dhcpd is bound to (e.g.
241      ``["eth0", "eth1"]``), or ``["*"]`` if it listens on all interfaces.
242      Empty list when *running* is ``False``.
243
244    Listening interfaces are read from the running process command line so
245    they reflect the actual state, not just the configuration file.
246    """
247    _, code = grade.test(machine, 'systemctl is-active isc-dhcp-server',
248                         allow_error=True)
249    if code != 0:
250        return False, []
251
252    cmdline_out, _ = grade.test(
253        machine,
254        r"tr '\000' '\n' < /proc/$(pidof -s dhcpd)/cmdline 2>/dev/null",
255        allow_error=True,
256    )
257    return True, _parse_dhcpd_interfaces(cmdline_out)
258
259
260def parse_ipv4_interfaces_in_default_dhcp_server_file(
261        grade: Grade0, machine: str, step: int = 1) -> list[str] | None:
262    """Return the list of interfaces from INTERFACESv4 in /etc/default/isc-dhcp-server.
263
264    Returns None if the file is absent or unreadable.
265    Returns an empty list if the file exists but INTERFACESv4 is not set.
266    """
267    output, code = grade.test(machine, 'cat /etc/default/isc-dhcp-server',
268                              step=step, allow_error=True)
269    if code != 0:
270        return None
271    for line in output.splitlines():
272        line = line.strip()
273        if line.startswith('#'):
274            continue
275        m = re.match(r'^INTERFACESv4\s*=\s*"([^"]*)"', line)
276        if m:
277            return m.group(1).split()
278    return []
279
280
281def parse_ipv6_interfaces_in_default_dhcp_server_file(
282        grade: Grade0, machine: str, step: int = 1) -> list[str] | None:
283    """Return the list of interfaces from INTERFACESv6 in /etc/default/isc-dhcp-server.
284
285    Returns None if the file is absent or unreadable.
286    Returns an empty list if the file exists but INTERFACESv6 is not set.
287    """
288    output, code = grade.test(machine, 'cat /etc/default/isc-dhcp-server',
289                              step=step, allow_error=True)
290    if code != 0:
291        return None
292    for line in output.splitlines():
293        line = line.strip()
294        if line.startswith('#'):
295            continue
296        m = re.match(r'^INTERFACESv6\s*=\s*"([^"]*)"', line)
297        if m:
298            return m.group(1).split()
299    return []
300
301
302def _parse_dhcpd_config(grade: Grade0, machine: str, step: int = 1) -> dict:
303    """Parse /etc/dhcp/dhcpd.conf on *machine* via grade.test() and return a
304    structured dict describing the ISC DHCP server configuration.
305
306    Returns a dict with three keys:
307
308    ``errors``
309        Number of syntactically incorrect or unrecognised statements.
310
311    ``global_parameters``
312        Dict of top-level parameters (name → value).  ``option`` parameters are
313        stored with their full two-word key, e.g. ``"option domain-name-servers"``.
314        ``authoritative`` maps to ``True``/``False``.  All other parameters map
315        to their value as a string (or space-joined string for multi-token values).
316
317    ``subnets``
318        Dict keyed by the network address string (e.g. ``"10.152.187.0"``).
319        Each value is a dict of *effective* parameters: global parameters are
320        copied first, then subnet-level declarations override them.  A special
321        key ``"fixed-addresses"`` holds a ``{MAC: IP-or-hostname}`` mapping
322        assembled from all ``host`` blocks whose ``fixed-address`` falls inside
323        that subnet (or that were declared directly inside that subnet block).
324    """
325    output, code = grade.test(machine, 'cat /etc/dhcp/dhcpd.conf', step=step)
326
327    empty: dict[str, Any] = {'errors': 0, 'global_parameters': {}, 'subnets': {}}
328    if code != 0 or not output:
329        return empty
330
331    # ------------------------------------------------------------------ #
332    # Strip comments: /* … */,  // … \n,  # … \n                         #
333    # ------------------------------------------------------------------ #
334    text = re.sub(r'/\*.*?\*/', '', output, flags=re.DOTALL)
335    text = re.sub(r'(?:#|//).*', '', text)
336
337    # ------------------------------------------------------------------ #
338    # Tokenise: {  }  ;  are single-character tokens; quoted strings are  #
339    # kept intact; everything else is split on whitespace.                #
340    # ------------------------------------------------------------------ #
341    token_re = re.compile(r'[{};]|"[^"]*"|[^\s{};]+')
342    tokens = token_re.findall(text)
343
344    # ------------------------------------------------------------------ #
345    # Known single-value global / subnet keywords                         #
346    # ------------------------------------------------------------------ #
347    _KNOWN_PARAMS = {
348        'default-lease-time', 'max-lease-time', 'min-lease-time',
349        'ddns-update-style', 'log-facility', 'server-identifier',
350        'filename', 'next-server', 'server-name',
351        'use-host-decl-names', 'get-lease-hostnames', 'ping-check',
352        'one-lease-per-client', 'dynamic-bootp-lease-length',
353        'lease-file-name', 'pid-file-name', 'omapi-port',
354        'update-conflict-detection', 'update-optimization',
355        'stash-agent-options', 'local-port', 'remote-port',
356        'db-time-format', 'bootp-lease-length', 'min-secs',
357        'always-reply-rfc1048', 'server-name',
358    }
359
360    _KNOWN_HOST_PARAMS = {
361        'hardware', 'fixed-address', 'filename', 'next-server',
362        'server-name', 'client-identifier', 'ddns-hostname',
363        'ddns-domainname', 'option', 'supersede', 'prepend', 'append',
364        'default', 'deny', 'allow', 'ignore',
365    }
366
367    # ------------------------------------------------------------------ #
368    # Parser state                                                         #
369    # ------------------------------------------------------------------ #
370    errors: int = 0
371
372    # Collected raw data
373    global_params: dict[str, Any] = {}
374    # net_addr -> {'netmask': str, 'fixed-addresses': {}, param: val, ...}
375    subnets: dict[str, dict[str, Any]] = {}
376
377    # Hosts collected during parsing: (parent_subnet_net or None, mac, ip)
378    all_hosts: list[tuple[str | None, str, str]] = []
379
380    # Context stack: list of (kind, data)
381    #   kind = 'global' | 'subnet' | 'host' | 'other'
382    #   data = net_addr str for 'subnet', host-info dict for 'host', None otherwise
383    context_stack: list[tuple[str, Any]] = [('global', None)]
384
385    pending: list[str] = []   # words accumulated before the next ; or {
386
387    def _find_parent_subnet() -> str | None:
388        for frame in reversed(context_stack):
389            if frame[0] == 'subnet':
390                return frame[1]
391        return None
392
393    def _parse_param(words: list[str]) -> tuple[str, Any] | None:
394        """Return (key, value) for a recognised parameter, or None on error."""
395        if not words:
396            return None
397        kw = words[0].lower()
398        rest = words[1:]
399
400        if kw == 'authoritative':
401            return 'authoritative', True
402
403        if kw == 'not' and rest and rest[0].lower() == 'authoritative':
404            return 'authoritative', False
405
406        if kw == 'option':
407            if not rest:
408                return None
409            opt_name = rest[0].lower()
410            opt_val = ' '.join(v.strip('"') for v in rest[1:]) if len(rest) > 1 else ''
411            return f'option {opt_name}', opt_val
412
413        if kw in ('allow', 'deny', 'ignore'):
414            if not rest:
415                return None
416            return f'{kw} {" ".join(rest)}', True
417
418        if kw == 'range':
419            # range [dynamic-bootp] start [end]
420            return 'range', ' '.join(rest)
421
422        if kw == 'include':
423            # include "file"; — silently skip
424            return 'include', ' '.join(v.strip('"') for v in rest)
425
426        if kw in _KNOWN_PARAMS:
427            val = ' '.join(v.strip('"') for v in rest) if rest else ''
428            return kw, val
429
430        return None  # unrecognised keyword
431
432    # ------------------------------------------------------------------ #
433    # Main token loop                                                      #
434    # ------------------------------------------------------------------ #
435    for tok in tokens:
436
437        if tok == ';':
438            if not pending:
439                continue  # empty statement is fine
440
441            kw = pending[0].lower()
442            kind, data = context_stack[-1]
443
444            if kind == 'host':
445                if kw == 'hardware' and len(pending) >= 3 and pending[1].lower() == 'ethernet':
446                    data['mac'] = pending[2].lower()
447                elif kw == 'fixed-address' and len(pending) >= 2:
448                    data['ip'] = pending[1]
449                elif kw in _KNOWN_HOST_PARAMS:
450                    pass  # valid host option, not needed for our output
451                else:
452                    errors += 1
453
454            elif kind in ('global', 'subnet'):
455                parsed = _parse_param(pending)
456                if parsed is not None:
457                    target = global_params if kind == 'global' else subnets[data]
458                    key, val = parsed
459                    if key != 'include':   # do not store include directives
460                        target[key] = val
461                else:
462                    errors += 1
463
464            # else: inside 'other' (pool, group, shared-network, …) — ignore content
465
466            pending = []
467
468        elif tok == '{':
469            kw = pending[0].lower() if pending else ''
470
471            if kw == 'subnet':
472                if len(pending) >= 4 and pending[2].lower() == 'netmask':
473                    net_addr = pending[1]
474                    netmask = pending[3]
475                    subnets[net_addr] = {'netmask': netmask, 'fixed-addresses': {}}
476                    context_stack.append(('subnet', net_addr))
477                else:
478                    errors += 1
479                    context_stack.append(('other', None))
480
481            elif kw == 'host':
482                if len(pending) >= 2:
483                    host_info: dict[str, Any] = {
484                        'mac': None,
485                        'ip': None,
486                        'parent_subnet': _find_parent_subnet(),
487                    }
488                    context_stack.append(('host', host_info))
489                else:
490                    errors += 1
491                    context_stack.append(('other', None))
492
493            elif kw in ('shared-network', 'group', 'class', 'subclass',
494                        'pool', 'failover', 'peer', 'key', 'zone',
495                        'on', 'if', 'elsif', 'else'):
496                context_stack.append(('other', None))
497
498            elif not pending:
499                # Bare { with no preceding keyword
500                errors += 1
501                context_stack.append(('other', None))
502
503            else:
504                errors += 1
505                context_stack.append(('other', None))
506
507            pending = []
508
509        elif tok == '}':
510            if len(context_stack) <= 1:
511                errors += 1   # unmatched closing brace
512                pending = []
513                continue
514
515            kind, data = context_stack.pop()
516
517            if kind == 'host' and data is not None:
518                mac = data.get('mac')
519                ip = data.get('ip')
520                parent = data.get('parent_subnet')
521                if mac and ip:
522                    all_hosts.append((parent, mac, ip))
523
524            pending = []
525
526        else:
527            pending.append(tok)
528
529    # Unfinished statement at end of file (missing semicolon)
530    if pending:
531        errors += 1
532
533    # Unclosed blocks (missing closing braces)
534    errors += max(0, len(context_stack) - 1)
535
536    # ------------------------------------------------------------------ #
537    # Assign collected hosts to their subnets                             #
538    # ------------------------------------------------------------------ #
539    # Build (IPv4Network, net_addr_str) pairs for membership tests
540    subnet_nets: list[tuple[IPv4Network, str]] = []
541    for net_addr, subnet_data in subnets.items():
542        netmask = subnet_data.get('netmask', '')
543        try:
544            subnet_nets.append((IPv4Network(f'{net_addr}/{netmask}', strict=False), net_addr))
545        except ValueError:
546            pass
547
548    for parent, mac, ip in all_hosts:
549        # If the host was declared directly inside a subnet block, use that subnet.
550        if parent is not None and parent in subnets:
551            subnets[parent]['fixed-addresses'][mac] = ip
552            continue
553        # Otherwise resolve by checking which subnet the fixed-address belongs to.
554        try:
555            host_ip = IPv4Address(ip)
556        except ValueError:
557            continue  # ip is a hostname — can't resolve to a subnet here
558        for net, net_addr in subnet_nets:
559            if host_ip in net:
560                subnets[net_addr]['fixed-addresses'][mac] = ip
561                break
562
563    # ------------------------------------------------------------------ #
564    # Build effective per-subnet parameters (global ← subnet overrides)  #
565    # ------------------------------------------------------------------ #
566    effective_subnets: dict[str, dict[str, Any]] = {}
567    for net_addr, subnet_data in subnets.items():
568        fixed = subnet_data.pop('fixed-addresses')
569        effective: dict[str, Any] = {**global_params, **subnet_data}
570        effective['fixed-addresses'] = fixed
571        effective_subnets[net_addr] = effective
572
573    return {
574        'errors': errors,
575        'global_parameters': global_params,
576        'subnets': effective_subnets,
577    }
578
579
580# ---------------------------------------------------------------------------
581# Running dhcpd (process command line)
582# ---------------------------------------------------------------------------
583
584def _parse_dhcpd_cmdlines(output: str) -> list[str] | None:
585    """Interfaces of the running IPv4 dhcpd, from one command line per process.
586
587    *output* holds one line per dhcpd process (arguments separated by spaces).  The
588    DHCPv6 process (``-6``) is ignored.  Returns ``None`` when no IPv4 dhcpd is running,
589    ``["*"]`` when it listens on every interface.
590    """
591    for line in output.splitlines():
592        tokens = line.split()
593        if not tokens or not tokens[0].endswith('dhcpd') or '-6' in tokens:
594            continue
595        return _parse_dhcpd_interfaces('\n'.join(tokens))
596    return None
597
598
599def get_dhcpd_interfaces(grade: Grade0, machine: str, step: int = 1) -> list[str] | None:
600    """Return the interfaces the running ``dhcpd`` (IPv4) of *machine* listens on.
601
602    ``None`` when dhcpd is not running, ``["*"]`` when it was started without interface.
603    The process table is read directly: the Debian init script (and the systemd unit
604    generated from it) can report a failed service while ``dhcpd -4`` is running.
605    """
606    output, _ = grade.test(
607        machine,
608        r"for p in $(pidof dhcpd); do tr '\000' ' ' < /proc/$p/cmdline; echo; done",
609        step=step, allow_error=True)
610    return _parse_dhcpd_cmdlines(output)
611
612
613# ---------------------------------------------------------------------------
614# DHCP relay (isc-dhcp-relay)
615# ---------------------------------------------------------------------------
616
617@dataclass
618class DhcpRelayParameters:
619    """Content of /etc/default/isc-dhcp-relay."""
620    servers: list[Any]                                      # SERVERS: addresses of the DHCP servers
621    interfaces: list[str] = field(default_factory=list)     # INTERFACES (-i, both directions); empty = all
622    options: str = ''                                       # OPTIONS, e.g. "-id eth2 -iu eth1"
623
624
625def render_dhcp_relay(relay_params: DhcpRelayParameters) -> str:
626    """Content of /etc/default/isc-dhcp-relay for *relay_params*."""
627    return (f'SERVERS="{" ".join(str(s) for s in relay_params.servers)}"\n'
628            f'INTERFACES="{" ".join(relay_params.interfaces)}"\n'
629            f'OPTIONS="{relay_params.options}"\n')
630
631
632def set_dhcp_relay(net_scheme: NetScheme0, machine: str,
633                   relay_params: DhcpRelayParameters, step: int = 1) -> None:
634    """Write /etc/default/isc-dhcp-relay (content: render_dhcp_relay()) and (re)start the relay
635    on *machine*."""
636    net_scheme.file(machine, '/etc/default/isc-dhcp-relay', render_dhcp_relay(relay_params), step=step)
637    net_scheme.cmd(machine, 'systemctl enable isc-dhcp-relay', step=step)
638    net_scheme.cmd(machine, 'systemctl restart isc-dhcp-relay', step=step)
639
640
641def _parse_dhcrelay_cmdline(cmdline_output: str) -> dict | None:
642    """Parse the dhcrelay command line (one token per line, or space separated).
643
644    Returns ``None`` when *cmdline_output* is not a dhcrelay command line (no running relay), else
645    ``{'servers': [...], 'interfaces': [...], 'upstream': [...], 'downstream': [...]}``
646    where *interfaces* are the ``-i`` (both directions) interfaces, *upstream* the ``-iu``
647    ones (towards the servers) and *downstream* the ``-id`` ones (towards the clients).
648    All three empty means that dhcrelay listens on every interface.
649    """
650    IFACE_FLAGS = {'-i': 'interfaces', '-iu': 'upstream', '-id': 'downstream'}
651    VALUE_FLAGS = {'-p', '-rp', '-c', '-A', '-m', '-U', '-g', '-pf', '-l', '-u', '-s', '-I'}
652    tokens = cmdline_output.split()
653    if not tokens or not tokens[0].endswith('dhcrelay'):
654        return None
655    result = {'servers': [], 'interfaces': [], 'upstream': [], 'downstream': []}
656    i = 1  # tokens[0] is the executable
657    while i < len(tokens):
658        tok = tokens[i]
659        if tok in IFACE_FLAGS:
660            if i + 1 < len(tokens):
661                result[IFACE_FLAGS[tok]].append(tokens[i + 1])
662            i += 2
663        elif tok in VALUE_FLAGS:
664            i += 2
665        elif tok.startswith('-'):
666            i += 1
667        else:
668            result['servers'].append(tok)
669            i += 1
670    return result
671
672
673def check_running_dhcp_relay(grade: Grade0, machine: str, step: int = 1) -> tuple[bool, dict]:
674    """Check whether ``dhcrelay`` is running on *machine*.
675
676    Returns ``(running, args)`` where *args* is the dict of :func:`_parse_dhcrelay_cmdline`
677    (empty lists when not running).  ``systemctl is-active isc-dhcp-relay`` is not used:
678    the init script always exits 0, even when dhcrelay refused to start.
679    """
680    output, _ = grade.test(
681        machine,
682        r"tr '\000' '\n' < /proc/$(pidof -s dhcrelay)/cmdline 2>/dev/null",
683        step=step, allow_error=True)
684    parsed = _parse_dhcrelay_cmdline(output)
685    if parsed is None:
686        return False, {'servers': [], 'interfaces': [], 'upstream': [], 'downstream': []}
687    return True, parsed
688
689
690# ---------------------------------------------------------------------------
691# Failover state and client leases
692# ---------------------------------------------------------------------------
693
694def parse_dhcpd_failover_state(leases_text: str) -> dict[str, dict[str, str]]:
695    """Extract the failover states recorded in a dhcpd.leases file.
696
697    Returns ``{peer_name: {'my_state': ..., 'partner_state': ...}}``; dhcpd appends a new
698    ``failover peer "name" state { ... }`` block at each transition, the last one wins.
699    """
700    states: dict[str, dict[str, str]] = {}
701    current = None
702    for line in leases_text.splitlines():
703        m = re.match(r'\s*failover\s+peer\s+"([^"]+)"\s+state\b', line)
704        if m:
705            current = states[m.group(1)] = {}
706            continue
707        m = re.match(r'\s*(my|partner)\s+state\s+([a-z-]+)', line)
708        if m and current is not None:
709            current[f'{m.group(1)}_state'] = m.group(2)
710    return states
711
712
713def get_dhcp_failover_state(grade: Grade0, machine: str, step: int = 1) -> dict[str, dict[str, str]]:
714    """Return the failover states of the dhcpd of *machine* (see :func:`parse_dhcpd_failover_state`)."""
715    output, _ = grade.test(
716        machine, "grep -A 2 '^failover peer' /var/lib/dhcp/dhcpd.leases",
717        step=step, allow_error=True)
718    return parse_dhcpd_failover_state(output)
719
720
721def parse_dhclient_leases(leases_text: str) -> list[dict]:
722    """Parse dhclient lease files; returns one dict per ``lease { ... }`` block, in file order.
723
724    Keys: ``interface``, ``fixed-address`` and every ``option`` by its name (values as
725    strings, quotes removed), plus ``renew`` / ``rebind`` / ``expire``.
726    """
727    leases = []
728    for m in re.finditer(r'lease\s*\{(.*?)\n\}', leases_text, re.DOTALL):
729        lease: dict[str, str] = {}
730        for statement in m.group(1).split(';'):
731            words = statement.split(None, 1)
732            if len(words) < 2:
733                continue
734            key, value = words[0], words[1].strip()
735            if key == 'option':
736                opt = value.split(None, 1)
737                if len(opt) == 2:
738                    lease[opt[0]] = opt[1].strip().strip('"')
739            else:
740                lease[key] = value.strip('"')
741        leases.append(lease)
742    return leases
743
744
745def get_dhclient_leases(grade: Grade0, machine: str, step: int = 1) -> list[dict]:
746    """Return the leases recorded by dhclient on *machine* (manual runs and ifup), oldest first."""
747    output, _ = grade.test(machine, "cat /var/lib/dhcp/dhclient*.leases 2>/dev/null",
748                           step=step, allow_error=True)
749    return parse_dhclient_leases(output)
750
751
752# ---------------------------------------------------------------------------
753# Classless static routes (option 121, RFC 3442)
754# ---------------------------------------------------------------------------
755
756CLASSLESS_ROUTES_OPTION = 'rfc3442-classless-static-routes'   # the name Debian's dhclient gives it
757# dhcpd does not know option 121 by name: dhcpd.conf has to declare it before giving it a value
758CLASSLESS_ROUTES_DECLARATION = f'option {CLASSLESS_ROUTES_OPTION} code 121 = array of unsigned integer 8;'
759
760
761def render_classless_routes(routes) -> str:
762    """Value of option 121 in dhcpd.conf for *routes*, ``(destination network, router)`` pairs:
763    ``24, 198,51,100, 192,0,2,254, 0, 192,0,2,1`` (prefix length, significant bytes of the
764    destination, router).  The default route is the network ``0.0.0.0/0``."""
765    parts = []
766    for destination, gateway in routes:
767        destination = IPv4Network(destination)
768        length = destination.prefixlen
769        significant = destination.network_address.packed[:(length + 7) // 8]
770        parts.append(str(length))
771        if significant:     # none for the default route
772            parts.append(','.join(str(b) for b in significant))
773        parts.append(','.join(str(b) for b in IPv4Address(gateway).packed))
774    return ', '.join(parts)
775
776
777def parse_classless_routes(value) -> list[tuple[IPv4Network, IPv4Address]]:
778    """Decode a value of option 121 written as integers separated by commas and/or spaces, as in
779    dhcpd.conf and in a dhclient lease file (``24,198,51,100,192,0,2,254,0,192,0,2,1``).
780    Returns the ``(destination, router)`` pairs, or ``[]`` when the value is malformed."""
781    try:
782        numbers = [int(n) for n in re.split(r'[\s,]+', str(value or '').strip()) if n]
783    except ValueError:
784        return []
785    routes, i = [], 0
786    while i < len(numbers):
787        length = numbers[i]
788        size = (length + 7) // 8
789        fields = numbers[i + 1:i + 5 + size]
790        if not 0 <= length <= 32 or len(fields) != size + 4 or any(not 0 <= n <= 255 for n in fields):
791            return []
792        destination = bytes(fields[:size]) + bytes(4 - size)
793        try:
794            routes.append((IPv4Network((destination, length)), IPv4Address(bytes(fields[size:]))))
795        except ValueError:      # host bits set in the destination
796            return []
797        i += 5 + size
798    return routes
799
800
801# ---------------------------------------------------------------------------
802# Active probe (lib/dhcp_probe.py run inside a container)
803# ---------------------------------------------------------------------------
804
805DHCP_PROBE_PATH = '/usr/local/sbin/dhcp_probe.py'
806
807
808@dataclass
809class DhcpProbeReply:
810    """One BOOTREPLY seen by the probe for a given query."""
811    interface: str
812    query: str
813    msg_type: str | None                         # 'OFFER', 'ACK', 'NAK'
814    yiaddr: IPv4Address | None = None
815    giaddr: IPv4Address | None = None            # relay agent address (0.0.0.0 → None)
816    server_id: IPv4Address | None = None         # option 54
817    src_ip: IPv4Address | None = None
818    subnet_mask: IPv4Address | None = None
819    routers: list[IPv4Address] = field(default_factory=list)
820    dns_servers: list[IPv4Address] = field(default_factory=list)
821    domain_name: str | None = None
822    lease_time: int | None = None
823    # option 121 (RFC 3442): (destination, router) pairs, the default route being 0.0.0.0/0
824    classless_routes: list[tuple[IPv4Network, IPv4Address]] = field(default_factory=list)
825    raw: dict = field(default_factory=dict)
826
827
828def install_dhcp_probe(net_scheme: NetScheme0, machine: str, step: int = 1) -> None:
829    """Copy lib/dhcp_probe.py to DHCP_PROBE_PATH on *machine* (usually a hidden one)."""
830    script = Path(__file__).with_name('dhcp_probe.py').read_text()
831    net_scheme.file(machine, DHCP_PROBE_PATH, script, permissions=0o755, step=step)
832
833
834def dhcp_probe_query(query_id: str, chaddr, msg_type: str = 'discover', lease: int | None = None,
835                     requested=None, late: bool = False, hostname: str | None = None) -> dict:
836    """Build one query of a probe spec (see lib/dhcp_probe.py).
837
838    *chaddr* is the client MAC address announced in the query.  ``msg_type='request'``
839    with *requested* sends an INIT-REBOOT DHCPREQUEST (an authoritative server answers
840    DHCPNAK when the address does not belong to the network).  *late* queries are sent
841    after the others: use it for a second DISCOVER of the same *chaddr*.
842    """
843    query = {'id': query_id, 'type': msg_type, 'chaddr': str(chaddr).replace('-', ':').lower()}
844    if lease is not None:
845        query['lease'] = lease
846    if requested is not None:
847        query['requested'] = str(requested)
848    if late:
849        query['late'] = True
850    if hostname:
851        query['hostname'] = hostname
852    return query
853
854
855def dhcp_probe_command(spec: dict) -> str:
856    """Shell command running the probe with *spec*; depends on *spec* only (stable across passes)."""
857    encoded = base64.urlsafe_b64encode(
858        json.dumps(spec, sort_keys=True, separators=(',', ':')).encode()).decode()
859    return f'python3 {DHCP_PROBE_PATH} {encoded}'
860
861
862def _probe_ip(value) -> IPv4Address | None:
863    try:
864        ip = IPv4Address(value)
865    except ValueError:
866        return None
867    return None if int(ip) == 0 else ip
868
869
870def _probe_routes(value) -> list[tuple[IPv4Network, IPv4Address]]:
871    routes = []
872    for item in value if isinstance(value, list) else []:
873        try:
874            destination, gateway = item
875            routes.append((IPv4Network(destination, strict=False), IPv4Address(gateway)))
876        except (TypeError, ValueError):
877            continue
878    return routes
879
880
881def parse_dhcp_probe(output: str) -> dict[str, dict[str, list[DhcpProbeReply]]]:
882    """Parse the JSON printed by the probe into ``{interface: {query_id: [DhcpProbeReply]}}``.
883
884    Returns ``{}`` on empty or malformed output.
885    """
886    try:
887        document = json.loads(output)
888        replies = document['replies']
889    except (ValueError, KeyError, TypeError):
890        return {}
891    result: dict[str, dict[str, list[DhcpProbeReply]]] = {}
892    if not isinstance(replies, dict):
893        return result
894    for iface, queries in replies.items():
895        if not isinstance(queries, dict):
896            continue
897        result[iface] = {}
898        for query_id, items in queries.items():
899            parsed = []
900            for r in items if isinstance(items, list) else []:
901                if not isinstance(r, dict):
902                    continue
903                lease_time = r.get('lease_time')
904                parsed.append(DhcpProbeReply(
905                    interface=iface, query=query_id, msg_type=r.get('msg_type'),
906                    yiaddr=_probe_ip(r.get('yiaddr')), giaddr=_probe_ip(r.get('giaddr')),
907                    server_id=_probe_ip(r.get('server_id')), src_ip=_probe_ip(r.get('src_ip')),
908                    subnet_mask=_probe_ip(r.get('subnet_mask')),
909                    routers=[ip for ip in map(_probe_ip, r.get('routers') or []) if ip],
910                    dns_servers=[ip for ip in map(_probe_ip, r.get('dns_servers') or []) if ip],
911                    domain_name=r.get('domain_name'),
912                    lease_time=lease_time if isinstance(lease_time, int) else None,
913                    classless_routes=_probe_routes(r.get('classless_routes')),
914                    raw=r))
915            result[iface][query_id] = parsed
916    return result
917
918
919def dhcp_probe(grade: Grade0, machine: str, spec: dict, step: int = 1,
920               timeout: int = 30) -> dict[str, dict[str, list[DhcpProbeReply]]]:
921    """Run the probe installed on *machine* and return its replies.
922
923    *spec* is ``{'interfaces': {'eth0': [dhcp_probe_query(...), ...]}, 'wait': 2.5,
924    'late_wait': 1.5}``.  It must be built from lab data only (never from test results)
925    so that the command is identical on every grade pass.
926    """
927    output, _ = grade.test(machine, dhcp_probe_command(spec), step=step, timeout=timeout, allow_error=True)
928    return parse_dhcp_probe(output)