vacuum-wall/lib/system_import.py

"""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 copy import deepcopy
from pathlib import Path
from typing import Any

from lib.common import (
    _APPLY_HASH_KEY,
    _LAST_APPLIED_CONFIG_KEY,
    load_json,
    run,
    save_json,
    stamp_applied,
)
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


def _carry_apply_meta(cfg: dict[str, Any], existing: dict[str, Any]) -> None:
    """Preserve apply bookkeeping when adopting live system state.

    Imported content replaces the declarative config but must not destroy
    the applied-state baseline. When *existing* carries apply meta keys,
    they are copied over so pending-change detection and cancel-all keep
    working against the last-applied baseline. When no baseline exists
    (first import), *cfg* is stamped as applied — the imported content is
    exactly the state the system is currently running.
    """
    if _APPLY_HASH_KEY in existing or _LAST_APPLIED_CONFIG_KEY in existing:
        for key in (_APPLY_HASH_KEY, _LAST_APPLIED_CONFIG_KEY):
            if key in existing:
                cfg[key] = deepcopy(existing[key])
    else:
        stamp_applied(cfg)


# ------------------------------------------------------------------
# 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"
    existing: dict[str, Any] = {}
    if cfg_path.exists():
        existing = load_json(cfg_path)
        if _cfgs_equal(existing, cfg):
            logger.debug("Skipping dnsmasq: config already matches")
            return False
    _carry_apply_meta(cfg, existing)
    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."""
    try:
        exists = WG_CONF.exists()
    except OSError:
        # Parent dir may be unreadable to the daemon user (e.g. /etc/wireguard
        # is 0700). Treat as not present rather than failing the import.
        logger.debug("Skipping wireguard: cannot stat %s", WG_CONF)
        return False
    if not 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"
    existing: dict[str, Any] = {}
    if cfg_path.exists():
        existing = load_json(cfg_path)
        if _cfgs_equal(existing, cfg):
            logger.debug("Skipping wireguard: config already matches")
            return False
    _carry_apply_meta(cfg, existing)
    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.

    Uses a simple state machine: [Interface] section populates the interface
    dict; each [Peer] section accumulates into current_peer until the next
    section header triggers _flush_peer() to commit it.
    """
    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:
        """Flush the current peer dict into the peers map if it has a public key.

        Resets ``current_peer`` and ``current_peer_name`` to ``None``,
        preparing for the next [Peer] section.

        Note:
            Only peers with a ``public_key`` are stored; sections without
            a key (malformed or incomplete) are silently skipped.
        """
        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/*.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("*.network"))
    if not network_files:
        logger.debug("Skipping network: no *.network files")
        return False

    parsed_interfaces: dict[str, dict[str, Any]] = {}
    for nf in network_files:
        # Extract interface name from filename, stripping optional priority prefix:
        # 99-eth0.network -> eth0
        # eth0.network -> eth0
        base = nf.stem
        if "-" in base and base.split("-", 1)[0].isdigit():
            iface_name = base.split("-", 1)[1]
        else:
            iface_name = base
        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.

    State machine: [Match] section is skipped; [Link] keys go to iface["link"];
    [Network] keys go directly on iface. Numbered sections ([Address#N], [Route#N])
    accumulate into cur_addr / cur_route dicts until a section boundary triggers
    _flush() to commit them into the corresponding list.
    """
    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:
        """Commit accumulated address/route dicts into the iface lists."""
        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:
    """Parse a [Link] section key-value pair and set the corresponding config field.

    Maps systemd-networkd Link INI keys to snake_case config keys.
    Boolean keys (ARP, Multicast, etc.) are auto-converted via ``_parse_bool``.

    Args:
        link: Link config dict to populate.
        key: INI key name from the .network file.
        val: Value string from the .network file.
    """
    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:
    """Parse a [Network] section key-value pair and set the corresponding config field.

    Maps systemd-networkd Network INI keys to snake_case config dict keys.
    Comma-separated values (DNS, Domains, etc.) are split into lists.
    Boolean and integer keys are auto-converted.

    Args:
        iface: Interface config dict to populate.
        key: INI key name from the .network file.
        val: Value string from the .network file.
    """
    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.

    Bootstraps config.json on hosts that already have rendered sites
    (repo reinstalled over an existing data/ dir). If the declarative
    config already exists it wins: sites are vacuum-wall's own generated
    output ("do not edit manually") and re-parsing them is lossy — backend
    references get flattened to inline paths, which render empty nginx
    sites and hide domains from the WebUI.
    """
    cfg_path = PROJECT_DIR / "config" / "nginx" / "config.json"
    if cfg_path.exists():
        logger.debug("Skipping nginx: %s already exists", cfg_path)
        return False

    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

    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: dict[str, dict[str, Any]] = {}
    for zone_name, parsed in zones.items():
        if not parsed["interfaces"]:
            continue
        zone_cfg: dict[str, Any] = {
            "interfaces": parsed["interfaces"],
            "services": parsed["services"],
            "masquerade": parsed["masquerade"],
            "rich_rules": [{"rule": r} for r in parsed["rich-rules"]],
            "forward_ports": parsed["forward-ports"],
        }
        # Omit the target key when the live target normalizes to firewalld's
        # implicit "default" so key-absence is the one canonical "unmanaged"
        # notation; keep explicit ACCEPT/DROP/REJECT targets.
        target = _live_target_to_config(parsed["target"])
        if target != "DEFAULT":
            zone_cfg["target"] = target
        zone_configs[zone_name] = zone_cfg

    if not zone_configs:
        logger.debug("Skipping firewall: no zones with interfaces")
        return False

    # Only reached when the config file is absent: the imported zones are
    # exactly what firewalld is running, so stamp them as the applied state.
    save_json(cfg_path, stamp_applied({"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 apply bookkeeping keys."""
    from lib.common import strip_apply_meta

    return strip_apply_meta(a) == strip_apply_meta(b)