mteehan
05524f3756
Phase 1 (critical bugs): - Fix firewall import string-to-list bug (system_import.py) - Add rich rules removal in firewall config apply (handlers/firewall.py) Phase 2 (security hardening): - Restrict sudo wildcards to specific paths (sudoers.d/vacuum-walld) - Fix TOCTOU: use /run/vacuum-wall/ for temp files (nginx, dnsmasq, network handlers) - Remove unnecessary sudo from wg genkey/pubkey (handlers/wireguard.py) Phase 3 (validation): - Validate poll intervals > 0 (daemon/server.py) - Restrict sysctl to whitelisted parameters (handlers/network.py) Phase 4 (defensive programming): - Enforce shell=False in run() and run_proc() (lib/common.py) - Track issuance tasks for graceful shutdown (handlers/acme.py) - Add nginx template marker consistency tests (tests/test_system_import.py)
889 lines
28 KiB
Python
889 lines
28 KiB
Python
"""System config import on daemon startup.
|
|
|
|
On first start, vacuum-walld parses each subsystem's live system config
|
|
file and writes the canonical JSON config. This reconciles any drift
|
|
caused by install.sh or manual edits to system files.
|
|
"""
|
|
|
|
import contextlib
|
|
import logging
|
|
import re
|
|
from pathlib import Path
|
|
from typing import Any
|
|
|
|
from lib.common import load_json, run, save_json
|
|
from lib.firewall import _live_target_to_config, _parse_all_zones_output
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
PROJECT_DIR = Path(__file__).resolve().parent.parent
|
|
|
|
|
|
def _safe_int(val: str) -> int | str:
|
|
"""Try to parse *val* as int; return original string on failure."""
|
|
with contextlib.suppress(ValueError):
|
|
return int(val)
|
|
return val
|
|
|
|
|
|
DNSMASQ_CONF = Path("/etc/dnsmasq.d/vacuum-wall.conf")
|
|
WG_CONF = Path("/etc/wireguard/wg0.conf")
|
|
NETWORKD_DIR = Path("/etc/systemd/network")
|
|
NGINX_SITES_DIR = PROJECT_DIR / "data" / "nginx" / "sites-enabled"
|
|
|
|
DNSTART = "# ---- vacuum-wall managed dnsmasq configuration ----"
|
|
DNEND = "# ---- end vacuum-wall config ----"
|
|
|
|
|
|
def import_all() -> list[str]:
|
|
"""Run all subsystem imports. Returns list of subsystems that were updated."""
|
|
updated: list[str] = []
|
|
for fn, name in [
|
|
(import_dnsmasq, "dnsmasq"),
|
|
(import_firewall, "firewall"),
|
|
(import_wireguard, "wireguard"),
|
|
(import_networkd, "network"),
|
|
(import_nginx, "nginx"),
|
|
]:
|
|
try:
|
|
if fn():
|
|
updated.append(name)
|
|
except Exception:
|
|
logger.warning("Import failed for %s", name, exc_info=True)
|
|
return updated
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Dnsmasq
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def import_dnsmasq() -> bool:
|
|
"""Parse /etc/dnsmasq.d/vacuum-wall.conf -> config/dnsmasq/config.json."""
|
|
if not DNSMASQ_CONF.exists():
|
|
logger.debug("Skipping dnsmasq: %s not found", DNSMASQ_CONF)
|
|
return False
|
|
|
|
try:
|
|
text = DNSMASQ_CONF.read_text()
|
|
except OSError:
|
|
logger.debug("Skipping dnsmasq: cannot read %s", DNSMASQ_CONF)
|
|
return False
|
|
|
|
# Extract managed block
|
|
start_idx = text.find(DNSTART)
|
|
end_idx = text.rfind(DNEND)
|
|
if start_idx == -1 or end_idx == -1 or end_idx <= start_idx:
|
|
logger.debug("Skipping dnsmasq: no managed markers found")
|
|
return False
|
|
|
|
block = text[start_idx + len(DNSTART) : end_idx]
|
|
|
|
try:
|
|
cfg = _parse_dnsmasq_block(block)
|
|
except Exception:
|
|
logger.warning("Failed to parse dnsmasq config", exc_info=True)
|
|
return False
|
|
|
|
cfg_path = PROJECT_DIR / "config" / "dnsmasq" / "config.json"
|
|
if cfg_path.exists():
|
|
existing = load_json(cfg_path)
|
|
if _cfgs_equal(existing, cfg):
|
|
logger.debug("Skipping dnsmasq: config already matches")
|
|
return False
|
|
|
|
save_json(cfg_path, cfg)
|
|
summary = f"upstreams={len(cfg.get('dns', {}).get('upstreams', []))}, ranges={len(cfg.get('dhcp', {}).get('ranges', []))}"
|
|
logger.info("Imported dnsmasq config from %s: %s", DNSMASQ_CONF, summary)
|
|
return True
|
|
|
|
|
|
def _parse_dnsmasq_block(block: str) -> dict[str, Any]:
|
|
"""Parse a dnsmasq managed block into JSON config dict."""
|
|
upstreams: list[str] = []
|
|
domains: str | None = None
|
|
ranges: list[dict[str, Any]] = []
|
|
static_leases: list[dict[str, Any]] = []
|
|
custom_records: list[dict[str, Any]] = []
|
|
|
|
for line in block.splitlines():
|
|
line = line.strip()
|
|
if not line or line.startswith("#"):
|
|
continue
|
|
|
|
if line.startswith("server="):
|
|
upstreams.append(line.split("=", 1)[1].strip())
|
|
elif line == "no-resolv":
|
|
upstreams = []
|
|
elif line.startswith("domain="):
|
|
domains = line.split("=", 1)[1].strip()
|
|
elif line.startswith("dhcp-range="):
|
|
rng = _parse_dhcp_range(line.split("=", 1)[1])
|
|
ranges.append(rng)
|
|
elif line.startswith("dhcp-option="):
|
|
target = _attach_dhcp_option(line.split("=", 1)[1], ranges)
|
|
if target:
|
|
_set_dhcp_option_option(target, line.split("=", 1)[1])
|
|
elif line.startswith("dhcp-host="):
|
|
lease = _parse_dhcp_host(line.split("=", 1)[1])
|
|
if lease:
|
|
static_leases.append(lease)
|
|
elif line.startswith("addr/"):
|
|
rec = _parse_addr_directive(line)
|
|
if rec:
|
|
custom_records.append(rec)
|
|
elif line.startswith("host-record="):
|
|
pass # paired with addr/, skip
|
|
|
|
return {
|
|
"dhcp": {
|
|
"ranges": ranges,
|
|
"static_leases": static_leases,
|
|
},
|
|
"dns": {
|
|
"upstreams": upstreams,
|
|
"domain": domains,
|
|
"custom_records": custom_records,
|
|
},
|
|
}
|
|
|
|
|
|
def _parse_dhcp_range(value: str) -> dict[str, Any]:
|
|
"""Parse dhcp-range value into dict."""
|
|
parts = value.split(",")
|
|
if parts[0].startswith("set:"):
|
|
iface = parts[0][4:]
|
|
addr_parts = parts[1:4]
|
|
else:
|
|
iface = None
|
|
addr_parts = parts[0:3]
|
|
|
|
rng: dict[str, Any] = {}
|
|
if iface:
|
|
rng["interface"] = iface
|
|
rng["start"] = addr_parts[0] if len(addr_parts) > 0 else ""
|
|
rng["end"] = addr_parts[1] if len(addr_parts) > 1 else ""
|
|
rng["lease_time"] = addr_parts[2] if len(addr_parts) > 2 else "12h"
|
|
return rng
|
|
|
|
|
|
def _attach_dhcp_option(
|
|
value: str, ranges: list[dict[str, Any]]
|
|
) -> dict[str, Any] | None:
|
|
"""Parse dhcp-option and return the target range to attach to."""
|
|
m = re.match(r"tag:([^,]+),(\d+),(.+)", value)
|
|
if not m:
|
|
return None
|
|
tag_iface = m.group(1)
|
|
for rng in reversed(ranges):
|
|
if rng.get("interface") == tag_iface:
|
|
return rng
|
|
return None
|
|
|
|
|
|
def _set_dhcp_option_option(target: dict[str, Any], value: str) -> None:
|
|
"""Parse dhcp-option value string and set gateway/dns on target range."""
|
|
m = re.match(r"tag:[^,]+,(\d+),(.+)", value)
|
|
if not m:
|
|
return
|
|
option_code = m.group(1)
|
|
option_val = m.group(2).strip()
|
|
if option_code == "3":
|
|
target["gateway"] = option_val
|
|
elif option_code == "6":
|
|
target["dns"] = option_val
|
|
|
|
|
|
def _parse_dhcp_host(value: str) -> dict[str, Any] | None:
|
|
"""Parse dhcp-host value into dict."""
|
|
parts = value.split(",")
|
|
if len(parts) < 2:
|
|
return None
|
|
lease: dict[str, Any] = {
|
|
"mac": parts[0].strip().lower(),
|
|
"ip": parts[1].strip(),
|
|
}
|
|
if len(parts) >= 3:
|
|
lease["hostname"] = parts[2].strip()
|
|
return lease
|
|
|
|
|
|
def _parse_addr_directive(line: str) -> dict[str, Any] | None:
|
|
"""Parse addr/name/addr into dict."""
|
|
parts = line.split("/")
|
|
if len(parts) < 3:
|
|
return None
|
|
return {"name": parts[1].strip(), "address": parts[2].strip()}
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# WireGuard
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def import_wireguard() -> bool:
|
|
"""Parse /etc/wireguard/wg0.conf -> config/wireguard/config.json."""
|
|
if not WG_CONF.exists():
|
|
logger.debug("Skipping wireguard: %s not found", WG_CONF)
|
|
return False
|
|
|
|
try:
|
|
text = WG_CONF.read_text()
|
|
except OSError:
|
|
logger.debug("Skipping wireguard: cannot read %s", WG_CONF)
|
|
return False
|
|
|
|
try:
|
|
cfg = _parse_wireguard_conf(text)
|
|
except Exception:
|
|
logger.warning("Failed to parse wireguard config", exc_info=True)
|
|
return False
|
|
|
|
cfg_path = PROJECT_DIR / "config" / "wireguard" / "config.json"
|
|
if cfg_path.exists():
|
|
existing = load_json(cfg_path)
|
|
if _cfgs_equal(existing, cfg):
|
|
logger.debug("Skipping wireguard: config already matches")
|
|
return False
|
|
|
|
save_json(cfg_path, cfg)
|
|
peer_count = len(cfg.get("peers", {}))
|
|
logger.info("Imported wireguard config from %s: peers=%d", WG_CONF, peer_count)
|
|
return True
|
|
|
|
|
|
def _parse_wireguard_conf(text: str) -> dict[str, Any]:
|
|
"""Parse wg-quick INI format into JSON config dict."""
|
|
interface: dict[str, Any] = {
|
|
"name": "wg0",
|
|
"listen_port": 51820,
|
|
"private_key": "",
|
|
"public_key": "",
|
|
"addresses": ["10.137.0.1/24"],
|
|
"post_up": None,
|
|
"post_down": None,
|
|
}
|
|
peers: dict[str, dict[str, Any]] = {}
|
|
|
|
current_section = None
|
|
current_peer_name = None
|
|
current_peer: dict[str, Any] | None = None
|
|
|
|
def _flush_peer() -> None:
|
|
nonlocal current_peer, current_peer_name
|
|
if (
|
|
current_peer is not None
|
|
and current_peer_name
|
|
and current_peer.get("public_key")
|
|
):
|
|
peers[current_peer_name] = current_peer
|
|
current_peer = None
|
|
current_peer_name = None
|
|
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
|
|
# Section headers — allow trailing content like comments
|
|
m = re.match(r"^\[(\w+)\](.*)?$", stripped)
|
|
if m:
|
|
section_name = m.group(1)
|
|
if section_name == "Interface":
|
|
current_section = "interface"
|
|
_flush_peer()
|
|
elif section_name == "Peer":
|
|
current_section = "peer"
|
|
_flush_peer()
|
|
|
|
# Look for name in comment after [Peer]
|
|
m2 = re.match(r"^\[Peer\]\s*#\s*(.+)", stripped)
|
|
current_peer_name = m2.group(1).strip() if m2 else None
|
|
|
|
current_peer = {}
|
|
else:
|
|
current_section = None
|
|
_flush_peer()
|
|
continue
|
|
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
|
|
if "=" not in stripped:
|
|
continue
|
|
|
|
key, _, val = stripped.partition("=")
|
|
key = key.strip()
|
|
val = val.strip()
|
|
|
|
if current_section == "interface":
|
|
if key == "PrivateKey":
|
|
interface["private_key"] = val
|
|
elif key == "Address":
|
|
interface["addresses"] = [a.strip() for a in val.split(",")]
|
|
elif key == "ListenPort":
|
|
parsed = _safe_int(val)
|
|
if isinstance(parsed, int):
|
|
interface["listen_port"] = parsed
|
|
elif key == "PostUp":
|
|
interface["post_up"] = val
|
|
elif key == "PostDown":
|
|
interface["post_down"] = val
|
|
|
|
elif current_section == "peer" and current_peer is not None:
|
|
if key == "PublicKey":
|
|
current_peer["public_key"] = val
|
|
if current_peer_name is None:
|
|
current_peer_name = val
|
|
elif key == "PresharedKey":
|
|
current_peer["preshared_key"] = val
|
|
elif key == "Endpoint":
|
|
current_peer["endpoint"] = val
|
|
elif key == "AllowedIPs":
|
|
current_peer["allowed_ips"] = [a.strip() for a in val.split(",")]
|
|
elif key == "PersistentKeepalive":
|
|
parsed = _safe_int(val)
|
|
current_peer["persistent_keepalive"] = (
|
|
parsed if isinstance(parsed, int) else None
|
|
)
|
|
|
|
# Flush any remaining peer
|
|
if (
|
|
current_peer is not None
|
|
and current_peer_name
|
|
and current_peer.get("public_key")
|
|
):
|
|
peers[current_peer_name] = current_peer
|
|
|
|
return {
|
|
"interface": interface,
|
|
"peers": peers,
|
|
}
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Networkd
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def import_networkd() -> bool:
|
|
"""Parse /etc/systemd/network/99-*.network -> config/network/config.json."""
|
|
if not NETWORKD_DIR.exists():
|
|
logger.debug("Skipping network: %s not found", NETWORKD_DIR)
|
|
return False
|
|
|
|
network_files = sorted(NETWORKD_DIR.glob("99-*.network"))
|
|
if not network_files:
|
|
logger.debug("Skipping network: no 99-*.network files")
|
|
return False
|
|
|
|
parsed_interfaces: dict[str, dict[str, Any]] = {}
|
|
for nf in network_files:
|
|
# Extract interface name from filename: 99-eth0.network -> eth0
|
|
iface_name = nf.stem[3:] # strip "99-"
|
|
try:
|
|
parsed_interface = _parse_network_file(nf)
|
|
if parsed_interface:
|
|
parsed_interfaces[iface_name] = parsed_interface
|
|
except Exception:
|
|
logger.warning("Failed to parse %s", nf, exc_info=True)
|
|
|
|
if not parsed_interfaces:
|
|
logger.debug("Skipping network: no valid .network files")
|
|
return False
|
|
|
|
cfg_path = PROJECT_DIR / "config" / "network" / "config.json"
|
|
existing: dict[str, Any] = load_json(cfg_path, {"interfaces": {}})
|
|
interfaces_cfg = existing.setdefault("interfaces", {})
|
|
|
|
# Only add/update interfaces with .network files; don't remove interfaces
|
|
# without a file (they may be pending apply).
|
|
changed = False
|
|
for name, entry in parsed_interfaces.items():
|
|
if name not in interfaces_cfg or not _cfgs_equal(interfaces_cfg[name], entry):
|
|
interfaces_cfg[name] = entry
|
|
changed = True
|
|
|
|
if not changed:
|
|
logger.debug("Skipping network: config already matches")
|
|
return False
|
|
|
|
save_json(cfg_path, existing)
|
|
logger.info(
|
|
"Imported network config from %s/*.network: interfaces=%s",
|
|
NETWORKD_DIR,
|
|
", ".join(parsed_interfaces.keys()),
|
|
)
|
|
return True
|
|
|
|
|
|
def _parse_network_file(path: Path) -> dict[str, Any] | None:
|
|
"""Parse a .network INI file into interface config dict."""
|
|
text = path.read_text()
|
|
iface: dict[str, Any] = {}
|
|
cur_section: str | None = None
|
|
cur_addr: dict[str, Any] | None = None
|
|
cur_route: dict[str, Any] | None = None
|
|
|
|
def _flush() -> None:
|
|
nonlocal cur_addr, cur_route
|
|
if cur_addr is not None:
|
|
if "address" in cur_addr and len(cur_addr) == 1:
|
|
iface.setdefault("addresses", []).append(cur_addr["address"])
|
|
elif cur_addr:
|
|
iface.setdefault("addresses", []).append(cur_addr)
|
|
cur_addr = None
|
|
if cur_route is not None and cur_route:
|
|
iface.setdefault("routes", []).append(cur_route)
|
|
cur_route = None
|
|
|
|
for line in text.splitlines():
|
|
stripped = line.strip()
|
|
if not stripped or stripped.startswith("#"):
|
|
continue
|
|
|
|
m = re.match(r"^\[(.+)\]$", stripped)
|
|
if m:
|
|
_flush()
|
|
sec = m.group(1)
|
|
if sec.startswith("Address"):
|
|
cur_section = sec
|
|
cur_addr = {}
|
|
elif sec.startswith("Route"):
|
|
cur_section = sec
|
|
cur_route = {}
|
|
else:
|
|
cur_section = sec
|
|
continue
|
|
|
|
if "=" not in stripped:
|
|
continue
|
|
|
|
key, _, val = stripped.partition("=")
|
|
key = key.strip()
|
|
val = val.strip()
|
|
|
|
if cur_addr is not None:
|
|
cur_addr = _parse_addr_key(cur_addr, key, val)
|
|
continue
|
|
|
|
if cur_route is not None:
|
|
cur_route = _parse_route_key(cur_route, key, val)
|
|
continue
|
|
|
|
_parse_network_section(iface, cur_section, key, val)
|
|
|
|
_flush()
|
|
return iface if iface else None
|
|
|
|
|
|
def _parse_addr_key(entry: dict[str, Any], key: str, val: str) -> dict[str, Any]:
|
|
"""Parse an [Address] key-value pair, return updated entry."""
|
|
if key == "Address":
|
|
entry["address"] = val
|
|
elif key == "Label":
|
|
entry["label"] = val
|
|
elif key == "Scope":
|
|
entry["scope"] = val
|
|
elif key == "RouteMetric":
|
|
parsed = _safe_int(val)
|
|
if isinstance(parsed, int):
|
|
entry["route_metric"] = parsed
|
|
elif key == "DuplicateAddressDetection":
|
|
entry["duplicate_address_detection"] = val
|
|
elif key == "ManageTemporaryAddress":
|
|
entry["manage_temporary_address"] = _parse_bool(val)
|
|
elif key == "AddPrefixRoute":
|
|
entry["add_prefix_route"] = _parse_bool(val)
|
|
return entry
|
|
|
|
|
|
def _parse_route_key(entry: dict[str, Any], key: str, val: str) -> dict[str, Any]:
|
|
"""Parse a [Route] key-value pair, return updated entry."""
|
|
if key == "Destination":
|
|
entry["destination"] = val
|
|
elif key == "Gateway":
|
|
entry["gateway"] = val
|
|
elif key == "Metric":
|
|
parsed = _safe_int(val)
|
|
if isinstance(parsed, int):
|
|
entry["metric"] = parsed
|
|
elif key == "Table":
|
|
entry["table"] = val
|
|
elif key == "Type":
|
|
entry["type"] = val
|
|
elif key == "Scope":
|
|
entry["scope"] = val
|
|
elif key == "GatewayOnLink":
|
|
entry["gateway_on_link"] = _parse_bool(val)
|
|
elif key == "IPv6Preference":
|
|
entry["ipv6_preference"] = val
|
|
elif key == "MTUBytes":
|
|
parsed = _safe_int(val)
|
|
if isinstance(parsed, int):
|
|
entry["mtu_bytes"] = parsed
|
|
return entry
|
|
|
|
|
|
def _parse_network_section(
|
|
iface: dict[str, Any], section: str | None, key: str, val: str
|
|
) -> None:
|
|
"""Parse a [Match]/[Link]/[Network] key-value pair into iface dict."""
|
|
if section == "Match":
|
|
return
|
|
|
|
if section == "Link":
|
|
link = iface.setdefault("link", {})
|
|
_set_link_key(link, key, val)
|
|
return
|
|
|
|
if section == "Network":
|
|
_set_network_key(iface, key, val)
|
|
return
|
|
|
|
|
|
def _set_link_key(link: dict[str, Any], key: str, val: str) -> None:
|
|
if key == "MTUBytes":
|
|
parsed = _safe_int(val)
|
|
if isinstance(parsed, int):
|
|
link["mtu_bytes"] = parsed
|
|
elif key == "MACAddress":
|
|
link["mac_address"] = val
|
|
elif key in ("ARP", "Multicast", "AllMulticast", "Promiscuous", "Unmanaged"):
|
|
pk = {
|
|
"ARP": "arp",
|
|
"Multicast": "multicast",
|
|
"AllMulticast": "all_multicast",
|
|
"Promiscuous": "promiscuous",
|
|
"Unmanaged": "unmanaged",
|
|
}[key]
|
|
link[pk] = _parse_bool(val)
|
|
elif key == "ActivationPolicy":
|
|
link["activation_policy"] = val
|
|
elif key == "RequiredForOnline":
|
|
link["required_for_online"] = val
|
|
|
|
|
|
def _set_network_key(iface: dict[str, Any], key: str, val: str) -> None:
|
|
if key == "DHCP":
|
|
iface["dhcp"] = val
|
|
elif key == "Gateway":
|
|
iface["gateway"] = val
|
|
elif key == "IPv6Gateway":
|
|
iface["ipv6_gateway"] = val
|
|
elif key == "DNS":
|
|
for d in val.split(","):
|
|
d = d.strip()
|
|
if d:
|
|
iface.setdefault("dns", []).append(d)
|
|
elif key == "IPv6DNS":
|
|
for d in val.split(","):
|
|
d = d.strip()
|
|
if d:
|
|
iface.setdefault("ipv6_dns", []).append(d)
|
|
elif key == "Domains":
|
|
for d in val.split(","):
|
|
d = d.strip()
|
|
if d:
|
|
iface.setdefault("domains", []).append(d)
|
|
elif key == "IPv6Domains":
|
|
for d in val.split(","):
|
|
d = d.strip()
|
|
if d:
|
|
iface.setdefault("ipv6_domains", []).append(d)
|
|
elif key == "DNSDefaultRoute":
|
|
iface["dns_default_route"] = _parse_bool(val)
|
|
elif key == "BindCarrier":
|
|
iface.setdefault("bind_carrier", []).append(val)
|
|
elif key == "IgnoreCarrierLoss":
|
|
iface["ignore_carrier_loss"] = val
|
|
elif key == "KeepConfiguration":
|
|
iface["keep_configuration"] = val
|
|
elif key == "ConfigureWithoutCarrier":
|
|
iface["configure_without_carrier"] = _parse_bool(val)
|
|
elif key == "LinkLocalAddressing":
|
|
iface["link_local_addressing"] = val
|
|
elif key == "IPv6LinkLocalAddressGenerationMode":
|
|
iface["ipv6_link_local_address_generation_mode"] = val
|
|
elif key == "IPv6StableSecretAddress":
|
|
iface["ipv6_stable_secret_address"] = val
|
|
elif key == "IPv4LLStartAddress":
|
|
iface["ipv4_ll_start_address"] = val
|
|
elif key == "IPv4LLRoute":
|
|
iface["ipv4_ll_route"] = _parse_bool(val)
|
|
elif key == "DefaultRouteOnDevice":
|
|
iface["default_route_on_device"] = _parse_bool(val)
|
|
elif key == "IPv6HopLimit":
|
|
parsed = _safe_int(val)
|
|
if isinstance(parsed, int):
|
|
iface["ipv6_hop_limit"] = parsed
|
|
elif key == "IPv6RetransmissionTimeSec":
|
|
iface["ipv6_retransmission_time_sec"] = val
|
|
elif key == "IPv4DuplicateAddressDetectionTimeoutSec":
|
|
iface["ipv4_duplicate_address_detection_timeout_sec"] = val
|
|
elif key == "IPv4ReversePathFilter":
|
|
iface["ipv4_reverse_path_filter"] = val
|
|
elif key == "IPv4AcceptLocal":
|
|
iface["ipv4_accept_local"] = _parse_bool(val)
|
|
elif key == "IPv4RouteLocalnet":
|
|
iface["ipv4_route_localnet"] = _parse_bool(val)
|
|
elif key == "IPv4ProxyARP":
|
|
iface["ipv4_proxy_arp"] = _parse_bool(val)
|
|
elif key == "IPv4ProxyARPPrivateVLAN":
|
|
iface["ipv4_proxy_arp_private_vlan"] = _parse_bool(val)
|
|
elif key == "IPv6ProxyNDP":
|
|
iface["ipv6_proxy_ndp"] = _parse_bool(val)
|
|
elif key == "IPv6ProxyNDPAddress":
|
|
iface["ipv6_proxy_ndp_address"] = val
|
|
elif key == "IPv6SendRA":
|
|
iface["ipv6_send_ra"] = _parse_bool(val)
|
|
elif key == "MPLSRouting":
|
|
iface["m_pls_routing"] = _parse_bool(val)
|
|
elif key == "KeepMaster":
|
|
iface["keep_master"] = _parse_bool(val)
|
|
elif key == "IPFamily":
|
|
iface["ip_family"] = val
|
|
|
|
|
|
def _parse_bool(val: str) -> bool | str:
|
|
"""Parse common boolean representations to bool, return as-is otherwise."""
|
|
if val.lower() in ("yes", "true", "1", "on"):
|
|
return True
|
|
if val.lower() in ("no", "false", "0", "off"):
|
|
return False
|
|
return val
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Nginx
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def import_nginx() -> bool:
|
|
"""Parse data/nginx/sites-enabled/*.conf -> config/nginx/config.json."""
|
|
if not NGINX_SITES_DIR.exists():
|
|
logger.debug("Skipping nginx: %s not found", NGINX_SITES_DIR)
|
|
return False
|
|
|
|
conf_files = sorted(NGINX_SITES_DIR.glob("*.conf"))
|
|
sites: dict[str, dict[str, Any]] = {}
|
|
|
|
for cf in conf_files:
|
|
# Skip acme challenge
|
|
if cf.name == "_acme-challenge.conf":
|
|
continue
|
|
|
|
domain = cf.stem
|
|
if not domain:
|
|
continue
|
|
|
|
try:
|
|
parsed = _parse_nginx_site(cf, domain)
|
|
if parsed:
|
|
sites[domain] = parsed
|
|
except Exception:
|
|
logger.warning("Failed to parse nginx site %s", cf, exc_info=True)
|
|
|
|
if not sites:
|
|
logger.debug("Skipping nginx: no valid site files")
|
|
return False
|
|
|
|
cfg_path = PROJECT_DIR / "config" / "nginx" / "config.json"
|
|
existing: dict[str, Any] = load_json(cfg_path, {"domains": {}, "ssl": {}})
|
|
domains_cfg = existing.setdefault("domains", {})
|
|
|
|
changed = False
|
|
for name, entry in sites.items():
|
|
if name not in domains_cfg or not _cfgs_equal(domains_cfg[name], entry):
|
|
domains_cfg[name] = entry
|
|
changed = True
|
|
|
|
if not changed:
|
|
logger.debug("Skipping nginx: config already matches")
|
|
return False
|
|
|
|
save_json(cfg_path, existing)
|
|
logger.info(
|
|
"Imported nginx config from %s/*.conf: domains=%s",
|
|
NGINX_SITES_DIR,
|
|
", ".join(sites.keys()),
|
|
)
|
|
return True
|
|
|
|
|
|
def _parse_nginx_site(path: Path, domain: str) -> dict[str, Any] | None:
|
|
"""Parse a vacuum-wall nginx site conf into domain config dict."""
|
|
text = path.read_text()
|
|
|
|
# Check it's our file
|
|
if "# Auto-generated by Vacuum Wall" not in text:
|
|
logger.debug("Skipping nginx site %s: not vacuum-wall generated", path)
|
|
return None
|
|
|
|
# Determine force_ssl: look for port 80 redirect block
|
|
force_ssl = bool(re.search(r"listen\s+80\b", text))
|
|
|
|
# Parse domain-level auth
|
|
domain_auth = None
|
|
auth_match = re.search(
|
|
r"auth_basic\s+.*;\s*\n\s*auth_basic_user_file\s+(.+?);", text
|
|
)
|
|
if auth_match:
|
|
domain_auth = {"htpasswd": auth_match.group(1).strip()}
|
|
|
|
# Infer cert type
|
|
cert = _infer_cert_type(text, domain)
|
|
|
|
# Parse location blocks
|
|
paths: dict[str, dict[str, Any]] = {}
|
|
_parse_location_blocks(text, paths, domain_auth)
|
|
|
|
entry: dict[str, Any] = {
|
|
"force_ssl": force_ssl,
|
|
"paths": paths,
|
|
}
|
|
if cert:
|
|
entry["cert"] = cert
|
|
if domain_auth:
|
|
entry["auth"] = domain_auth
|
|
return entry
|
|
|
|
|
|
def _infer_cert_type(text: str, domain: str) -> str | None:
|
|
"""Infer cert type from ssl_certificate path."""
|
|
m = re.search(r"ssl_certificate\s+(.+?);", text)
|
|
if not m:
|
|
return None
|
|
|
|
cert_path = m.group(1).strip()
|
|
if "acme" in cert_path:
|
|
return "acme"
|
|
elif "data/certs" in cert_path:
|
|
return "selfsigned"
|
|
else:
|
|
return "file"
|
|
|
|
|
|
def _parse_location_blocks(
|
|
text: str, paths: dict[str, dict[str, Any]], domain_auth: dict[str, Any] | None
|
|
) -> None:
|
|
"""Extract location blocks and their annotations."""
|
|
# Match annotation comment before location block
|
|
# # /path -> host:port (WebSocket)
|
|
# # /path -> host:port
|
|
loc_re = re.compile(
|
|
r"#\s*(/[\S]*)\s*->\s*(\S+?)(?:\s*\(WebSocket\))?\s*\n"
|
|
r"\s*location\s+\1\s*\{",
|
|
re.MULTILINE,
|
|
)
|
|
|
|
for m in loc_re.finditer(text):
|
|
ppath = m.group(1)
|
|
backend_str = m.group(2)
|
|
is_ws = "WebSocket" in m.group(0)
|
|
|
|
host, _, port_str = backend_str.partition(":")
|
|
try:
|
|
port = int(port_str)
|
|
except ValueError:
|
|
port = 80
|
|
|
|
is_websocket = is_ws
|
|
|
|
# Check for auth_basic off inside this location block
|
|
# Find the location block for this path
|
|
block_start = m.end()
|
|
# Find matching closing brace
|
|
depth = 1
|
|
idx = block_start
|
|
while idx < len(text) and depth > 0:
|
|
if text[idx] == "{":
|
|
depth += 1
|
|
elif text[idx] == "}":
|
|
depth -= 1
|
|
idx += 1
|
|
block_text = text[block_start:idx]
|
|
|
|
entry: dict[str, Any] = {
|
|
"backend": {
|
|
"host": host,
|
|
"port": port,
|
|
"proto": "http",
|
|
},
|
|
}
|
|
|
|
if is_websocket:
|
|
entry["is_websocket"] = True
|
|
elif "auth_basic off" in block_text:
|
|
entry["auth"] = None
|
|
|
|
if "is_management" in block_text or (
|
|
"/api" not in ppath and "management" in block_text.lower()
|
|
):
|
|
entry["is_management"] = True
|
|
|
|
# Check for WebSocket path
|
|
if ppath == "/ws":
|
|
entry["is_websocket"] = True
|
|
|
|
paths[ppath] = entry
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Firewall
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def import_firewall() -> bool:
|
|
"""Run firewall-cmd --list-all-zones -> config/firewall/config.json."""
|
|
cfg_path = PROJECT_DIR / "config" / "firewall" / "config.json"
|
|
if cfg_path.exists():
|
|
logger.debug("Skipping firewall: config already exists at %s", cfg_path)
|
|
return False
|
|
|
|
try:
|
|
output = run(["firewall-cmd", "--list-all-zones"], sudo=True)
|
|
except RuntimeError:
|
|
logger.warning("Import failed for firewall: firewall-cmd unavailable")
|
|
return False
|
|
|
|
try:
|
|
zones = _parse_all_zones_output(output)
|
|
except Exception:
|
|
logger.warning("Failed to parse firewall zones", exc_info=True)
|
|
return False
|
|
|
|
zone_configs = {
|
|
zone_name: {
|
|
"target": _live_target_to_config(parsed["target"]),
|
|
"interfaces": parsed["interfaces"],
|
|
"services": parsed["services"],
|
|
"masquerade": parsed["masquerade"],
|
|
"rich_rules": [{"rule": r} for r in parsed["rich-rules"]],
|
|
"forward_ports": parsed["forward-ports"],
|
|
}
|
|
for zone_name, parsed in zones.items()
|
|
if parsed["interfaces"]
|
|
}
|
|
|
|
if not zone_configs:
|
|
logger.debug("Skipping firewall: no zones with interfaces")
|
|
return False
|
|
|
|
save_json(cfg_path, {"zones": zone_configs})
|
|
logger.info(
|
|
"Imported firewall config: zones=%s",
|
|
", ".join(zone_configs.keys()),
|
|
)
|
|
return True
|
|
|
|
|
|
# ------------------------------------------------------------------
|
|
# Helpers
|
|
# ------------------------------------------------------------------
|
|
|
|
|
|
def _cfgs_equal(a: dict[str, Any], b: dict[str, Any]) -> bool:
|
|
"""Compare two configs ignoring _last_applied_hash."""
|
|
a_clean = {k: v for k, v in a.items() if k != "_last_applied_hash"}
|
|
b_clean = {k: v for k, v in b.items() if k != "_last_applied_hash"}
|
|
return a_clean == b_clean
|