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"


def config_hash(cfg: dict[str, Any]) -> str:
    """Compute a SHA-256 hash of *cfg* excluding the ``_last_applied_hash`` key.

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

    Returns:
        Hex digest of the stripped config JSON.
    """
    clean = {k: v for k, v in cfg.items() if k != _APPLY_HASH_KEY}
    return hashlib.sha256(json.dumps(clean, sort_keys=True).encode()).hexdigest()


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",
    "_hash_password",
    "config_hash",
    "deep_merge",
    "ensure_dirs",
    "get_interface_ip",
    "load_json",
    "run",
    "run_proc",
    "save_json",
    "validate_interface_name",
]