lib.net_config

   1import re
   2from collections import deque
   3from typing import List, Tuple, Dict, Any, Literal, TypeAlias
   4from ipaddress import (ip_address, ip_interface, ip_network,
   5                       IPv4Address, IPv4Interface, IPv4Network,
   6                       IPv6Address, IPv6Interface, IPv6Network)
   7
   8from SRE.lib_sre import NetScheme0, Grade0
   9
  10IPAddress: TypeAlias = IPv4Address | IPv6Address
  11IPInterface: TypeAlias = IPv4Interface | IPv6Interface
  12IPNetwork: TypeAlias = IPv4Network | IPv6Network
  13
  14# One interface: (addresses, routes).  Both families may share one entry (dual stack), IPv4
  15# first by convention:
  16#   ([IPv4Interface('10.0.0.1/24'), IPv6Interface('fd00:1::1/64')],
  17#    [(IPv4Network('0.0.0.0/0'), gw4), (IPv6Network('::/0'), gw6)])
  18NetConfigInterface: TypeAlias = (
  19        Tuple[
  20            List[IPInterface],
  21            List[Tuple[IPNetwork, IPAddress | IPInterface]]
  22        ]
  23        | Literal['dhcp']
  24        | None
  25)
  26
  27NetConfigEntry: TypeAlias = List[NetConfigInterface]  # indexed by eth number
  28
  29NetConfig: TypeAlias = Dict[str, NetConfigEntry]  # machine name -> entry
  30
  31SysctlConfig = Dict[str, Any]
  32
  33_IFACE_TYPES = (IPv4Interface, IPv6Interface)
  34_DEFAULT_NETS = {'0.0.0.0/0', '::/0'}
  35_DEFAULT_NET_BY_VERSION = {4: IPv4Network('0.0.0.0/0'), 6: IPv6Network('::/0')}
  36_INET_RE = re.compile(r'^\s+inet\s+(\d+\.\d+\.\d+\.\d+)/(\d+)')
  37# the inet6 line of `ip a` does not end with the interface name: "inet6 fd00::1/64 scope global"
  38_INET6_RE = re.compile(r'^\s+inet6\s+([0-9a-fA-F:.]+)/(\d+)(.*)$')
  39_SCOPE_LINK_RE = re.compile(r'\bscope\s+link\b')
  40_TEMPORARY_RE = re.compile(r'\btemporary\b')
  41# first word of `ip -6 route` lines that are not plain routes
  42_V6_ROUTE_TYPES = {'unreachable', 'blackhole', 'prohibit', 'throw', 'anycast', 'local', 'multicast',
  43                   'broadcast', 'nat'}
  44
  45
  46def _gw_ip(via):
  47    """Bare address of a next hop given either as an Address or as an Interface."""
  48    return via.ip if isinstance(via, _IFACE_TYPES) else via
  49
  50
  51def _skip_inet6_line(line: str, link_local: bool) -> bool:
  52    """True for the inet6 lines that are never part of a configuration: link-local addresses
  53    (unless *link_local*) and temporary (privacy) SLAAC addresses."""
  54    return (not link_local and bool(_SCOPE_LINK_RE.search(line))) or bool(_TEMPORARY_RE.search(line))
  55
  56
  57def get_ip_addresses(grade: Grade0, machine_name: str, step: int = 1, ipv6: bool = False,
  58                     link_local: bool = False) -> Dict[str, List[Tuple[str, int]]]:
  59    """Run 'ip a' on machine_name and parse the output.
  60
  61    Returns a dict mapping interface name -> list of (address, prefix_len).
  62
  63    Only IPv4 addresses are returned unless ``ipv6=True``, which adds the IPv6 addresses
  64    (``inet6`` lines) after the IPv4 ones: global addresses, SLAAC (``dynamic`` /
  65    ``mngtmpaddr``) addresses and ``::1``; link-local addresses (``scope link``) only with
  66    ``link_local=True``; temporary (privacy) addresses never.
  67
  68    Addresses for each interface are sorted by:
  69      1. family (IPv4 before IPv6)
  70      2. prefix length descending
  71      3. IP address lexicographically ascending
  72
  73    Edge cases:
  74      - virtual interfaces (e.g. eth0@if5) -> name stripped to 'eth0'
  75      - non-zero exit code or empty output  -> {}
  76    """
  77    output, code = grade.test(machine_name, 'ip a', step=step)
  78    if code != 0 or not output:
  79        return {}
  80    result: Dict[str, List[Tuple[str, int]]] = {}
  81    current_iface = None
  82    for line in output.splitlines():
  83        m = re.match(r'^\d+:\s+(\S+?)(?:@\S+)?:', line)
  84        if m:
  85            current_iface = m.group(1)
  86            result.setdefault(current_iface, [])
  87            continue
  88        m = _INET_RE.match(line)
  89        if m and current_iface is not None:
  90            result[current_iface].append((m.group(1), int(m.group(2))))
  91            continue
  92        if ipv6 and current_iface is not None:
  93            m = _INET6_RE.match(line)
  94            if m and not _skip_inet6_line(m.group(3), link_local):
  95                result[current_iface].append((m.group(1), int(m.group(2))))
  96    for iface in result:
  97        result[iface].sort(key=lambda x: (6 if ':' in x[0] else 4, -x[1], x[0]))
  98    return result
  99
 100
 101def _parse_ip_route(output: str, v6: bool) -> Dict[Tuple[str, int], Tuple[str, str, int]]:
 102    result: Dict[Tuple[str, int], Tuple[str, str, int]] = {}
 103    for line in output.splitlines():
 104        parts = line.split()
 105        if not parts:
 106            continue
 107        dest = parts[0]
 108        if v6 and dest in _V6_ROUTE_TYPES:
 109            continue
 110        if dest == 'default':
 111            net, mask = ('::', 0) if v6 else ('0.0.0.0', 0)
 112        elif '/' in dest:
 113            net, mask_str = dest.split('/')
 114            mask = int(mask_str)
 115        else:
 116            net, mask = dest, (128 if v6 else 32)
 117        via, dev, metric = '', '', 0
 118        i = 1
 119        while i < len(parts):
 120            if parts[i] == 'via' and i + 1 < len(parts):
 121                via = parts[i + 1]
 122                i += 2
 123            elif parts[i] == 'dev' and i + 1 < len(parts):
 124                dev = parts[i + 1]
 125                i += 2
 126            elif parts[i] == 'metric' and i + 1 < len(parts):
 127                metric = int(parts[i + 1])
 128                i += 2
 129            else:
 130                i += 1
 131        result[(net, mask)] = (via, dev, metric)
 132    return result
 133
 134
 135def get_routes(grade: Grade0, machine_name: str, step: int = 1,
 136               ipv6: bool = False) -> Dict[Tuple[str, int], Tuple[str, str, int]]:
 137    """Run 'ip route' (and 'ip -6 route' with ``ipv6=True``) on machine_name and parse the output.
 138
 139    Returns a dict mapping (network, mask) -> (via, dev, metric).
 140
 141    Edge cases:
 142      - 'default' route              -> key ('0.0.0.0', 0), or ('::', 0) in the IPv6 table
 143      - host route without mask      -> mask 32 (128 in the IPv6 table)
 144      - direct route (no via)        -> via ''
 145      - no explicit metric           -> metric 0
 146      - non-zero exit code or empty output -> {} (IPv4 table; the IPv6 routes are still
 147        returned when the IPv4 command fails)
 148      - IPv6 table: 'unreachable', 'blackhole', ... entries are skipped; the kernel's
 149        connected routes (fd00::/64 dev eth0 proto kernel, fe80::/64) have no via, like
 150        the IPv4 ones, and the default route of a router advertisement is kept
 151        (default via fe80::... proto ra)
 152    """
 153    output, code = grade.test(machine_name, 'ip route', step=step)
 154    if ipv6:
 155        output6, code6 = grade.test(machine_name, 'ip -6 route', step=step)
 156    result = {} if (code != 0 or not output) else _parse_ip_route(output, False)
 157    if ipv6 and code6 == 0 and output6:
 158        result.update(_parse_ip_route(output6, True))
 159    return result
 160
 161
 162def get_sysctl_conf(grade: Grade0, machine_name: str, step: int = 1) -> Dict[str, str]:
 163    """Read /etc/sysctl.conf and all files under /etc/sysctl.d/ and parse kernel parameters.
 164
 165    Returns a dict mapping parameter name -> value (both as strings).
 166
 167    Parsing rules:
 168      - lines starting with '#' or ';' are comments and ignored
 169      - blank lines are ignored
 170      - accepted formats: 'key = value' and 'key=value'
 171      - later definitions override earlier ones (sysctl.d files override sysctl.conf)
 172
 173    Edge cases:
 174      - non-zero exit code from any grade.test() -> {} immediately
 175      - first-pass empty output                  -> {} and continues to next call
 176    """
 177    result: Dict[str, str] = {}
 178
 179    def _parse(text: str) -> None:
 180        for line in text.splitlines():
 181            line = line.strip()
 182            if not line or line[0] in ('#', ';'):
 183                continue
 184            if '=' in line:
 185                key, _, value = line.partition('=')
 186                result[key.strip()] = value.strip()
 187
 188    output, code = grade.test(machine_name, 'cat /etc/sysctl.conf', step=step)
 189    if code != 0:
 190        return {}
 191    _parse(output)
 192
 193    output2, code = grade.test(machine_name, 'cat /etc/sysctl.d/*', step=step)
 194    if code != 0:
 195        return {}
 196    _parse(output2)
 197
 198    return result
 199
 200
 201def get_net_config_entry(grade: Grade0, machine_name: str, step: int = 1, ipv6: bool = False) -> NetConfigEntry:
 202    """Run 'ip a', 'ip route', and 'ip link show' on machine_name and return a NetConfig.
 203
 204    Each entry in the returned list corresponds to one interface (eth0, eth1, …)
 205    in index order.
 206
 207    Entry types:
 208      - Tuple (addresses, routes): statically configured interface.
 209      - 'dhcp': interface whose address(es) were all assigned dynamically (DHCP).
 210      - None: interface present in the kernel but carrying no IP address.
 211
 212    The list covers eth0 … ethN where N is the highest eth index reported by
 213    'ip link show'.  Interfaces beyond the highest present index are not included.
 214
 215    ``ipv6=True`` adds the IPv6 addresses (see get_ip_addresses: no link-local, no
 216    temporary address) and the routes of the IPv6 table to the same entries, IPv4 first.
 217    A SLAAC address counts as dynamic: an interface with a static IPv4 address and a
 218    SLAAC IPv6 address is a tuple holding both, an IPv6-only SLAAC interface is 'dhcp'.
 219    The default route learnt from a router advertisement has a link-local next hop: it
 220    is attached to the interface named by its 'dev'.
 221
 222    Edge cases:
 223      - non-zero exit code or empty output from any command -> []
 224      - interfaces whose name does not match eth\\d+  -> ignored
 225      - routes with an unknown gateway -> attached to the first interface, or
 226        dropped if there are no interfaces
 227    """
 228    raw_addrs = get_ip_addresses(grade, machine_name, step=step, ipv6=ipv6)
 229    raw_routes = get_routes(grade, machine_name, step=step, ipv6=ipv6)
 230
 231    # Re-use the cached 'ip a' output to detect dynamic (DHCP) addresses.
 232    ip_a_out, _ = grade.test(machine_name, 'ip a', step=step)
 233
 234    # Parse which interfaces have *all* their inet addresses marked dynamic.
 235    dynamic_ifaces: set[str] = set()
 236    current_iface: str | None = None
 237    iface_addr_flags: dict[str, list[bool]] = {}  # iface -> [is_dynamic, ...]
 238    addr_line_re = re.compile(r'^\s+inet6?\s+\S+') if ipv6 else re.compile(r'^\s+inet\s+\S+')
 239    for line in (ip_a_out or '').splitlines():
 240        m = re.match(r'^\d+:\s+(\S+?)(?:@\S+)?:', line)
 241        if m:
 242            current_iface = m.group(1)
 243            iface_addr_flags.setdefault(current_iface, [])
 244            continue
 245        m = addr_line_re.match(line)
 246        if m and current_iface is not None:
 247            if ipv6 and line.lstrip().startswith('inet6') and _skip_inet6_line(line, link_local=False):
 248                continue
 249            iface_addr_flags.setdefault(current_iface, []).append('dynamic' in line)
 250    for iface, flags in iface_addr_flags.items():
 251        if flags and all(flags):
 252            dynamic_ifaces.add(iface)
 253
 254    # Discover all present eth<N> interfaces via 'ip link show'.
 255    ip_link_out, link_code = grade.test(machine_name, 'ip link show', step=step)
 256    if link_code != 0 or not ip_link_out:
 257        return []
 258    present_eths: set[str] = set(
 259        re.findall(r'^\d+:\s+(eth\d+)(?:@\S+)?:', ip_link_out, re.MULTILINE)
 260    )
 261    if not present_eths:
 262        return []
 263
 264    max_n = max(int(name[3:]) for name in present_eths)
 265
 266    # Build interface → addresses lookup from raw_addrs (only non-empty entries).
 267    eth_addrs: dict[str, list[tuple[str, int]]] = {
 268        name: addrs
 269        for name, addrs in raw_addrs.items()
 270        if name.startswith('eth') and name[3:].isdigit() and addrs
 271    }
 272
 273    # Collect static eth interfaces in order (for route distribution).
 274    static_eth_ifaces: list[tuple[str, list[IPInterface]]] = []
 275    for n in range(max_n + 1):
 276        name = f'eth{n}'
 277        if name not in present_eths:
 278            continue
 279        if name in eth_addrs and name not in dynamic_ifaces:
 280            ifaces = [ip_interface(f"{addr}/{plen}") for addr, plen in eth_addrs[name]]
 281            if ifaces:
 282                static_eth_ifaces.append((name, ifaces))
 283
 284    # Distribute routes among static interfaces (existing logic).
 285    iface_networks: list[list[IPNetwork]] = [
 286        [iface.network for iface in ifaces]
 287        for _, ifaces in static_eth_ifaces
 288    ]
 289    routes_per_iface: list[list[tuple[IPNetwork, IPAddress]]] = [
 290        [] for _ in static_eth_ifaces
 291    ]
 292    for (net_str, mask), (via, dev, _metric) in raw_routes.items():
 293        if not via:
 294            continue
 295        dest = ip_network(f"{net_str}/{mask}")
 296        gw = ip_address(via)
 297        matched = None
 298        for idx, nets in enumerate(iface_networks):
 299            if any(gw in net for net in nets):  # always False across families
 300                matched = idx
 301                break
 302        if matched is None and ipv6:
 303            # a router advertisement's default route: link-local next hop, in no configured prefix
 304            matched = next((idx for idx, (name, _) in enumerate(static_eth_ifaces) if name == dev), None)
 305        if matched is None:
 306            matched = 0
 307        if routes_per_iface:
 308            routes_per_iface[matched].append((dest, gw))
 309
 310    static_map: dict[str, tuple[list[IPInterface], list[tuple[IPNetwork, IPAddress]]]] = {
 311        name: (ifaces, routes_per_iface[idx])
 312        for idx, (name, ifaces) in enumerate(static_eth_ifaces)
 313    }
 314
 315    # Assemble final NetConfig list.
 316    result: NetConfigEntry = []
 317    for n in range(max_n + 1):
 318        name = f'eth{n}'
 319        if name not in present_eths:
 320            result.append(None)
 321            continue
 322        if name in dynamic_ifaces:
 323            result.append('dhcp')
 324        elif name in static_map:
 325            result.append(static_map[name])
 326        else:
 327            result.append(None)
 328
 329    return result
 330
 331
 332def get_persistent_net_config_entry(grade: Grade0, machine_name: str, step: int = 1,
 333                                    ipv6: bool = False) -> tuple[NetConfigEntry, int]:
 334    """Parse /etc/network/interfaces (+ interfaces.d/) and return (NetConfig, error_count).
 335
 336    Reconstructs the network configuration that would result from
 337    'systemctl networking start' (i.e. ifup -a).
 338
 339    Entry types in the returned NetConfig:
 340      - Tuple (addresses, routes): inet-static stanza for eth<N>.
 341      - 'dhcp': inet-dhcp stanza for eth<N>.
 342      - None: eth<N> mentioned in an auto/allow-hotplug line but with no stanza.
 343
 344    Only eth<N> interfaces are included, in index order.
 345
 346    error_count counts lines that are syntactically wrong or have invalid
 347    values (bad IP/network, unknown stanza keywords, malformed iface lines…).
 348
 349    Parsing rules:
 350      - 'address <ip>[/<prefix>]' — primary address; prefix may be absent
 351      - 'netmask <mask>'          — applied to primary address when it has no prefix
 352      - 'gateway <ip>'            — inserted as a 0.0.0.0/0 route
 353      - 'post-up / up ip addr add <addr> dev <iface>' — extra addresses
 354      - 'post-up / up ip route add <net> via <gw>'    — static routes
 355      - Other post-up/up/down commands are silently ignored
 356
 357    ``ipv6=False`` (default): 'inet6' stanzas are ignored like 'manual' ones (not parsed,
 358    not counted).  ``ipv6=True``: 'iface ethN inet6 static' stanzas are parsed too
 359    ('address' with a prefix or 'netmask <prefixlen>', 'gateway' -> ::/0 route, 'post-up ip
 360    [-6] route add' / 'ip [-6] addr add'); the IPv4 and IPv6 parts of an interface are merged
 361    into one tuple (IPv4 first); 'inet6 dhcp' / 'inet6 auto' count as dynamic, so an
 362    interface that is dynamic in both families is 'dhcp'.
 363    """
 364    out1, code1 = grade.test(machine_name, 'cat /etc/network/interfaces', step=step, allow_error=True)
 365    out2, code2 = grade.test(machine_name, 'cat /etc/network/interfaces.d/*', step=step, allow_error=True)
 366
 367    lines: list[str] = []
 368    include_interfaces_d = False
 369    if code1 == 0 and out1:
 370        lines.extend(out1.splitlines())
 371        for raw_line in out1.splitlines():
 372            stripped = raw_line.strip()
 373            if not stripped or stripped.startswith('#'):
 374                continue
 375            parts = stripped.split()
 376            if (parts[0] in ('source', 'source-directory')
 377                    and len(parts) >= 2
 378                    and '/etc/network/interfaces.d' in parts[1]):
 379                include_interfaces_d = True
 380                break
 381    if include_interfaces_d and code2 == 0 and out2:
 382        lines.extend(out2.splitlines())
 383
 384    errors = 0
 385
 386    # Known stanza-level options that ifup accepts — won't be counted as errors.
 387    _KNOWN_OPTS = {
 388        'address', 'netmask', 'broadcast', 'network', 'gateway',
 389        'metric', 'hwaddress', 'mtu', 'scope',
 390        'pre-up', 'up', 'post-up', 'down', 'pre-down', 'post-down',
 391        'dns-nameservers', 'dns-search', 'dns-domain',
 392        'vlan-raw-device', 'bridge_ports', 'bridge_stp', 'bridge_fd',
 393        'bond-master', 'bond-slaves', 'bond-mode', 'bond-miimon',
 394        'wpa-ssid', 'wpa-psk', 'wpa-conf',
 395    }
 396    # inet6-specific options (ifupdown's inet6.defn)
 397    _KNOWN_OPTS6 = _KNOWN_OPTS | {
 398        'accept_ra', 'autoconf', 'privext', 'dad-attempts', 'dad-interval', 'preferred-lifetime',
 399        'media',
 400    }
 401
 402    def _new_bucket():
 403        return {
 404            'primary': None, 'has_prefix': False,
 405            'netmask': None, 'extras': [],
 406            'routes': [], 'gateway': None,
 407        }
 408
 409    # Accumulate per-interface data keyed by (interface name, family).
 410    iface_data: dict[tuple[str, int], dict] = {}
 411    dhcp_ifaces: set[str] = set()  # interfaces with 'inet dhcp' stanza
 412    dyn6_ifaces: set[str] = set()  # interfaces with 'inet6 dhcp' / 'inet6 auto' stanza (ipv6=True)
 413    auto_ifaces: set[str] = set()  # interfaces seen in auto/allow-hotplug lines
 414    current: tuple[str, int] | None = None  # (name, family) of the active static stanza
 415    in_ignored_stanza: bool = False  # True inside loopback / dhcp / manual stanza
 416
 417    for raw_line in lines:
 418        stripped = raw_line.strip()
 419        if not stripped or stripped.startswith('#'):
 420            continue
 421
 422        parts = stripped.split()
 423        kw = parts[0]
 424
 425        if kw == 'iface':
 426            if len(parts) < 4:
 427                errors += 1
 428                current = None
 429                in_ignored_stanza = False
 430                continue
 431            iface_name, family, method = parts[1], parts[2], parts[3]
 432            if family == 'inet' and method == 'static':
 433                current = (iface_name, 4)
 434                in_ignored_stanza = False
 435                iface_data.setdefault(current, _new_bucket())
 436            elif family == 'inet' and method == 'dhcp':
 437                dhcp_ifaces.add(iface_name)
 438                current = None
 439                in_ignored_stanza = True
 440            elif ipv6 and family == 'inet6' and method == 'static':
 441                current = (iface_name, 6)
 442                in_ignored_stanza = False
 443                iface_data.setdefault(current, _new_bucket())
 444            elif ipv6 and family == 'inet6' and method in ('dhcp', 'auto'):
 445                dyn6_ifaces.add(iface_name)
 446                current = None
 447                in_ignored_stanza = True
 448            else:
 449                current = None  # loopback / manual / other — irrelevant
 450                in_ignored_stanza = True
 451
 452        elif kw in ('auto', 'allow-hotplug', 'allow-auto'):
 453            for iface_name in parts[1:]:
 454                auto_ifaces.add(iface_name)
 455
 456        elif kw in ('source', 'source-directory', 'mapping',
 457                    'no-auto-down', 'no-scripts'):
 458            pass  # Valid top-level directives, nothing to record
 459
 460        else:
 461            # Option line within a stanza (indentation not required by ifupdown).
 462            # If outside any stanza entirely, count as an error.
 463            directive = kw
 464
 465            if current is None:
 466                if not in_ignored_stanza:
 467                    errors += 1
 468                continue
 469
 470            data = iface_data[current]
 471            fam = current[1]
 472            iface_cls = IPv4Interface if fam == 4 else IPv6Interface
 473            addr_cls = IPv4Address if fam == 4 else IPv6Address
 474
 475            if directive == 'address':
 476                if len(parts) < 2:
 477                    errors += 1
 478                else:
 479                    try:
 480                        iface = iface_cls(parts[1])
 481                        data['primary'] = iface
 482                        data['has_prefix'] = '/' in parts[1]
 483                    except ValueError:
 484                        errors += 1
 485
 486            elif directive == 'netmask':
 487                if len(parts) < 2:
 488                    errors += 1
 489                else:
 490                    data['netmask'] = parts[1]
 491
 492            elif directive == 'gateway':
 493                if len(parts) < 2:
 494                    errors += 1
 495                else:
 496                    try:
 497                        data['gateway'] = addr_cls(parts[1])
 498                    except ValueError:
 499                        errors += 1
 500
 501            elif directive in ('post-up', 'up'):
 502                cmd = stripped[len(directive):].strip()
 503                if not ipv6:
 504                    m = re.match(r'ip\s+route\s+add\s+(\S+)\s+via\s+(\S+)', cmd)
 505                    if m:
 506                        try:
 507                            net = IPv4Network(m.group(1), strict=False)
 508                            gw = IPv4Address(m.group(2))
 509                            data['routes'].append((net, gw))
 510                        except ValueError:
 511                            errors += 1
 512                        continue
 513                    m = re.match(r'ip\s+addr\s+add\s+(\S+)\s+dev\s+\S+', cmd)
 514                    if m:
 515                        try:
 516                            data['extras'].append(IPv4Interface(m.group(1)))
 517                        except ValueError:
 518                            errors += 1
 519                else:
 520                    m = re.match(r'ip\s+(?:-[46]\s+)?route\s+add\s+(\S+)\s+via\s+(\S+)', cmd)
 521                    if m:
 522                        try:
 523                            net = ip_network(m.group(1), strict=False)
 524                            gw = ip_address(m.group(2))
 525                            if net.version != gw.version:
 526                                raise ValueError("route and next hop of different families")
 527                            data['routes'].append((net, gw))
 528                        except ValueError:
 529                            errors += 1
 530                        continue
 531                    m = re.match(r'ip\s+(?:-[46]\s+)?addr\s+add\s+(\S+)\s+dev\s+\S+', cmd)
 532                    if m:
 533                        try:
 534                            data['extras'].append(ip_interface(m.group(1)))
 535                        except ValueError:
 536                            errors += 1
 537                # Other post-up / up commands (iptables, arp…) are not errors
 538
 539            elif directive not in (_KNOWN_OPTS6 if fam == 6 else _KNOWN_OPTS):
 540                errors += 1
 541
 542    # Collect all eth<N> interface names we know about, in index order.
 543    all_known = (
 544            {name for name, _ in iface_data} | dhcp_ifaces | dyn6_ifaces |
 545            {n for n in auto_ifaces if n.startswith('eth') and n[3:].isdigit()}
 546    )
 547    eth_names = sorted(
 548        [n for n in all_known if n.startswith('eth') and n[3:].isdigit()],
 549        key=lambda n: int(n[3:])
 550    )
 551
 552    def _finalize(data: dict, fam: int):
 553        """(addresses, routes) of one family bucket, or None without a primary address."""
 554        nonlocal errors
 555        primary = data['primary']
 556        if primary is None:
 557            return None
 558
 559        # Combine bare IP with netmask when no CIDR prefix was given.
 560        if not data['has_prefix'] and data['netmask']:
 561            try:
 562                if fam == 4:
 563                    prefix = IPv4Network(f'0.0.0.0/{data["netmask"]}').prefixlen
 564                    primary = IPv4Interface(f'{primary.ip}/{prefix}')
 565                else:
 566                    prefix = int(data['netmask'])
 567                    primary = IPv6Interface(f'{primary.ip}/{prefix}')  # ValueError beyond 128
 568            except ValueError:
 569                errors += 1
 570
 571        addresses = [primary] + data['extras']
 572
 573        routes = list(data['routes'])
 574        if data['gateway']:
 575            routes.insert(0, (_DEFAULT_NET_BY_VERSION[fam], data['gateway']))
 576
 577        return addresses, routes
 578
 579    result: NetConfigEntry = []
 580    for name in eth_names:
 581        if not ipv6:
 582            if name in dhcp_ifaces:
 583                result.append('dhcp')
 584            elif (name, 4) not in iface_data:
 585                # Mentioned in auto but no inet stanza — unconfigured.
 586                result.append(None)
 587            else:
 588                result.append(_finalize(iface_data[(name, 4)], 4))
 589            continue
 590
 591        v4 = None if (name in dhcp_ifaces or (name, 4) not in iface_data) else _finalize(iface_data[(name, 4)], 4)
 592        v6 = None if (name in dyn6_ifaces or (name, 6) not in iface_data) else _finalize(iface_data[(name, 6)], 6)
 593        if v4 is None and v6 is None:
 594            result.append('dhcp' if (name in dhcp_ifaces or name in dyn6_ifaces) else None)
 595        else:
 596            addresses = (v4[0] if v4 else []) + (v6[0] if v6 else [])
 597            routes = (v4[1] if v4 else []) + (v6[1] if v6 else [])
 598            result.append((addresses, routes))
 599
 600    return result, errors
 601
 602
 603class _AttrDict(dict):
 604    __getattr__ = dict.__getitem__
 605
 606
 607def net_config_entry_family(nc_entry: NetConfigEntry, version: int) -> NetConfigEntry:
 608    """Keep only the addresses and routes of IP *version* (4 or 6) in each tuple of
 609    *nc_entry*; 'dhcp' and None entries are returned unchanged.  Use it to grade the two
 610    families of a dual-stack configuration separately with eval_net_config()."""
 611    result: NetConfigEntry = []
 612    for entry in nc_entry:
 613        if not isinstance(entry, tuple):
 614            result.append(entry)
 615            continue
 616        ifaces, routes = entry
 617        result.append(([i for i in ifaces if i.version == version],
 618                       [(net, gw) for net, gw in routes if net.version == version]))
 619    return result
 620
 621
 622def eval_net_config(grade: Grade0, expected: NetConfigEntry, machine_name: str = None,
 623                    current: NetConfigEntry | None = None, step: int = 1) -> _AttrDict:
 624    """Compare current and expected NetConfig; return a dict with 10 keys:
 625
 626    Existing keys (unchanged semantics — 'dhcp' and None entries are skipped):
 627    - "ips"                   : number of IP addresses that are correct
 628    - "default_route"         : 1 if the default route matches the expected one, 0 otherwise
 629    - "other_routes"          : number of correct non-default static routes
 630    - "wrong_routes"          : number of non-default static routes present in current but not in expected
 631    - "ips_expected"          : number of IP addresses in expected
 632    - "default_route_expected": 1 if expected has a default route, 0 otherwise
 633    - "other_routes_expected" : number of non-default static routes in expected
 634
 635    New keys for DHCP/None tracking:
 636    - "dhcp_interfaces"         : positions where both expected and current are 'dhcp'
 637    - "dhcp_interfaces_expected": count of 'dhcp' entries in expected
 638    - "none_interfaces_expected": count of None entries in expected
 639
 640    If *current* is None, get_net_config() is called first.
 641    IP addresses are compared as interface strings (address + prefix), IPv4 and IPv6 alike.
 642    Routes are compared as (network, gateway) pairs; 0.0.0.0/0 and ::/0 are both default
 643    routes ("default_route" is 1 only when every expected default route is present and no
 644    other one is: split the families with net_config_entry_family() to grade them apart).
 645    """
 646    if current is None:
 647        if machine_name is None:
 648            raise ValueError("eval_net_config: current and machine_name can't be both None")
 649        current = get_net_config_entry(grade, machine_name, step=step)
 650
 651    def _collect_ips(nc: NetConfigEntry) -> set[str]:
 652        result = set()
 653        for entry in nc:
 654            if not isinstance(entry, tuple):
 655                continue
 656            ifaces, _ = entry
 657            for iface in ifaces:
 658                result.add(str(iface))
 659        return result
 660
 661    def _collect_routes(nc: NetConfigEntry) -> set[tuple[str, str]]:
 662        result = set()
 663        for entry in nc:
 664            if not isinstance(entry, tuple):
 665                continue
 666            _, routes = entry
 667            for net, gw in routes:
 668                if gw is None:
 669                    gw_str = ''
 670                elif isinstance(gw, _IFACE_TYPES):
 671                    gw_str = str(gw.ip)
 672                else:
 673                    gw_str = str(gw)
 674                result.add((str(net), gw_str))
 675        return result
 676
 677    exp_ips = _collect_ips(expected)
 678    cur_ips = _collect_ips(current)
 679
 680    exp_routes = _collect_routes(expected)
 681    cur_routes = _collect_routes(current)
 682
 683    exp_default = {r for r in exp_routes if r[0] in _DEFAULT_NETS}
 684    cur_default = {r for r in cur_routes if r[0] in _DEFAULT_NETS}
 685    exp_non_default = {r for r in exp_routes if r[0] not in _DEFAULT_NETS}
 686    cur_non_default = {r for r in cur_routes if r[0] not in _DEFAULT_NETS}
 687
 688    dhcp_match = sum(
 689        1 for e, c in zip(expected, current)
 690        if e == 'dhcp' and c == 'dhcp'
 691    )
 692
 693    return _AttrDict(
 694        ips=len(exp_ips & cur_ips),
 695        default_route=1 if exp_default == cur_default else 0,
 696        other_routes=len(exp_non_default & cur_non_default),
 697        wrong_routes=len(cur_non_default - exp_non_default),
 698        ips_expected=len(exp_ips),
 699        default_route_expected=1 if exp_default else 0,
 700        other_routes_expected=len(exp_non_default),
 701        dhcp_interfaces=dhcp_match,
 702        dhcp_interfaces_expected=sum(1 for e in expected if e == 'dhcp'),
 703        none_interfaces_expected=sum(1 for e in expected if e is None),
 704    )
 705
 706
 707def _route_line(interface: str, net, via) -> str:
 708    """'post-up' line adding a static route in /etc/network/interfaces."""
 709    via_addr = _gw_ip(via)
 710    if net.version == 6:
 711        line = f'    post-up ip -6 route add {net.compressed} via {via_addr}'
 712        if via_addr.is_link_local:
 713            line += f' dev {interface}'
 714        return line
 715    return f'    post-up ip route add {net.compressed} via {via_addr}'
 716
 717
 718def _static_stanza(interface: str, family: int, addresses: list, routes: list, extra_routes: list) -> list[str]:
 719    """Lines of one 'iface ethN inet|inet6 static' stanza.
 720
 721    *routes* are the routes of the stanza's family (the default one becomes 'gateway');
 722    *extra_routes* are routes of the other family when that family has no address on the
 723    interface (they can only go through 'post-up' lines).
 724    """
 725    default_net = _DEFAULT_NET_BY_VERSION[family]
 726    lines = [
 727        f'iface {interface} {"inet" if family == 4 else "inet6"} static',
 728        f'    address {addresses[0]}',
 729    ]
 730    for extra in addresses[1:]:
 731        lines.append(f'    post-up ip addr add {extra} dev {interface}')
 732    for net, via in routes:
 733        if net == default_net:
 734            lines.append(f'    gateway {_gw_ip(via)}')
 735    for net, via in routes:
 736        if net != default_net:
 737            lines.append(_route_line(interface, net, via))
 738    for net, via in extra_routes:
 739        lines.append(_route_line(interface, net, via))
 740    return lines
 741
 742
 743def render_persistent_net_config_entry(nc_entry: NetConfigEntry, machine_name: str = '') -> str:
 744    """Content of /etc/network/interfaces for *nc_entry* (*machine_name* only names the machine
 745    in the error raised for an interface without address).
 746
 747    One 'inet static' stanza per interface with IPv4 addresses and one 'inet6 static' stanza
 748    per interface with IPv6 addresses (both for a dual-stack entry): the first address goes
 749    to 'address', the others to 'post-up ip addr add', the default route to 'gateway' and
 750    the other routes to 'post-up ip [-6] route add'.
 751    """
 752    lines = ['auto lo', 'iface lo inet loopback', '']
 753    for i, entry in enumerate(nc_entry):
 754        interface = f'eth{i}'
 755        if entry is None:
 756            continue
 757        if entry == 'dhcp':
 758            lines += [
 759                f'auto {interface}',
 760                f'iface {interface} inet dhcp',
 761                '',
 762            ]
 763            continue
 764        iface_list, routes = entry
 765        addresses = {fam: [a for a in iface_list if a.version == fam] for fam in (4, 6)}
 766        fam_routes = {fam: [(net, via) for net, via in routes if net.version == fam] for fam in (4, 6)}
 767        families = [fam for fam in (4, 6) if addresses[fam]]
 768        if not families:
 769            raise ValueError(f"set_persistent_net_config_entry: {machine_name} {interface} has no address")
 770        lines.append(f'auto {interface}')
 771        for fam in families:
 772            other = 6 if fam == 4 else 4
 773            extra_routes = fam_routes[other] if other not in families else []
 774            lines += _static_stanza(interface, fam, addresses[fam], fam_routes[fam], extra_routes)
 775            lines.append('')
 776    return '\n'.join(lines)
 777
 778
 779def set_persistent_net_config_entry(net_scheme: NetScheme0, machine_name: str, nc_entry: NetConfigEntry):
 780    """Write /etc/network/interfaces on *machine_name* from *nc_entry* (content:
 781    render_persistent_net_config_entry())."""
 782    net_scheme.file(machine_name, '/etc/network/interfaces',
 783                    render_persistent_net_config_entry(nc_entry, machine_name))
 784
 785
 786def set_net_config_entry(net_scheme: NetScheme0, machine_name: str, nc_entry: NetConfigEntry):
 787    """Apply *nc_entry* on *machine_name* with 'ip link / ip addr add / ip route add' commands
 788    ('dhclient' for a 'dhcp' entry).  IPv4 and IPv6 addresses and routes are handled alike
 789    ('ip' infers the family); an IPv6 route through a link-local next hop gets 'dev ethN'."""
 790    for i, entry in enumerate(nc_entry):
 791        interface = f'eth{i}'
 792        if entry is None:
 793            continue
 794        if entry == 'dhcp':
 795            net_scheme.cmd(machine_name, f'ip link set {interface} up')
 796            net_scheme.cmd(machine_name, f'dhclient {interface}')
 797            continue
 798        iface_list, routes = entry
 799        net_scheme.cmd(machine_name, f'ip link set {interface} up')
 800        for iface in iface_list:
 801            net_scheme.cmd(machine_name, f'ip addr add {iface} dev {interface}')
 802        for net, via in routes:
 803            via_addr = _gw_ip(via)
 804            command = f'ip route add {net.compressed} via {str(via_addr)}'
 805            if net.version == 6 and via_addr.is_link_local:
 806                command += f' dev {interface}'
 807            net_scheme.cmd(machine_name, command)
 808
 809
 810def set_persistent_sysctl(net_scheme: NetScheme0, machine_name: str, sysctl_config: SysctlConfig):
 811    for name, value in sysctl_config.items():
 812        net_scheme.cmd(machine_name, f'echo "{name}={value}" >> /etc/sysctl.conf')
 813    if sysctl_config:
 814        remount_proc_sys(net_scheme, machine_name)
 815        net_scheme.cmd(machine_name, 'sysctl -p')
 816
 817# Docker bind-mounts /proc/sys read-only in a non-privileged container: it must be remounted
 818# before any sysctl write.  A privileged container has no such mount (Docker applies no read-only
 819# path there), and remounting a path that is not a mount point fails with exit status 32
 820# ("mount point not mounted or bad option"): the remount is therefore conditional.  As root,
 821# `test -w` on a sysctl *file* is false on a read-only bind mount (access(2) answers EROFS) and
 822# true on writable procfs; a sysctl *directory* would not do (0555, write refused to root too).
 823REMOUNT_PROC_SYS_CMD = 'test -w /proc/sys/net/ipv4/ip_forward || mount -o rw,remount /proc/sys'
 824
 825
 826def remount_proc_sys(net_scheme: NetScheme0, machine_name: str):
 827    """Make ``/proc/sys`` writable on *machine_name* when it is not (once per machine and
 828    NetScheme instance): ``REMOUNT_PROC_SYS_CMD``, a no-op on a privileged machine."""
 829    if not hasattr(net_scheme, "remount_proc_sys_done"):
 830        net_scheme.remount_proc_sys_done = {}
 831    if net_scheme.remount_proc_sys_done.get(machine_name):
 832        return
 833    net_scheme.cmd(machine_name, REMOUNT_PROC_SYS_CMD)
 834    net_scheme.remount_proc_sys_done[machine_name] = True
 835
 836
 837def set_sysctl(net_scheme: NetScheme0, machine_name: str, sysctl_config: SysctlConfig):
 838    if len(sysctl_config) > 0:
 839        remount_proc_sys(net_scheme, machine_name)
 840        for name, value in sysctl_config.items():
 841            net_scheme.cmd(machine_name, f'sysctl -w {name}={value}')
 842
 843
 844def set_ip_forward(net_scheme: NetScheme0, machine_name: str, ip_forward: bool, step: int = 1):
 845    value = 1 if ip_forward else 0
 846    remount_proc_sys(net_scheme, machine_name)
 847    net_scheme.cmd(machine_name, f'sysctl -w net.ipv4.ip_forward={value}', step=step)
 848
 849
 850def get_ip_forward(grade: Grade0, machine_name: str, step: int = 1) -> bool:
 851    output, code = grade.test(machine_name, 'cat /proc/sys/net/ipv4/ip_forward', step=step)
 852    if code != 0:
 853        return False
 854    return output.strip() == '1'
 855
 856
 857def set_ipv6_forward(net_scheme: NetScheme0, machine_name: str, ipv6_forward: bool, step: int = 1):
 858    """Set IPv6 forwarding on *machine_name* (``net.ipv6.conf.all.forwarding``, which the kernel
 859    propagates to every interface).  Kathara starts the machines of an IPv6 lab with forwarding
 860    enabled: call it with ``False`` on the hosts, as set_ip_forward() for IPv4."""
 861    value = 1 if ipv6_forward else 0
 862    remount_proc_sys(net_scheme, machine_name)
 863    net_scheme.cmd(machine_name, f'sysctl -w net.ipv6.conf.all.forwarding={value}', step=step)
 864
 865
 866def get_ipv6_forward(grade: Grade0, machine_name: str, step: int = 1) -> bool:
 867    """IPv6 forwarding state of *machine_name* (``/proc/sys/net/ipv6/conf/all/forwarding``)."""
 868    output, code = grade.test(machine_name, 'cat /proc/sys/net/ipv6/conf/all/forwarding', step=step)
 869    if code != 0:
 870        return False
 871    return output.strip() == '1'
 872
 873
 874def get_sys_parameter_bool(grade: Grade0, machine_name: str, parameter: str, step: int = 1) -> bool | None:
 875    """Read a boolean kernel parameter via sysctl and return its value.
 876
 877    Returns True for '1', False for '0', None if the parameter does not exist or the command fails.
 878    """
 879    value = get_sys_parameter(grade, machine_name, parameter, step=step)
 880    if value is None:
 881        return None
 882    if value == '1':
 883        return True
 884    if value == '0':
 885        return False
 886    return None
 887
 888
 889def get_net_config_from_topology(
 890        net_scheme: NetScheme0,
 891        topology=None,
 892        gateway: str = None,
 893        default_route: IPv4Address | IPv4Interface = IPv4Interface("172.17.0.1/24"),
 894        *,
 895        ipv4: bool = True,
 896        ipv6: bool = False,
 897        default_route6: IPv6Address | IPv6Interface | None = None,
 898) -> NetConfig:
 899    """Build a NetConfig for every machine in the topology so they can all reach each other.
 900
 901    Returns {machine_name: NetConfig}.
 902
 903    Each machine gets:
 904    - One interface entry per network it belongs to, with the IP from net_scheme.data.ips.
 905    - Static routes to every non-directly-connected network (found via BFS through
 906      multi-homed router machines).
 907
 908    If *gateway* is not None:
 909    - Every machine except the gateway gets a default route (0.0.0.0/0) toward the
 910      gateway machine via the appropriate next-hop IP.
 911    - The gateway itself gets a default route via *default_route* (typically the
 912      Docker host bridge, e.g. 172.17.0.1, reachable when the gateway is ``bridged=True``);
 913      ``default_route=None`` gives it none.
 914
 915    The interface indices (eth0, eth1, …) are computed with the same counter logic as
 916    NetScheme0.__init__, so they match the actual Kathara deployment.
 917
 918    IP names follow the convention set by random_ips_from_topology():
 919    - data.ips.m       when machine m is in exactly one network
 920    - data.ips.m_netX  when machine m is in multiple networks
 921
 922    IPv6: ``ipv6=True`` adds, in the same entries, the IPv6 addresses of ``data.ips6`` and
 923    routes to the IPv6 prefixes of ``data.nets6`` (same naming).  A network without a prefix
 924    in ``data.nets6``, or a machine without an address in ``data.ips6`` for a network (a
 925    SLAAC host, an IPv4-only host), simply has no IPv6 part there (no address, no IPv6
 926    route through it).  The gateway gets a ``::/0`` route only when *default_route6* is
 927    given (the Docker bridge has no IPv6 by default).  ``ipv4=False`` builds an IPv6-only
 928    configuration (``data.ips6`` / ``data.nets6`` are then mandatory like their IPv4
 929    counterparts).
 930    """
 931    if not ipv4 and not ipv6:
 932        raise ValueError("get_net_config_from_topology: at least one of ipv4/ipv6 must be True")
 933    if topology is None:
 934        topology = net_scheme.get_topology()
 935    data = net_scheme.data
 936    families = [fam for fam, enabled in ((4, ipv4), (6, ipv6)) if enabled]
 937    ips_containers = {4: data.ips, 6: getattr(data, 'ips6', None)}
 938    nets_containers = {4: data.nets, 6: getattr(data, 'nets6', None)}
 939
 940    # --- 1. Replicate NetScheme0 eth-index assignment ---
 941    # machine_iface: {machine: {net_name: eth_idx}}
 942    machine_iface: dict[str, dict[str, int]] = {}
 943    _ctr: dict[str, int] = {}
 944    for net_name, machines in topology.items():
 945        items = list(machines.items()) if isinstance(machines, dict) else [(m, None) for m in machines]
 946        for mname, iface_spec in items:
 947            iface = iface_spec[0] if isinstance(iface_spec, tuple) else iface_spec
 948            if iface is None:
 949                iface = _ctr.get(mname, 0)
 950            _ctr[mname] = max(_ctr.get(mname, 0), iface) + 1
 951            machine_iface.setdefault(mname, {})[net_name] = iface
 952
 953    machine_nets: dict[str, list[str]] = {m: list(d.keys()) for m, d in machine_iface.items()}
 954
 955    # --- 2. IP lookup (matches random_ips_from_topology naming) ---
 956    # IPv4 (or the only family): a missing entry is an error (AttributeError), as before.
 957    # IPv6 next to IPv4: a missing entry means "no IPv6 there" (SLAAC or IPv4-only host).
 958    def get_ip(machine: str, net_name: str, fam: int = 4) -> IPInterface | None:
 959        container = ips_containers[fam]
 960        attr = machine if len(machine_nets[machine]) == 1 else f"{machine}_{net_name}"
 961        if fam == 6 and ipv4:
 962            return getattr(container, attr, None)
 963        return getattr(container, attr)
 964
 965    def net_has(net_name: str, fam: int) -> bool:
 966        """Whether *net_name* has a prefix of family *fam* (the prefix of the only family is
 967        mandatory, as before)."""
 968        if fam == 4 or not ipv4:
 969            return True
 970        return nets_containers[6] is not None and hasattr(nets_containers[6], net_name)
 971
 972    # --- 3. Network routing graph ---
 973    # Multi-homed machines act as routers; each pair of their networks is an edge.
 974    net_graph: dict[str, list[tuple[str, str]]] = {n: [] for n in topology}
 975    for mname, nets in machine_nets.items():
 976        for i, n1 in enumerate(nets):
 977            for n2 in nets[i + 1:]:
 978                net_graph[n1].append((n2, mname))
 979                net_graph[n2].append((n1, mname))
 980
 981    # --- 4. BFS: find (via_start_net, first_router) from start_nets to a target ---
 982    def _bfs(start_nets: set[str], is_target) -> tuple[str, str] | None:
 983        visited = set(start_nets)
 984        q: deque = deque()
 985        for n in start_nets:
 986            for adj, router in net_graph.get(n, []):
 987                if adj not in visited:
 988                    visited.add(adj)
 989                    q.append((adj, n, router))
 990        while q:
 991            cur, via, router = q.popleft()
 992            if is_target(cur):
 993                return (via, router)
 994            for adj, r in net_graph.get(cur, []):
 995                if adj not in visited:
 996                    visited.add(adj)
 997                    q.append((adj, via, router))
 998        return None
 999
