vacuum-wall/daemon/collectors/firewall.py

"""Firewall state collector (read-only sudo queries)."""

import contextlib
from typing import Any

from lib import schema
from lib.common import load_json, run, strip_apply_meta
from lib.firewall import (
    _parse_active_zones,
    _parse_all_zones_output,
    get_service_descriptions,
)
from lib.firewall import (
    config_pending as _config_pending,
)
from lib.network import get_config as _network_get_config
from lib.state import (
    PROJECT_DIR,
    _now_iso,
    register_collector,
    register_volatile,
)


def _fp_to_str(fp: dict[str, Any]) -> str:
    """Convert a port-forward dict to a compact string representation.

    Args:
        fp: Port-forward entry containing port and proto keys.

    Returns:
        Comma-separated string of key=value pairs (e.g. ``port=443,proto=tcp``).
    """
    parts = [f"port={fp['port']}", f"proto={fp['proto']}"]
    if "toaddr" in fp:
        parts.append(f"toaddr={fp['toaddr']}")
    if "toport" in fp:
        parts.append(f"toport={fp['toport']}")
    return "/".join(parts)


def _collect_firewall() -> schema.FirewallState:
    """Return the complete current state of firewalld.

    Returns:
        Dict containing firewall zones, interfaces, rules, config, and
        pending changes.
    """
    active_raw = run(["firewall-cmd", "--get-active-zones"], sudo=True)
    active = _parse_active_zones(active_raw)
    default_zone = run(["firewall-cmd", "--get-default-zone"], sudo=True).strip()
    services = run(["firewall-cmd", "--get-services"], sudo=True).split() or []
    link_out = run(["ip", "-o", "link", "show"], sudo=True)
    addr_out = run(["ip", "-o", "addr", "show"], sudo=True)

    iface_map: dict[str, dict[str, Any]] = {}
    for line in link_out.splitlines():
        if not line:
            continue
        parts = line.split()
        if len(parts) < 2:
            continue
        raw_name = parts[1].rstrip(":").split("@")[0]
        iface_state = "UNKNOWN"
        mtu = None
        mac = None
        for i, p in enumerate(parts):
            if p == "state" and i + 1 < len(parts):
                iface_state = parts[i + 1]
            if p == "mtu" and i + 1 < len(parts):
                mtu = int(parts[i + 1])
            if p.startswith("link/ether") and i + 1 < len(parts):
                mac = parts[i + 1]
        iface_map[raw_name] = {
            "name": raw_name,
            "mac": mac,
            "state": iface_state,
            "mtu": mtu,
            "ips": [],
            "ipv6": [],
            "zone": None,
        }

    for line in addr_out.splitlines():
        if not line:
            continue
        parts = line.split()
        if len(parts) < 4:
            continue
        addr_name = parts[1].split("@")[0]
        addr_key = "ipv6" if parts[2] == "inet6" else "ips"
        for entry in iface_map.values():
            if entry["name"] == addr_name:
                entry[addr_key].append(parts[3])
                break

    for zone_name, ifaces in active.items():
        for raw_if in ifaces:
            for entry in iface_map.values():
                if entry["name"] == raw_if:
                    entry["zone"] = zone_name
                    break

    ifaces = list(iface_map.values())

    # Collect all zones in a single call (replaces per-zone loop)
    zones: dict[str, dict[str, Any]] = {}
    try:
        all_zones_raw = run(["firewall-cmd", "--list-all-zones"], sudo=True)
        zones = _parse_all_zones_output(all_zones_raw)
    except Exception:
        pass

    # Load config (strip apply bookkeeping keys, as the other collectors do)
    fw_config_path = PROJECT_DIR / "config" / "firewall" / "config.json"
    config_data = {}
    if fw_config_path.exists():
        with contextlib.suppress(Exception):
            config_data = strip_apply_meta(load_json(fw_config_path))

    # Pending changes
    full_state = {
        "active_zones": active,
        "default_zone": default_zone,
        "interfaces": ifaces,
        "available_services": services,
        "zones": zones,
        "rich_rules": {n: z.get("rich-rules", []) for n, z in zones.items()},
        "timestamp": _now_iso(),
    }
    pending = {}
    with contextlib.suppress(Exception):
        pending = _config_pending(full_state)

    net_cfg: dict[str, Any] = {}
    with contextlib.suppress(Exception):
        net_cfg = _network_get_config()
    covered: set[str] = set()
    for zone_ifaces in active.values():
        covered.update(zone_ifaces)
    for zone in zones.values():
        covered.update(zone.get("interfaces", []))
    uncovered_interfaces = [
        name
        for name in net_cfg.get("interfaces", {})
        if name != "lo" and not name.startswith("wg") and name not in covered
    ]

    return {
        "active_zones": active,
        "default_zone": default_zone,
        "interfaces": ifaces,
        "available_services": services,
        # Parsed from the firewalld service XML definitions; cached per
        # process so the 30s poll does not re-read the files.
        "service_descriptions": get_service_descriptions(),
        "uncovered_interfaces": uncovered_interfaces,
        "zones": zones,
        "rich_rules": {n: z.get("rich-rules", []) for n, z in zones.items()},
        "config": config_data,
        "pending": pending,
        "timestamp": _now_iso(),
    }


register_collector("firewall", _collect_firewall)
register_volatile(
    "firewall",
    frozenset(
        {
            "interfaces[].ips",
            "interfaces[].ipv6",
        }
    ),
)