706 lines
26 KiB
Python
706 lines
26 KiB
Python
"""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-<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": [],
|
|
"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-<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"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
|