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 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