"""Pre-computed state store for vacuum-walld. Collects system state at startup and on demand. Handlers read from the state instead of invoking subprocesses on every request. """ import contextlib import logging import os import re import shutil import subprocess from copy import deepcopy from datetime import UTC, datetime from pathlib import Path from typing import Any, ClassVar from lib.common import load_json, run, run_proc from lib.firewall import ( _parse_active_zones, _parse_zone_output, ) from lib.firewall import ( config_pending as _config_pending, ) logger = logging.getLogger(__name__) PROJECT_DIR = Path(__file__).resolve().parent.parent # --------------------------------------------------------------------------- # State store # --------------------------------------------------------------------------- class State: """In-memory state store keyed by subsystem name. Each subsystem's value is a dict collected from the corresponding ``collect_*`` function. A value of ``None`` means the subsystem has not been populated yet or the last collection failed. """ SUBSYSTEMS: ClassVar[list[str]] = ["firewall", "dnsmasq", "nginx", "acme", "wireguard"] def __init__(self) -> None: self._data: dict[str, dict[str, Any] | None] = { name: None for name in self.SUBSYSTEMS } def get(self, subsystem: str) -> dict[str, Any] | None: return self._data.get(subsystem) def set(self, subsystem: str, data: dict[str, Any] | None) -> None: self._data[subsystem] = data def populate(self, subsystems: list[str] | None = None) -> None: """Collect state for *subsystems* (all if None).""" targets = subsystems or self.SUBSYSTEMS for name in targets: collector = _COLLECTORS.get(name) if collector is None: continue try: self._data[name] = collector() except Exception: logger.warning( "State collection failed for %s, clearing state", name, exc_info=True, ) self._data[name] = None def is_populated(self) -> bool: return all(v is not None for v in self._data.values()) # Singleton state = State() # --------------------------------------------------------------------------- # Collector registry # --------------------------------------------------------------------------- _COLLECTORS: dict[str, Any] = {} def register_collector(subsystem: str, fn: Any) -> Any: _COLLECTORS[subsystem] = fn return fn # --------------------------------------------------------------------------- # Firewall collector # --------------------------------------------------------------------------- def _now_iso() -> str: return datetime.now(UTC).isoformat() def _fp_to_str(fp: dict[str, Any]) -> str: 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() -> dict[str, Any]: """Return the complete current state of firewalld.""" zone_names = run(["firewall-cmd", "--get-zones"], sudo=True).split() active_raw = run(["firewall-cmd", "--get-active-zones"], sudo=True) active = _parse_active_zones(active_raw) 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(":") 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, "display_name": raw_name.partition("@")[0], "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] addr_key = "ipv6" if parts[2] == "inet6" else "ips" for entry in iface_map.values(): if entry["display_name"] == addr_name: entry[addr_key].append(parts[3]) break for zone_name, ifaces in active.items(): for raw_if in ifaces: clean = raw_if.partition("@")[0] for entry in iface_map.values(): if entry["display_name"] == clean or entry["name"] == raw_if: entry["zone"] = zone_name break ifaces = list(iface_map.values()) zones: dict[str, dict[str, Any]] = {} for zn in zone_names: try: zones[zn] = _parse_zone_output( zn, run(["firewall-cmd", f"--zone={zn}", "--list-all"], sudo=True) ) except Exception: continue # Load config fw_config_path = PROJECT_DIR / "config" / "firewall" / "config.json" config_data = {} if fw_config_path.exists(): with contextlib.suppress(Exception): config_data = load_json(fw_config_path) # Pending changes full_state = { "active_zones": active, "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) return { "active_zones": active, "interfaces": ifaces, "available_services": services, "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) # --------------------------------------------------------------------------- # DNSMasq collector # --------------------------------------------------------------------------- def _collect_dnsmasq() -> dict[str, Any]: """Collect dnsmasq status, config, and leases.""" CONFIG_DIR = PROJECT_DIR / "config" / "dnsmasq" CONFIG_PATH = CONFIG_DIR / "config.json" DNSMASQ_CONF = "/etc/dnsmasq.d/vacuum-wall.conf" LEASE_FILE = "/var/lib/dnsmasq/dnsmasq.leases" DEFAULT_CFG: dict[str, Any] = { "dhcp": {"ranges": [], "static_leases": []}, "dns": { "upstreams": ["8.8.8.8", "1.1.1.1"], "domain": None, "custom_records": [], }, } # Load config cfg: dict[str, Any] = {} if CONFIG_PATH.exists(): try: raw = load_json(CONFIG_PATH) if raw: from lib.common import deep_merge cfg = deep_merge(deepcopy(DEFAULT_CFG), raw) else: cfg = deepcopy(DEFAULT_CFG) except Exception: cfg = deepcopy(DEFAULT_CFG) else: cfg = deepcopy(DEFAULT_CFG) # Service status service_active = False try: proc = run_proc(["systemctl", "is-active", "dnsmasq"], sudo=True) service_active = proc.stdout.strip() == "active" except Exception: pass # Leases leases: list[dict[str, Any]] = [] try: result = run_proc(["cat", LEASE_FILE], sudo=True, check=True) for line in result.stdout.splitlines(): line = line.strip() if not line or line.startswith("#"): continue parts = line.split() if len(parts) < 3: continue try: ts = datetime.fromtimestamp(int(parts[0]), tz=UTC) except (ValueError, OSError): ts = None leases.append( { "expires_at": ts, "mac": parts[1], "ip": parts[2], "hostname": parts[3] if len(parts) > 3 else "", "interface": parts[4] if len(parts) > 4 else "", } ) except RuntimeError: pass # Check config file on disk conf_exists = Path(DNSMASQ_CONF).is_file() return { "config": cfg, "status": { "service_active": service_active, "config_file_exists": conf_exists, "active_leases": len(leases), }, "leases": leases, "timestamp": _now_iso(), } register_collector("dnsmasq", _collect_dnsmasq) # --------------------------------------------------------------------------- # Nginx collector # --------------------------------------------------------------------------- def _collect_nginx() -> dict[str, Any]: """Collect nginx config and domains list.""" CONFIG_DIR = PROJECT_DIR / "config" / "nginx" CONFIG_FILE = CONFIG_DIR / "config.json" SITES_DIR = PROJECT_DIR / "data" / "nginx" / "sites-enabled" DEFAULT_SSL: dict[str, Any] = { "protocols": "TLSv1.2 TLSv1.3", "ciphers": ( "ECDHE-ECDSA-AES128-GCM-SHA256:" "ECDHE-RSA-AES128-GCM-SHA256:" "ECDHE-ECDSA-AES256-GCM-SHA384:" "ECDHE-RSA-AES256-GCM-SHA384:" "ECDHE-ECDSA-CHACHA20-POLY1305:" "ECDHE-RSA-CHACHA20-POLY1305" ), "prefer_server_ciphers": False, } default_cfg: dict[str, Any] = { "domains": {}, "management": None, "ssl": deepcopy(DEFAULT_SSL), } cfg = deepcopy(default_cfg) if CONFIG_FILE.exists(): try: raw = load_json(CONFIG_FILE) if raw: from lib.common import deep_merge default_cfg: dict[str, Any] = { "domains": {}, "management": None, "ssl": deepcopy(DEFAULT_SSL), } cfg = deep_merge(default_cfg, raw) if "ssl" not in cfg: cfg["ssl"] = deepcopy(DEFAULT_SSL) except Exception: pass # Build domains list with site existence domains: list[dict[str, Any]] = [] for name, dom in cfg.get("domains", {}).items(): site = SITES_DIR / f"{name}.conf" domains.append( { "domain": name, "backend": dom.get("backend", {}), "online": site.exists() if SITES_DIR.exists() else False, "force_ssl": dom.get("force_ssl", True), } ) return { "config": cfg, "domains": domains, "timestamp": _now_iso(), } register_collector("nginx", _collect_nginx) # --------------------------------------------------------------------------- # ACME collector # --------------------------------------------------------------------------- def _find_acme() -> str: acme_home = PROJECT_DIR / "data" / "acme" candidates = [acme_home / "acme.sh", Path("/usr/local/bin/acme.sh")] for path in candidates: if path.is_file() and os.access(path, os.X_OK): return str(path) acme = shutil.which("acme.sh") if acme: return acme raise FileNotFoundError("acme.sh not found") def _run_acme(args: list[str]) -> str: acme_bin = _find_acme() acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme")) cmd = [acme_bin, "--home", acme_home_env, "--config-home", acme_home_env, *args] _ACME_ENVIRON = { "HOME": str(PROJECT_DIR), "PATH": os.environ.get( "PATH", "/usr/local/sbin:/usr/local/bin:/usr/sbin:/usr/bin:/sbin:/bin" ), } try: result = subprocess.run( cmd, capture_output=True, text=True, timeout=120, env={**os.environ, **_ACME_ENVIRON}, ) except subprocess.TimeoutExpired as exc: raise RuntimeError(f"acme.sh timed out: {' '.join(cmd)}") from exc output = result.stdout if result.stderr: output = output + result.stderr if output else result.stderr if result.returncode != 0: raise RuntimeError(f"acme.sh failed (rc={result.returncode}): {output.strip()}") return output def _days_until(date_str: str) -> int | None: if not date_str: return None for fmt in ("%Y-%m-%d", "%Y-%m-%dT%H:%M:%S"): try: dt = datetime.strptime(date_str, fmt).replace(tzinfo=UTC) return (dt - datetime.now(UTC)).days except ValueError: continue try: dt = datetime.strptime(date_str, "%Y%m%d%H%M%z").astimezone(UTC) return (dt - datetime.now(UTC)).days except ValueError: pass return None def _parse_acme_list_output(raw: str) -> list[dict]: entries: list[dict] = [] for line in raw.strip().splitlines(): line = line.strip() if not line: continue entry: dict[str, str] = {} for token in line.split(): if ":" not in token: continue key, _, value = token.partition(":") entry[key.lower()] = value if entry: entries.append(entry) return entries def _has_auto_renew(domain: str) -> bool: acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme")) return bool(Path(acme_home_env) / f"{domain}.conf") def _get_acme_email() -> str: acme_home_default = str(PROJECT_DIR / "data" / "acme") try: acme_home = Path(os.environ.get("ACME_HOME", acme_home_default)) account_conf = acme_home / "account.conf" if account_conf.is_file(): text = account_conf.read_text() match = re.search(r"^ACME_LEEMAIL=(.+)$", text, re.MULTILINE) if match: return match.group(1).strip().strip("'\"") except OSError: pass return "" def _collect_acme() -> dict[str, Any]: """Collect ACME certificate list and email.""" email = _get_acme_email() certs: list[dict[str, Any]] = [] try: raw = _run_acme(["--list"]) entries = _parse_acme_list_output(raw) acme_home_env = os.environ.get("ACME_HOME", str(PROJECT_DIR / "data" / "acme")) acme_home = Path(acme_home_env) for entry in entries: main = entry.get("main_domain", "") if not main: continue san_domains = [ d.strip() for d in entry.get("san_domain", "").split(",") if d.strip() ] cert_dir = acme_home / main days = _days_until(entry.get("certificate_expires", "")) certs.append( { "domain": main, "issuer": entry.get("CA", ""), "expiry": entry.get("certificate_expires", ""), "days_remaining": days, "expired": days is not None and days <= 0, "cert_path": str(cert_dir / "fullchain.cer"), "key_path": str(cert_dir / f"{main}.key"), "ca_path": str(cert_dir / "ca.cer"), "issued_at": entry.get("certificate_date", ""), "expires_at": entry.get("certificate_expires", ""), "days_until_expiry": days, "auto_renew": _has_auto_renew(main), "san_domains": san_domains, } ) except Exception: pass return { "certs": certs, "email": email, "timestamp": _now_iso(), } register_collector("acme", _collect_acme) # --------------------------------------------------------------------------- # WireGuard collector # --------------------------------------------------------------------------- def _collect_wireguard() -> dict[str, Any]: """Collect WireGuard config, status, and peers.""" CONFIG_PATH = PROJECT_DIR / "config" / "wireguard" / "config.json" DEFAULT_CONFIG: dict[str, Any] = { "interface": { "name": "wg0", "listen_port": 51820, "private_key": "", "public_key": "", "addresses": ["10.137.0.1/24"], "post_up": None, "post_down": None, }, "peers": {}, } from lib.common import deep_merge cfg: dict[str, Any] = deepcopy(DEFAULT_CONFIG) if CONFIG_PATH.exists(): try: raw = load_json(CONFIG_PATH) if raw: cfg = deep_merge(deepcopy(DEFAULT_CONFIG), raw) except Exception: pass # Safe config (strip private key) safe = dict(cfg) if "interface" in safe: safe["interface"] = dict(safe["interface"]) safe["interface"].pop("private_key", None) # Peers list (safe) peers: list[dict[str, Any]] = [] for name, info in cfg.get("peers", {}).items(): entry = dict(info) entry["name"] = name entry.pop("private_key", None) peers.append(entry) # Runtime status status: dict[str, Any] = {"up": False, "interface": {}, "peers": []} name = cfg["interface"]["name"] peer_name = name if isinstance(name, str) else "wg0" try: res = run_proc(["wg", "show", peer_name], sudo=True, check=False) if res.returncode == 0: raw = res.stdout.strip() current_peer: dict[str, Any] | None = None status_peers: list[dict[str, Any]] = [] for line in raw.splitlines(): line = line.strip() if not line: continue if line.startswith("interface:"): status["up"] = True status["interface"] = {} current_peer = None continue if line.startswith("public key:"): status["interface"]["public_key"] = line.split(":", 1)[1].strip() continue if line.startswith("listening port:"): status["interface"]["listen_port"] = int( line.split(":", 1)[1].strip() ) continue if line.startswith("fwmark:"): status["interface"]["fwmark"] = line.split(":", 1)[1].strip() continue if line.startswith("peer:"): cur_key = line.split(":", 1)[1].strip() current_peer = { "public_key": cur_key, "endpoint": None, "allowed_ips": [], "latest_handshake": None, "transfer_received": "0", "transfer_sent": "0", "persistent_keepalive": None, } status_peers.append(current_peer) continue if current_peer is None: continue if line.startswith("endpoint:"): current_peer["endpoint"] = line.split(":", 1)[1].strip() elif line.startswith("allowed ips:"): current_peer["allowed_ips"] = ( line.split(":", 1)[1].strip().split(", ") ) elif line.startswith("latest handshake:"): current_peer["latest_handshake"] = line.split(":", 1)[1].strip() elif line.startswith("transfer:"): rest = line.split(":", 1)[1].strip().split(", ") if rest: current_peer["transfer_received"] = rest[0].strip() if len(rest) > 1: current_peer["transfer_sent"] = rest[1].strip() elif line.startswith("persistent-keepalive:"): with contextlib.suppress(ValueError): current_peer["persistent_keepalive"] = int( line.split(":", 1)[1].strip() ) status["peers"] = status_peers except Exception: pass return { "config": safe, "status": status, "peers": peers, "timestamp": _now_iso(), } register_collector("wireguard", _collect_wireguard) __all__ = [ "State", "state", ]