vacuum-wall/lib/common.py

"""Shared utilities for Vacuum Wall lib/ modules.

Provides common helpers for JSON persistence, subprocess execution,
deep merging, and directory creation used across all subsystem modules.
"""

import hashlib
import json
import os
import re
import subprocess
from copy import deepcopy
from pathlib import Path
from typing import Any

from passlib.hash import sha256_crypt

_APPLY_HASH_KEY = "_last_applied_hash"
_LAST_APPLIED_CONFIG_KEY = "_last_applied_config"
_APPLY_META_KEYS = (_APPLY_HASH_KEY, _LAST_APPLIED_CONFIG_KEY)


def strip_apply_meta(cfg: dict[str, Any]) -> dict[str, Any]:
    """Return *cfg* without any apply bookkeeping keys.

    Strips both the last-applied hash and the last-applied config snapshot
    so the returned dict reflects only real configuration.
    """
    return {k: v for k, v in cfg.items() if k not in _APPLY_META_KEYS}


def config_hash(cfg: dict[str, Any]) -> str:
    """Compute a SHA-256 hash of *cfg*, ignoring apply bookkeeping keys.

    Args:
        cfg: Config dict, possibly containing ``_last_applied_hash`` and
            ``_last_applied_config``.

    Returns:
        Hex digest of the stripped config JSON.
    """
    clean = strip_apply_meta(cfg)
    return hashlib.sha256(json.dumps(clean, sort_keys=True).encode()).hexdigest()


def stamp_applied(cfg: dict[str, Any]) -> dict[str, Any]:
    """Record that *cfg* is the last applied configuration.

    Writes both the content snapshot (``_last_applied_config``) and its hash
    (``_last_applied_hash``) so a later pending check can detect drift and a
    diff can report exactly which fields changed.
    """
    cfg[_APPLY_HASH_KEY] = config_hash(cfg)
    cfg[_LAST_APPLIED_CONFIG_KEY] = strip_apply_meta(cfg)
    return cfg


def deep_diff(old: Any, new: Any, prefix: str = "") -> list[dict[str, Any]]:
    """Return a list of field-level changes between two configurations.

    Each entry is ``{"path", "action", "old", "new"}`` where *action* is one
    of ``"added"``, ``"removed"`` or ``"changed"``. Dicts are recursed with
    dotted paths; lists of equal length are compared element-by-element,
    while any other value that differs is reported as a single change.
    Apply bookkeeping keys are ignored.
    """
    if isinstance(old, dict):
        old = strip_apply_meta(old)
    if isinstance(new, dict):
        new = strip_apply_meta(new)
    out: list[dict[str, Any]] = []
    _diff_nodes(old, new, prefix, out)
    return out


def _diff_nodes(old: Any, new: Any, path: str, out: list[dict[str, Any]]) -> None:
    """Recursively collect field-level changes from *old* into *new*."""
    if isinstance(old, dict) and isinstance(new, dict):
        for key in sorted(set(old) | set(new)):
            child = f"{path}.{key}" if path else str(key)
            if key in old and key in new:
                _diff_nodes(old[key], new[key], child, out)
            elif key in old:
                out.append(
                    {"path": child, "action": "removed", "old": old[key], "new": None}
                )
            else:
                out.append(
                    {"path": child, "action": "added", "old": None, "new": new[key]}
                )
        return
    if isinstance(old, list) and isinstance(new, list) and len(old) == len(new):
        for i, (o, n) in enumerate(zip(old, new, strict=True)):
            _diff_nodes(o, n, f"{path}[{i}]", out)
        return
    if old != new:
        out.append({"path": path, "action": "changed", "old": old, "new": new})


def validate_interface_name(name: str) -> str:
    """Validate a Linux network interface name.

    Args:
        name: Interface name to validate.

    Returns:
        The validated (stripped) name.

    Raises:
        ValueError: When the name is empty, contains path components,
            or does not match Linux interface naming rules.
    """
    if not name or not isinstance(name, str):
        raise ValueError("Interface name must be a non-empty string")
    name = name.strip()
    if not name:
        raise ValueError("Interface name must not be blank")
    if "/" in name or ".." in name or " " in name:
        raise ValueError(f"Invalid interface name: {name!r}")
    if not re.match(r"^[a-zA-Z0-9][a-zA-Z0-9._-]*$", name):
        raise ValueError(f"Invalid interface name: {name!r}")
    return name