1000    def first_hop_to_net(start_nets: set[str], target_net: str) -> tuple[str, str] | None:
1001        """Return (via_net, router) for the first hop to reach target_net, or None if direct."""
1002        if target_net in start_nets:
1003            return None
1004        return _bfs(start_nets, lambda n: n == target_net)
1005
1006    def first_hop_to_machine(start_nets: set[str], target: str) -> tuple[str, str] | None:
1007        """Return (via_net, router) for the first hop toward target machine.
1008        When directly connected, router is the target machine itself."""
1009        target_nets = set(machine_nets.get(target, []))
1010        shared = start_nets & target_nets
1011        if shared:
1012            return (next(iter(shared)), target)
1013        return _bfs(start_nets, lambda n: n in target_nets)
1014
1015    # --- 5. external default routes as bare addresses (None: no external default route) ---
1016    external_hop = {
1017        4: None if default_route is None else _gw_ip(default_route),
1018        6: None if default_route6 is None else _gw_ip(default_route6),
1019    }
1020
1021    # --- 6. Assemble NetConfig per machine ---
1022    result: dict[str, NetConfigEntry] = {}
1023
1024    for machine, net_to_eth in machine_iface.items():
1025        start_nets = set(machine_nets[machine])
1026
1027        # Collect all candidate routes as (net, eth_idx, next_hop).
1028        route_triples: list[tuple[IPNetwork, int, IPAddress]] = []
1029
1030        for fam in families:
1031            if not any(net_has(n, fam) and get_ip(machine, n, fam) is not None for n in start_nets):
1032                continue  # no address of this family on the machine: no route of this family
1033
1034            # Specific routes to every non-directly-connected network
1035            for target_net in topology:
1036                if target_net in start_nets or not net_has(target_net, fam):
1037                    continue
1038                hop = first_hop_to_net(start_nets, target_net)
1039                if hop is None:
1040                    continue
1041                via_net, router = hop
1042                if not net_has(via_net, fam):
1043                    continue
1044                next_hop = get_ip(router, via_net, fam)
1045                if next_hop is None:
1046                    continue
1047                route_triples.append((
1048                    getattr(nets_containers[fam], target_net),
1049                    net_to_eth[via_net],
1050                    next_hop.ip,
1051                ))
1052
1053            # Default route
1054            if gateway is not None:
1055                default_net = _DEFAULT_NET_BY_VERSION[fam]
1056                if machine == gateway:
1057                    # External default route: on the interface of the network holding the hop when
1058                    # it is a lab network (a machine standing for the outside world), else on the
1059                    # lowest-index interface (the Docker bridge of a bridged gateway)
1060                    if external_hop[fam] is not None:
1061                        hop = external_hop[fam]
1062                        eth = next((net_to_eth[n] for n in sorted(start_nets, key=net_to_eth.get)
1063                                    if net_has(n, fam) and hop in getattr(nets_containers[fam], n)),
1064                                   min(net_to_eth.values()))
1065                        route_triples.append((default_net, eth, hop))
1066                else:
1067                    hop = first_hop_to_machine(start_nets, gateway)
1068                    if hop is not None:
1069                        via_net, router = hop
1070                        next_hop = get_ip(router, via_net, fam) if net_has(via_net, fam) else None
1071                        if next_hop is not None:
1072                            route_triples.append((
1073                                default_net,
1074                                net_to_eth[via_net],
1075                                next_hop.ip,
1076                            ))
1077
1078        # Drop redundant routes: a route (net, hop) is redundant when another route
1079        # (net2, hop2) exists with the same next-hop and net2 is a strict supernet of net
1080        # (i.e. net2 covers net entirely — the less-specific route already handles it).
1081        def _is_redundant(net: IPNetwork, hop: IPAddress) -> bool:
1082            return any(
1083                net2 != net and net2.version == net.version and net.subnet_of(net2) and hop == hop2
1084                for net2, _, hop2 in route_triples
1085            )
1086
1087        eth_routes: dict[int, list] = {eth: [] for eth in net_to_eth.values()}
1088        for net, eth, hop in route_triples:
1089            if not _is_redundant(net, hop):
1090                eth_routes[eth].append((net, hop))
1091
1092        # Build the NetConfig list indexed by eth number
1093        max_eth = max(net_to_eth.values())
1094        nc: NetConfigEntry = [None] * (max_eth + 1)
1095        for net_name, eth_idx in net_to_eth.items():
1096            addresses = [ip for fam in families if net_has(net_name, fam)
1097                         for ip in [get_ip(machine, net_name, fam)] if ip is not None]
1098            nc[eth_idx] = (addresses, eth_routes[eth_idx])
1099        result[machine] = nc
1100
1101    return result
1102
1103
1104def get_sys_parameter(grade: Grade0, machine_name: str, parameter: str, step: int = 1) -> str | None:
1105    """Read a kernel parameter via sysctl and return its value as a string.
1106
1107    Returns None if the parameter does not exist or the command fails.
1108    """
1109    output, code = grade.test(machine_name, f'sysctl -n {parameter}', step=step)
1110    if code != 0:
1111        return None
1112    return output.strip() or None