"""Networkd/IP configuration module. Reads/writes config/network/config.json, renders .network INI files, and parses networkctl JSON output for runtime state. """ import contextlib import ipaddress import json import logging from pathlib import Path from typing import Any from lib.common import load_json, save_json logger = logging.getLogger(__name__) PROJECT_DIR = Path(__file__).resolve().parent.parent CONFIG_DIR = PROJECT_DIR / "config" / "network" CONFIG_FILE = CONFIG_DIR / "config.json" DATA_DIR = PROJECT_DIR / "data" / "networkd" DEFAULT_CONFIG: dict[str, Any] = {"interfaces": {}} KNOWN_INTERFACE_KEYS: set[str] = { "addresses", "ipv6_addresses", "gateway", "ipv6_gateway", "dns", "ipv6_dns", "domains", "ipv6_domains", "dns_default_route", "dhcp", "routes", "bind_carrier", "ignore_carrier_loss", "keep_configuration", "configure_without_carrier", "link_local_addressing", "ipv6_link_local_address_generation_mode", "ipv6_stable_secret_address", "ipv4_ll_start_address", "ipv4_ll_route", "default_route_on_device", "ipv6_hop_limit", "ipv6_retransmission_time_sec", "ipv4_duplicate_address_detection_timeout_sec", "ipv4_reverse_path_filter", "ipv4_accept_local", "ipv4_route_localnet", "ipv4_proxy_arp", "ipv4_proxy_arp_private_vlan", "ipv6_proxy_ndp", "ipv6_proxy_ndp_address", "ipv6_send_ra", "m_pls_routing", "keep_master", "ip_family", "link", "dhcp_client", } __all__ = [ "KNOWN_INTERFACE_KEYS", "collect_upstream_dns", "generate_network_files", "get_config", "infer_dhcp_ranges", "infer_zones", "parse_networkctl_status", "render_network_file", "save_config", ] def get_config() -> dict[str, Any]: """Read network config from config/network/config.json. Returns: Dict with ``interfaces`` mapping interface names to config entries. """ if not CONFIG_FILE.exists(): CONFIG_DIR.mkdir(parents=True, exist_ok=True) save_json(CONFIG_FILE, DEFAULT_CONFIG, indent=2) return load_json(CONFIG_FILE) def save_config(cfg: dict[str, Any]) -> None: """Persist network config to disk. Args: cfg: Full config dict to write. """ CONFIG_DIR.mkdir(parents=True, exist_ok=True) save_json(CONFIG_FILE, cfg, indent=2) # ----------------------------------------------------------------------- # Inline key-value emitters — append to a lines list # ----------------------------------------------------------------------- def _emit_str(lines: list[str], key: str, py_key: str, d: dict[str, Any]) -> None: v = d.get(py_key) if v is not None: lines.append(f"{key}={v}") def _emit_int(lines: list[str], key: str, py_key: str, d: dict[str, Any]) -> None: v = d.get(py_key) if v is not None: lines.append(f"{key}={v}") def _emit_bool(lines: list[str], key: str, py_key: str, d: dict[str, Any]) -> None: v = d.get(py_key) if v is not None: lines.append(f"{key}={'yes' if v else 'no'}") def _emit_bool_opt(lines: list[str], key: str, py_key: str, d: dict[str, Any]) -> None: v = d.get(py_key) if v is not None: lines.append(f"{key}={'yes' if v else 'no'}") def _emit_any(lines: list[str], key: str, py_key: str, d: dict[str, Any]) -> None: v = d.get(py_key) if v is not None: if isinstance(v, bool): lines.append(f"{key}={'yes' if v else 'no'}") else: lines.append(f"{key}={v}") def render_network_file(iface_name: str, cfg_entry: dict[str, Any]) -> str: """Render a .network INI file for an interface. Args: iface_name: Interface name (e.g. "eth0"). cfg_entry: Dict with networkd config keys per the schema in todo.md (addresses, gateway, dns, routes, link, dhcp_client, etc.). Returns: INI content string ready to write as 50-.network file. """ lines: list[str] = [] d = cfg_entry # ------------------------------------------------------------------ # [Match] # ------------------------------------------------------------------ lines.append("[Match]") lines.append(f"Name={iface_name}") lines.append("") # ------------------------------------------------------------------ # [Link] # ------------------------------------------------------------------ link = d.get("link", {}) if link: lines.append("[Link]") _emit_int(lines, "MTUBytes", "mtu_bytes", link) _emit_str(lines, "MACAddress", "mac_address", link) _emit_bool_opt(lines, "ARP", "arp", link) _emit_bool_opt(lines, "Multicast", "multicast", link) _emit_bool_opt(lines, "AllMulticast", "all_multicast", link) _emit_bool_opt(lines, "Promiscuous", "promiscuous", link) _emit_bool(lines, "Unmanaged", "unmanaged", link) _emit_str(lines, "ActivationPolicy", "activation_policy", link) _emit_any(lines, "RequiredForOnline", "required_for_online", link) lines.append("") # ------------------------------------------------------------------ # [Network] # ------------------------------------------------------------------ lines.append("[Network]") _emit_str(lines, "DHCP", "dhcp", d) _emit_str(lines, "Gateway", "gateway", d) _emit_str(lines, "IPv6Gateway", "ipv6_gateway", d) for dns in d.get("dns", []): lines.append(f"DNS={dns}") for dns in d.get("ipv6_dns", []): lines.append(f"IPv6DNS={dns}") for dm in d.get("domains", []): lines.append(f"Domains={dm}") for dm in d.get("ipv6_domains", []): lines.append(f"IPv6Domains={dm}") _emit_bool(lines, "DNSDefaultRoute", "dns_default_route", d) for bc in d.get("bind_carrier", []): lines.append(f"BindCarrier={bc}") _emit_any(lines, "IgnoreCarrierLoss", "ignore_carrier_loss", d) _emit_any(lines, "KeepConfiguration", "keep_configuration", d) _emit_bool(lines, "ConfigureWithoutCarrier", "configure_without_carrier", d) _emit_str(lines, "LinkLocalAddressing", "link_local_addressing", d) _emit_str( lines, "IPv6LinkLocalAddressGenerationMode", "ipv6_link_local_address_generation_mode", d, ) _emit_str(lines, "IPv6StableSecretAddress", "ipv6_stable_secret_address", d) _emit_str(lines, "IPv4LLStartAddress", "ipv4_ll_start_address", d) _emit_bool(lines, "IPv4LLRoute", "ipv4_ll_route", d) _emit_bool(lines, "DefaultRouteOnDevice", "default_route_on_device", d) _emit_int(lines, "IPv6HopLimit", "ipv6_hop_limit", d) _emit_str(lines, "IPv6RetransmissionTimeSec", "ipv6_retransmission_time_sec", d) _emit_str( lines, "IPv4DuplicateAddressDetectionTimeoutSec", "ipv4_duplicate_address_detection_timeout_sec", d, ) _emit_str(lines, "IPv4ReversePathFilter", "ipv4_reverse_path_filter", d) _emit_bool(lines, "IPv4AcceptLocal", "ipv4_accept_local", d) _emit_bool(lines, "IPv4RouteLocalnet", "ipv4_route_localnet", d) _emit_bool(lines, "IPv4ProxyARP", "ipv4_proxy_arp", d) _emit_bool(lines, "IPv4ProxyARPPrivateVLAN", "ipv4_proxy_arp_private_vlan", d) _emit_bool(lines, "IPv6ProxyNDP", "ipv6_proxy_ndp", d) _emit_str(lines, "IPv6ProxyNDPAddress", "ipv6_proxy_ndp_address", d) _emit_bool(lines, "IPv6SendRA", "ipv6_send_ra", d) _emit_bool(lines, "MPLSRouting", "m_pls_routing", d) _emit_bool(lines, "KeepMaster", "keep_master", d) _emit_str(lines, "IPFamily", "ip_family", d) lines.append("") # ------------------------------------------------------------------ # [Address] sections — one per entry # ------------------------------------------------------------------ addresses = d.get("addresses", []) for i, addr in enumerate(addresses): if not isinstance(addr, dict): lines.append("[Address]" if i == 0 else f"[Address#{i}]") lines.append(f"Address={addr}") lines.append("") continue lines.append("[Address]" if i == 0 else f"[Address#{i}]") _emit_str(lines, "Address", "address", addr) _emit_str(lines, "Label", "label", addr) _emit_str(lines, "Scope", "scope", addr) _emit_int(lines, "RouteMetric", "route_metric", addr) _emit_str( lines, "DuplicateAddressDetection", "duplicate_address_detection", addr ) _emit_bool(lines, "ManageTemporaryAddress", "manage_temporary_address", addr) _emit_bool(lines, "AddPrefixRoute", "add_prefix_route", addr) lines.append("") # ------------------------------------------------------------------ # [Address] sections — IPv6 # ------------------------------------------------------------------ ipv6_addrs = d.get("ipv6_addresses", []) offset = len(addresses) for i, addr in enumerate(ipv6_addrs): if not isinstance(addr, dict): lines.append(f"[Address#{offset + i}]") lines.append(f"Address={addr}") lines.append("") continue lines.append(f"[Address#{offset + i}]") _emit_str(lines, "Address", "address", addr) _emit_str(lines, "Label", "label", addr) _emit_str(lines, "Scope", "scope", addr) _emit_int(lines, "RouteMetric", "route_metric", addr) _emit_str( lines, "DuplicateAddressDetection", "duplicate_address_detection", addr ) _emit_bool(lines, "ManageTemporaryAddress", "manage_temporary_address", addr) _emit_bool(lines, "AddPrefixRoute", "add_prefix_route", addr) lines.append("") # ------------------------------------------------------------------ # [Route#N] sections # ------------------------------------------------------------------ routes = d.get("routes", []) for i, route in enumerate(routes): if not isinstance(route, dict): continue lines.append("[Route]" if i == 0 else f"[Route#{i}]") _emit_str(lines, "Destination", "destination", route) _emit_str(lines, "Gateway", "gateway", route) _emit_int(lines, "Metric", "metric", route) _emit_any(lines, "Table", "table", route) _emit_str(lines, "Type", "type", route) _emit_str(lines, "Scope", "scope", route) _emit_bool(lines, "GatewayOnLink", "gateway_on_link", route) _emit_str(lines, "IPv6Preference", "ipv6_preference", route) _emit_int(lines, "InitialCongestionWindow", "initial_congestion_window", route) _emit_int( lines, "InitialAdvertisedReceiveWindow", "initial_advertised_receive_window", route, ) _emit_bool(lines, "QuickAck", "quick_ack", route) _emit_bool(lines, "FastOpenNoCookie", "fast_open_no_cookie", route) _emit_int(lines, "MTUBytes", "mtu_bytes", route) _emit_any(lines, "Protocol", "protocol", route) _emit_int(lines, "NextHop", "next_hop", route) for mpr in route.get("multi_path_route", []): lines.append(f"MultiPathRoute={mpr}") lines.append("") # ------------------------------------------------------------------ # [DHCPv4] / [DHCPv6] # ------------------------------------------------------------------ dhcp_client = d.get("dhcp_client", {}) if dhcp_client: dhcp_mode = d.get("dhcp", "no") render_v4 = dhcp_mode in ("yes", "ipv4") render_v6 = dhcp_mode in ("yes", "ipv6") if not render_v4 and not render_v6: render_v4 = True render_v6 = True if render_v4: lines.append("[DHCPv4]") _emit_str(lines, "Hostname", "hostname", dhcp_client) _emit_any(lines, "DUID", "duid", dhcp_client) _emit_str(lines, "DUIDType", "duid_type", dhcp_client) _emit_any(lines, "DUIDRawData", "duid_raw_data", dhcp_client) _emit_str(lines, "IAID", "iaid", dhcp_client) _emit_any(lines, "ClientIdentifier", "client_identifier", dhcp_client) _emit_bool(lines, "RapidCommit", "rapid_commit", dhcp_client) _emit_bool(lines, "Anonymize", "anonymize", dhcp_client) _emit_bool(lines, "UseDNS", "use_dns", dhcp_client) _emit_bool(lines, "UseNTP", "use_ntp", dhcp_client) _emit_bool(lines, "UseSIP", "use_sip", dhcp_client) _emit_bool(lines, "UseCaptivePortal", "use_captive_portal", dhcp_client) _emit_bool(lines, "UseMTU", "use_mtu", dhcp_client) _emit_bool(lines, "UseHostname", "use_hostname", dhcp_client) _emit_any(lines, "UseDomains", "use_domains", dhcp_client) _emit_bool(lines, "UseRoutes", "use_routes", dhcp_client) _emit_int(lines, "RouteMetric", "route_metric", dhcp_client) _emit_bool(lines, "SendDecline", "send_decline", dhcp_client) _emit_str(lines, "NetLabel", "net_label", dhcp_client) _emit_str(lines, "NFTSet", "nft_set", dhcp_client) _emit_str(lines, "IPServiceType", "ip_service_type", dhcp_client) _emit_int(lines, "SocketPriority", "socket_priority", dhcp_client) _emit_bool(lines, "BOOTP", "bootp", dhcp_client) _emit_str(lines, "Label", "label", dhcp_client) _emit_int(lines, "MaxAttempts", "max_attempts", dhcp_client) _emit_int(lines, "ListenPort", "listen_port", dhcp_client) _emit_int(lines, "ServerPort", "server_port", dhcp_client) _emit_str(lines, "MUDURL", "mud_url", dhcp_client) _emit_str(lines, "BootFilename", "boot_filename", dhcp_client) for opt in dhcp_client.get("send_option", []): if isinstance(opt, dict): lines.append( f"SendOption={opt.get('code', '-')} {opt.get('value', '')}" ) else: lines.append(f"SendOption={opt}") for opt in dhcp_client.get("send_vendor_option", []): if isinstance(opt, dict): lines.append( f"SendVendorOption={opt.get('code', '-')}" f" {opt.get('vendor_code', '')}" f" {opt.get('value', '')}" ) else: lines.append(f"SendVendorOption={opt}") for uc in dhcp_client.get("user_class", []): lines.append(f"UserClass={uc}") _emit_str( lines, "VendorClassIdentifier", "vendor_class_identifier", dhcp_client ) _emit_str(lines, "RequestOptions", "request_options", dhcp_client) lines.append("") if render_v6: lines.append("[DHCPv6]") _emit_bool(lines, "SendHostname", "send_hostname", dhcp_client) _emit_str(lines, "Hostname", "hostname", dhcp_client) _emit_any(lines, "DUID", "duid", dhcp_client) _emit_str(lines, "DUIDType", "duid_type", dhcp_client) _emit_any(lines, "DUIDRawData", "duid_raw_data", dhcp_client) _emit_str(lines, "IAID", "iaid", dhcp_client) _emit_bool(lines, "Anonymize", "anonymize", dhcp_client) _emit_str(lines, "RapidCommit", "rapid_commit", dhcp_client) _emit_str( lines, "PrefixDelegationHint", "prefix_delegation_hint", dhcp_client ) _emit_str( lines, "UnassignedSubnetPolicy", "unassigned_subnet_policy", dhcp_client ) _emit_bool(lines, "UseAddress", "use_address", dhcp_client) _emit_bool(lines, "UseCaptivePortal", "use_captive_portal", dhcp_client) _emit_bool(lines, "UseDelegatedPrefix", "use_delegated_prefix", dhcp_client) _emit_bool(lines, "UseDNS", "use_dns", dhcp_client) _emit_bool(lines, "UseNTP", "use_ntp", dhcp_client) _emit_bool(lines, "UseSIP", "use_sip", dhcp_client) _emit_bool(lines, "UseDNR", "use_dnr", dhcp_client) _emit_bool(lines, "UseHostname", "use_hostname", dhcp_client) _emit_any(lines, "UseDomains", "use_domains", dhcp_client) _emit_bool(lines, "SendRelease", "send_release", dhcp_client) _emit_str(lines, "NetLabel", "net_label", dhcp_client) _emit_str(lines, "NFTSet", "nft_set", dhcp_client) _emit_str(lines, "WithoutRA", "without_ra", dhcp_client) for opt in dhcp_client.get("send_option", []): if isinstance(opt, dict): lines.append( f"SendOption={opt.get('code', '-')} {opt.get('value', '')}" ) else: lines.append(f"SendOption={opt}") for opt in dhcp_client.get("send_vendor_option", []): if isinstance(opt, dict): lines.append( f"SendVendorOption={opt.get('code', '-')}" f" {opt.get('vendor_code', '')}" f" {opt.get('value', '')}" ) else: lines.append(f"SendVendorOption={opt}") for uc in dhcp_client.get("user_class", []): lines.append(f"UserClass={uc}") for vc in dhcp_client.get("vendor_class", []): lines.append(f"VendorClass={vc}") lines.append("") return "\n".join(lines) def _bytes_to_ip(addr_bytes: list[int], family: int) -> str: """Convert networkctl JSON address byte array to string.""" if family == 2: return str(ipaddress.ip_address(bytes(addr_bytes))) return str(ipaddress.IPv6Address(bytes(addr_bytes))) def parse_networkctl_status(output: str) -> dict[str, Any]: """Parse ``networkctl status --json=short --all`` JSON output into runtime state dict. Args: output: JSON command output from networkctl status. Returns: Dict mapping interface names to their runtime state including addresses, gateway, DNS, and link state. """ try: data = json.loads(output) except (json.JSONDecodeError, TypeError): return {} result: dict[str, Any] = {} for iface in data.get("Interfaces", []): name = iface.get("Name") if not name: continue # Addresses addresses = [] for a in iface.get("Addresses", []): try: ip = _bytes_to_ip(a["Address"], a["Family"]) addresses.append(f"{ip}/{a['PrefixLength']}") except (KeyError, ValueError, TypeError): continue # Gateway — find default route (Destination 0.0.0.0/0) gateway = None for route in iface.get("Routes", []): if route.get("Family") != 2: continue dest = route.get("Destination", []) prefix = route.get("DestinationPrefixLength", 32) if len(dest) == 4 and all(d == 0 for d in dest) and prefix == 0: gw_bytes = route.get("Gateway") if gw_bytes: with contextlib.suppress(ValueError, TypeError): gateway = _bytes_to_ip(gw_bytes, 2) break # DNS dns = [] for d in iface.get("DNS", []): try: dns.append(_bytes_to_ip(d["Address"], d["Family"])) except (KeyError, ValueError, TypeError): continue # MAC mac = None hw = iface.get("HardwareAddress") if hw: mac = ":".join(f"{b:02x}" for b in hw) # State state = iface.get("OperationalState") or "unknown" result[name] = { "addresses": addresses, "gateway": gateway, "dns": dns, "mac": mac, "state": state, "link": iface.get("Type", "unknown"), } return result def generate_network_files(cfg: dict[str, Any]) -> dict[str, list[Path]]: """Walk config and write all 99-.network files to data/networkd/. Also removes stale .network files that no longer match config. Args: cfg: Network config dict (from get_config). Returns: Dict with ``generated`` (new/updated files) and ``cleaned`` (removed stale files) path lists. """ DATA_DIR.mkdir(parents=True, exist_ok=True) generated: list[Path] = [] cleaned: list[Path] = [] # Build set of expected filenames expected_names: set[str] = set() interfaces_cfg = cfg.get("interfaces", {}) for iface_name, entry in interfaces_cfg.items(): if not isinstance(entry, dict): continue fname = f"99-{iface_name}.network" expected_names.add(fname) content = render_network_file(iface_name, entry) out_path = DATA_DIR / fname out_path.write_text(content) generated.append(out_path) # Remove stale files from DATA_DIR existing_files: set[str] = { f.name for f in DATA_DIR.iterdir() if f.name.endswith(".network") } for fname in existing_files - expected_names: (DATA_DIR / fname).unlink() cleaned.append(DATA_DIR / fname) if cleaned: logger.info("Cleaned %d stale .network files from %s", len(cleaned), DATA_DIR) logger.info("Generated %d .network files in %s", len(generated), DATA_DIR) return {"generated": generated, "cleaned": cleaned} # --------------------------------------------------------------------------- # TF-8: upstream DNS collection # --------------------------------------------------------------------------- _IS_LOCAL = [ ipaddress.ip_network("127.0.0.0/8"), ipaddress.ip_network("10.0.0.0/8"), ipaddress.ip_network("172.16.0.0/12"), ipaddress.ip_network("192.168.0.0/16"), ipaddress.ip_network("169.254.0.0/16"), ipaddress.ip_network("::1/128"), ipaddress.ip_network("fc00::/7"), ipaddress.ip_network("fe80::/10"), ] def _is_local_dns(addr: str) -> bool: try: ip = ipaddress.ip_address(addr) for net in _IS_LOCAL: if ip.version == net.version and ip in net: return True except ValueError: pass return False def collect_upstream_dns(cfg: dict[str, Any]) -> list[str]: """Collect public DNS servers from networkd config, filtering local ranges. Args: cfg: Network config dict (from get_config). Returns: Deduplicated list of upstream DNS server addresses. """ seen: set[str] = set() result: list[str] = [] for entry in cfg.get("interfaces", {}).values(): if not isinstance(entry, dict): continue for dns in entry.get("dns", []): if not _is_local_dns(dns) and dns not in seen: seen.add(dns) result.append(dns) for dns in entry.get("ipv6_dns", []): if not _is_local_dns(dns) and dns not in seen: seen.add(dns) result.append(dns) return result # --------------------------------------------------------------------------- # TF-9: DHCP range inference # --------------------------------------------------------------------------- def infer_dhcp_ranges(cfg: dict[str, Any]) -> dict[str, dict[str, Any]]: """Infer candidate DHCP ranges from static interface IPs. For each interface with a static IPv4 address, calculates a candidate DHCP range covering the usable addresses in the subnet. Args: cfg: Network config dict (from get_config). Returns: Dict mapping interface name to dict with ``subnet``, ``prefix``, ``start``, and ``end`` keys. Interfaces with no inferrable range are omitted. """ result: dict[str, dict[str, Any]] = {} for name, entry in cfg.get("interfaces", {}).items(): if not isinstance(entry, dict): continue for addr in entry.get("addresses", []): if isinstance(addr, dict): addr = addr.get("address", "") if not addr or "/" not in str(addr): continue try: net = ipaddress.ip_network(addr, strict=False) except ValueError: continue if net.version != 4: continue if net.num_addresses < 4: continue net_addr = net.network_address broadcast = net.broadcast_address _start = net_addr + 100 _end = net_addr + 200 _start = min(_start, broadcast - 1) _end = min(_end, broadcast - 1) if _start > _end: continue result[name] = { "subnet": str(net_addr), "prefix": net.prefixlen, "start": str(_start), "end": str(_end), } break return result # --------------------------------------------------------------------------- # TF-10: firewalld zone inference # --------------------------------------------------------------------------- def _looks_wan(entry: dict[str, Any]) -> bool: """Heuristic: interface has DHCP or public-facing address.""" if entry.get("dhcp") in ("yes", "ipv4"): return True for addr in entry.get("addresses", []): if isinstance(addr, dict): addr = addr.get("address", "") if not addr or "/" not in str(addr): continue try: net = ipaddress.ip_network(addr, strict=False) except ValueError: continue if net.version != 4: continue ga = list(net.hosts()) if ga and not _is_local_dns(str(ga[0])): return True return False def _looks_management(entry: dict[str, Any]) -> bool: """Heuristic: interface has management-subnet addresses.""" return bool(entry.get("routes")) def infer_zones( cfg: dict[str, Any], ) -> dict[str, str]: """Classify network interfaces into firewalld zones. Heuristics: - Interface name contains ``wg`` → ``"wan"`` - DHCP-enabled or public-facing IP → ``"wan"`` - Has explicit routes configured → ``"management"`` - Everything else → ``"lan"`` Args: cfg: Network config dict (from get_config). Returns: Dict mapping interface name to suggested zone name. """ result: dict[str, str] = {} for name, entry in cfg.get("interfaces", {}).items(): if not isinstance(entry, dict): continue if "wg" in name or _looks_wan(entry): result[name] = "wan" elif _looks_management(entry): result[name] = "management" else: result[name] = "lan" return result