feat: add networkd subsystem and fix code review issues
Phase 1-4: Networkd subsystem - lib/network.py: systemd-networkd config renderer (.network INI files) with full schema support: [Match], [Link], [Network], [Address], [Route], [DHCPv4], [DHCPv6] sections. One Address/=DNS= line per value per spec. Route sections use #N suffix per systemd.syntax(7). - lib/network.py: generate_network_files() with 50-<name>.network prefix and stale file cleanup - lib/network.py: collect_upstream_dns() filters local/private DNS - lib/network.py: infer_dhcp_ranges() and infer_zones() helpers - daemon/handlers/network.py: routes for GET/POST /network/interfaces and full apply with DNS upstream sync to dnsmasq - webui/api/network.py: Flask blueprint for /api/network/* endpoints - webui/api: interfaces page updated with IP config inline editing - lib/state.py: networkd collector using parse_networkctl_status() - system/sudoers.d/vacuum-walld: networkctl + systemd-network rules - system/systemd/vacuum-walld.service: ReadWritePaths for /etc/systemd/network - install.sh: ACME email now optional, configured from WebUI - lib/acme.py: get_email() falls back to declarative config Phase 5: Code review fixes - daemon/server.py: path params now win over JSON body and query params in request body merge (prevents config save name override) - daemon/server.py: remove dead 'import re' - daemon/handlers/network.py: replace Path.mkdir() with sudo mkdir for /etc/systemd/network (ProtectSystem=strict compatibility) - system/sudoers.d/vacuum-walld: pin systemctl to specific commands (reload/is-active dnsmasq instead of wildcard) - system/sudoers.d/vacuum-walld: restore !requiretty and section comment - lib/network.py: remove unused _MANAGEMENT_PORTS constant - webui/api/network.py: remove redundant body[\name\] = name in save_interface Tests: 332 passing (110 new/updated), ruff clean
This commit is contained in:
+655
@@ -0,0 +1,655 @@
|
||||
"""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": {}}
|
||||
|
||||
__all__ = [
|
||||
"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-<name>.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": [],
|
||||
"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("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 50-<name>.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"50-{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
|
||||
result[name] = {
|
||||
"subnet": str(net_addr),
|
||||
"prefix": net.prefixlen,
|
||||
"start": str(net_addr + 1),
|
||||
"end": str(broadcast - 1),
|
||||
}
|
||||
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
|
||||
Reference in New Issue
Block a user