def run(
    cmd: list[str],
    check: bool = True,
    sudo: bool = False,
    timeout: int | None = None,
) -> str:
    """Run a command and return stripped stdout.

    Args:
        cmd: Command arguments.
        check: Raise RuntimeError on non-zero exit.
        sudo: Prefix command with ``sudo``.
        timeout: Timeout in seconds (``None`` → no timeout).

    Returns:
        ``stdout`` with trailing whitespace removed.

    Raises:
        RuntimeError: When ``check=True`` and the process exits non-zero.
    """
    full_cmd = ["sudo", *cmd] if sudo else list(cmd)
    try:
        result = subprocess.run(
            full_cmd,
            capture_output=True,
            text=True,
            shell=False,
            check=check,
            timeout=timeout,
        )
        return result.stdout.strip()
    except subprocess.CalledProcessError as exc:
        raise RuntimeError(
            f"Command failed: {' '.join(full_cmd)} (rc={exc.returncode}): "
            f"{exc.stderr.strip()}"
        ) from exc


def run_proc(
    cmd: list[str],
    check: bool = True,
    sudo: bool = False,
    timeout: int | None = None,
    input: str | None = None,
) -> subprocess.CompletedProcess[str]:
    """Run a command and return the full ``CompletedProcess``.

    Args:
        cmd: Command arguments.
        check: Raise ``subprocess.CalledProcessError`` on non-zero exit.
        sudo: Prefix command with ``sudo``.
        timeout: Timeout in seconds.
        input: String to pass as stdin to the subprocess.

    Returns:
        The completed process object.
    """
    full_cmd = ["sudo", *cmd] if sudo else list(cmd)
    return subprocess.run(
        full_cmd,
        capture_output=True,
        text=True,
        shell=False,
        check=check,
        timeout=timeout,
        input=input,
    )


def load_json(path: Path, default: dict[str, Any] | None = None) -> dict[str, Any]:
    """Load JSON from *path*.

    Returns *default* (default ``{}``) if the file does not exist.
    """
    if default is None:
        default = {}
    if not path.exists():
        return deepcopy(default)
    with open(path) as f:
        return json.load(f)


def save_json(path: Path, data: dict[str, Any], indent: int = 4) -> None:
    """Atomically write *data* as JSON to *path*.

    Writes to ``path.tmp`` first, then replaces *path* via ``os.replace()``
    to avoid partial writes.
    """
    path.parent.mkdir(parents=True, exist_ok=True)
    tmp = path.with_suffix(path.suffix + ".tmp")
    with open(tmp, "w") as f:
        json.dump(data, f, indent=indent)
        f.write("\n")
    os.replace(tmp, path)


def deep_merge(base: dict[str, Any], overrides: dict[str, Any]) -> dict[str, Any]:
    """Recursively merge *overrides* into a deep copy of *base*.

    For nested dicts the merge recurses; for all other values
    *overrides* wins.
    """
    result = deepcopy(base)
    for k, v in overrides.items():
        if k in result and isinstance(result[k], dict) and isinstance(v, dict):
            result[k] = deep_merge(result[k], v)
        else:
            result[k] = deepcopy(v)
    return result


def ensure_dirs(*dirs: Path) -> None:
    """Create each directory (and parents) if it does not exist."""
    for d in dirs:
        d.mkdir(parents=True, exist_ok=True)


def _hash_password(password: str) -> str:
    """Hash *password* using SHA-256 crypt (``$5$`` format) via passlib.

    Used for nginx htpasswd files. NOT used for auth user passwords —
    those use Argon2id via ``lib.password``.

    Args:
        password: Plain-text password to hash.

    Returns:
        The hashed password string suitable for ``.htpasswd``
        (e.g. ``$5$rounds=…$…``).
    """
    return sha256_crypt.hash(password)


def get_interface_ip(iface: str) -> str | None:
    """Return the primary IPv4 address of *iface* (without CIDR), or ``None``.

    Uses ``ip -o addr show`` which is in the daemon sudo whitelist.
    """
    if not iface:
        return None
    try:
        raw = run(["ip", "-o", "addr", "show", iface], sudo=True)
        for line in raw.splitlines():
            parts = line.split()
            # -o format: "NUM: IFACE   inet/6 ADDR/MASK ..."
            if len(parts) >= 4 and parts[2] == "inet":
                return parts[3].split("/", 1)[0]
    except Exception:
        pass
    return None


__all__ = [
    "_APPLY_HASH_KEY",
    "_LAST_APPLIED_CONFIG_KEY",
    "_hash_password",
    "config_hash",
    "deep_diff",
    "deep_merge",
    "ensure_dirs",
    "get_interface_ip",
    "load_json",
    "run",
    "run_proc",
    "save_json",
    "stamp_applied",
    "strip_apply_meta",
    "validate_interface_name",
]