"""Networkd/IP configuration module. Reads/writes config/network/config.json, renders .network INI files, and parses networkctl status output for runtime state. """ import ipaddress 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 parse_networkctl_status(output: str) -> dict[str, Any]: """Parse ``networkctl status --all`` output into runtime state dict. Args: output: Raw command output from networkctl status. Returns: Dict mapping interface names to their runtime state including addresses, gateway, DNS, and link state. """ result: dict[str, Any] = {} current_iface: dict[str, Any] | None = None def _is_iface_header(line: str) -> bool: """Check if a line looks like an interface header (digits:name ...).""" colon_idx = line.find(":") if colon_idx < 0: return False header = line[:colon_idx].strip() return bool(header) and header[-1].isdigit() for raw_line in output.splitlines(): stripped = raw_line.strip() if not stripped: continue # Interface header: "1: eth0" or similar if _is_iface_header(raw_line): parts = raw_line.split(":", 1)[1].strip().split() if parts: iface_name = parts[0] current_iface = { "addresses": [], "gateway": None, "dns": [], "mac": None, "state": "unknown", "link": parts[1] if len(parts) > 1 else "unknown", } result[iface_name] = current_iface continue if current_iface is None: continue if stripped.startswith("State:"): current_iface["state"] = stripped.split(":", 1)[1].strip() elif stripped.startswith("Gateway:"): gw = stripped.split(":", 1)[1].strip() if gw and gw.lower() not in ("n/a", ""): current_iface["gateway"] = gw elif stripped.startswith("DNS:"): dns_str = stripped.split(":", 1)[1].strip() if dns_str and dns_str.lower() != "n/a": current_iface["dns"] = [d.strip() for d in dns_str.split() if d.strip()] elif stripped.startswith("Hardware Address:"): current_iface["mac"] = stripped.split(":", 2)[2].strip() elif stripped.startswith("Addresses:"): addr_str = stripped.split(":", 1)[1].strip() if addr_str and addr_str.lower() != "n/a": for tok in addr_str.split(): addr = tok.rstrip(",") if "/" in addr: current_iface["addresses"].append(addr) 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