dashboards/tests/test_deploy_scripts.py

"""Contract tests for the deployment scripts in `deploy/`.

There is no shell test framework here — and adding one would buy nothing a subprocess does
not — but phase 10 rests on a sentence that lives in **two** files: the sudoers grant in
`deploy/agent-sudoers`, and the hint `deploy/install-agent.sh` prints when it is run without
root by the unprivileged `ops` account. Two copies of one instruction is exactly what drifts
silently, and the failure is an install that cannot happen on a host nobody is sitting at.
So these tests read the file, run the script the way the operator does (unprivileged, which
exits before the script can touch anything), and require the two lines to be the same line
apart from the path the script was invoked through.
"""

from __future__ import annotations

import os
import re
import shutil
import stat
import subprocess
from pathlib import Path

import pytest

ROOT = Path(__file__).resolve().parent.parent
DEPLOY = ROOT / "deploy"
INSTALLER = DEPLOY / "install-agent.sh"
SUDOERS = DEPLOY / "agent-sudoers"
BASH = shutil.which("bash") or "/bin/bash"

#: The grant as the file documents it. `/home/ops/agent-install` is the canonical staging
#: path: a monitored host has no repo checkout, so `deploy/` and `agent/` are copied there
#: (phase 10 task 02) and every host ends up with the same absolute path to grant.
CANONICAL_GRANT = "ops ALL=(root) NOPASSWD: /home/ops/agent-install/deploy/install-agent.sh"

#: The prefix a correct hint line starts with — user, run-as, and NOPASSWD, everything but
#: the path, which is the one thing the installer is allowed to substitute.
GRANT_PREFIX = "ops ALL=(root) NOPASSWD: "

#: Everything a non-root operator needs to hear, besides the grant line itself.
HINT_MUST_SAY = (
    "visudo -cf",  # check the syntax before installing it
    "install -m 0440 -o root -g root",  # and install it the way sudoers files are installed
    "/etc/sudoers.d/dashboards-agent",  # the destination both files name
    "deploy/agent-sudoers",  # the pointer back to the documented file
)


def _run(args: list[str], cwd: Path | None = None) -> subprocess.CompletedProcess[str]:
    return subprocess.run(args, capture_output=True, text=True, cwd=cwd, check=False)


def _grant_lines(text: str) -> list[str]:
    """The sudoers *rules* in a file — comments start with `#`, so this never matches prose."""
    return [line for line in text.splitlines() if line.startswith(GRANT_PREFIX)]


def _installer_without_root(*flags: str) -> subprocess.CompletedProcess[str]:
    """Run the installer the way `ops` does.

    Skipped as root on purpose: root would go past `need_root` and really install a systemd
    unit on the machine running pytest.
    """
    if hasattr(os, "geteuid") and os.geteuid() == 0:
        pytest.skip("running as root — the installer would install a real unit here")
    return _run([BASH, str(INSTALLER), *flags])


# --- deploy/agent-sudoers ---------------------------------------------------------------


def test_agent_sudoers_exists_and_is_a_config_file_not_a_script() -> None:
    assert SUDOERS.is_file(), "deploy/agent-sudoers is missing (phase 10 task 01)"
    assert not os.access(SUDOERS, os.X_OK), "a sudoers file is data; only install-agent.sh runs"


def test_agent_sudoers_is_exactly_one_grant_line_at_the_canonical_path() -> None:
    """One rule, the installer, the staging path — nothing else may be granted."""
    lines = _grant_lines(SUDOERS.read_text())
    assert lines == [CANONICAL_GRANT], (
        "deploy/agent-sudoers must grant exactly one command, the staged installer:\n" + "\n".join(lines)
    )
    assert "/home/ops/agent-install" in SUDOERS.read_text()


def test_agent_sudoers_grants_no_shell_editor_or_second_path() -> None:
    """The 'deliberately NOT granted' block has to stay a comment, not become a rule."""
    text = SUDOERS.read_text()
    not_granted = text.split("deliberately NOT granted", 1)[-1]
    for never in ("visudo", "sh", "tee", "systemctl", "usermod", "upscmd", "upsrw"):
        assert never in not_granted, f"the file must name {never!r} as deliberately not granted"
    assert not_granted.lstrip().startswith("-"), "the block heading is followed by the list"
    # Every line of that block is a comment: a stray rule there would be a second grant.
    assert _grant_lines(not_granted) == []
    # And it names the token file, because "we did not grant reading it" is the point.
    assert "/etc/dashboards-agent.token" in not_granted


def test_agent_sudoers_documents_the_why_and_the_two_stage_install() -> None:
    """House style (deploy/nut-sudoers): why it is needed, then check-first install steps."""
    text = SUDOERS.read_text()
    assert "WHY THIS IS NEEDED" in text
    for step in HINT_MUST_SAY:
        assert step in text, f"deploy/agent-sudoers must document {step!r}"
    # The installed copy is checked too — the file being valid is not the same as the
    # installed file being valid (nut-sudoers' third step, and the reason stays).
    assert text.count("visudo -cf") >= 2


@pytest.mark.skipif(shutil.which("visudo") is None, reason="visudo is not installed on this box")
def test_agent_sudoers_passes_the_syntax_check_it_documents() -> None:
    """The file tells the operator to run `visudo -cf` on it, so it must survive that."""
    result = _run([shutil.which("visudo") or "/usr/sbin/visudo", "-cf", str(SUDOERS)])
    assert result.returncode == 0, f"visudo -cf failed:\n{result.stdout}{result.stderr}"


# --- deploy/install-agent.sh without root -----------------------------------------------


def test_installer_without_root_exits_1_and_prints_the_grant_line() -> None:
    result = _installer_without_root("--bind", "127.0.0.1")
    assert result.returncode == 1, "the non-root refusal keeps its exit code (1)"
    assert result.stdout == "", "the refusal is a failure: it goes to stderr, not stdout"
    lines = _grant_lines(result.stderr)
    assert lines == [f"{GRANT_PREFIX}{Path(os.path.realpath(INSTALLER))}"], (
        "the hint must print one grant line naming this script's resolved path:\n" + result.stderr
    )


def test_hint_and_the_sudoers_file_are_the_same_line_apart_from_the_path() -> None:
    """The pinned version of the diff in data/logs/NOTES-sudoers.md."""
    printed = _installer_without_root("--bind", "127.0.0.1").stderr
    hint = _grant_lines(printed)[0]
    canonical = _grant_lines(SUDOERS.read_text())[0]
    assert hint.startswith(GRANT_PREFIX) and canonical.startswith(GRANT_PREFIX)
    assert hint.removeprefix(GRANT_PREFIX).endswith("/deploy/install-agent.sh")
    assert hint.removeprefix(GRANT_PREFIX).removeprefix(str(DEPLOY.parent)) == (
        canonical.removeprefix(GRANT_PREFIX).removeprefix("/home/ops/agent-install")
    ), "same user, same (root), same NOPASSWD, same script — only the checkout path differs"


def test_hint_resolves_a_relative_invocation_to_an_absolute_path() -> None:
    """`bash deploy/install-agent.sh` from the repo root still prints a path sudo can match."""
    result = _run([BASH, "deploy/install-agent.sh", "--bind", "10.0.0.5"], cwd=ROOT)
    assert result.returncode == 1
    assert _grant_lines(result.stderr) == [f"{GRANT_PREFIX}{Path(os.path.realpath(INSTALLER))}"]


def test_installer_names_the_two_stage_install_and_points_at_the_file() -> None:
    stderr = _installer_without_root("--bind", "127.0.0.1").stderr
    for phrase in HINT_MUST_SAY:
        assert phrase in stderr, f"the non-root hint must say {phrase!r}, not just 'run as root'"
    # The pointer is the file next to *this* script, wherever the checkout lives.
    assert str(Path(os.path.realpath(SUDOERS))) in stderr


def test_installer_hint_repeats_the_command_that_was_asked_for() -> None:
    """Copy-pasteable: the hint ends with the same invocation, with the flags it was given."""
    stderr = _installer_without_root("--uninstall").stderr
    assert f"sudo {Path(os.path.realpath(INSTALLER))} --uninstall" in stderr


def test_installer_help_points_at_the_sudoers_grant() -> None:
    """The header comment *is* the help text, so the grant is discoverable from --help."""
    result = _run([BASH, str(INSTALLER), "--help"])
    assert result.returncode == 0
    assert "deploy/agent-sudoers" in result.stdout


# --- the tunnel machinery stays retired (phase 10) ---------------------------------------

#: Names and ports that only ever existed to serve the ssh-tunnel dev transport, which phase
#: 10 deleted. Every pattern is written as a character class (`[d]ev-…`) so that **this file
#: cannot match its own grep**: the phase criterion demands the repo stay free of those
#: literals, and a guard that reintroduces one to forbid it is the oldest trick there is.
#: Scope excludes `.agents/` (phase files and run reports are history — rewriting a finished
#: phase's record to satisfy a grep would falsify it) and `tests/fixtures/` (verbatim
#: upstream captures: a sector counter or a wheel hash may contain those digits by accident,
#: and rule 7 forbids editing a capture to please a grep).
TUNNEL_PATTERNS = ("[d]ev-tunnels", "DEV_[T]UNNELS", "DEV_[F]ORWARDS", r"912[123]\b", r"[3]494\b")

#: Where those names must not appear: every file that *tells someone what to run*.
#: `tests/fixtures/` is excluded (verbatim captures, see above) and so is `.agents/` —
#: phase files and run reports are history, and rewriting a finished phase's record to
#: satisfy a grep would falsify it.
TUNNEL_FREE_ROOTS = ("scripts", "deploy", "src", "agent", "web")


def _tunnel_scope_files() -> list[Path]:
    files: list[Path] = []
    for top in TUNNEL_FREE_ROOTS:
        for path in (ROOT / top).rglob("*"):
            if path.is_file() and path.suffix in (".py", ".sh", ".md", ".css", ".js", ".html"):
                files.append(path)
    files.append(ROOT / "README.md")
    files.append(ROOT / "AGENTS.md")
    files.append(ROOT / ".env.example")
    files.append(ROOT / ".env.dev.example")
    for path in (ROOT / "tests").glob("*.py"):
        files.append(path)
    return files


def test_dev_tunnel_script_is_deleted() -> None:
    """The tunnel script is gone, not shelved: git history keeps it if it is ever needed."""
    # Name built at runtime so this file never carries the literal the grep forbids.
    script = ROOT / "scripts" / ("dev-" + "tunnels.sh")
    assert not script.exists(), (
        "the retired ssh-tunnel dev script is back — phase 10 deleted it deliberately"
    )


def test_no_operational_file_mentions_the_tunnel_machinery() -> None:
    """A stale tunnel-env instruction is a setup step nobody can complete any more."""
    offenders: list[str] = []
    for path in _tunnel_scope_files():
        text = path.read_text(errors="replace")
        for pattern in TUNNEL_PATTERNS:
            if re.search(pattern, text):
                offenders.append(f"{path.relative_to(ROOT)}: /{pattern}/")
    assert not offenders, "tunnel-era names/ports are back:\n" + "\n".join(offenders)


# --- shell syntax (the only gate these scripts get) -------------------------------------


#: Names a shell function must never take. Bash resolves a **function** before an external
#: command of the same name, so defining `install()` in a script that then calls
#: `install -d /somewhere` is not a directory creation — it is the function calling itself,
#: forever. That is not hypothetical: `deploy/install-agent.sh` did exactly this, and the
#: first real install of phase 10 spun on the SMART block for two minutes and installed
#: nothing (no unit, no agent file, no token) while looking like a working script to `bash -n`.
SHADOW_FORBIDDEN = frozenset(
    {
        "install",
        "test",
        "time",
        "date",
        "sort",
        "grep",
        "sed",
        "tee",
        "type",
        "ulimit",
        "wait",
        "kill",
        "cd",
        "pwd",
    }
)

_FUNCTION_DEF_RE = r"^([A-Za-z_][A-Za-z0-9_]*)\s*\(\)"


def _function_names(script: Path) -> list[str]:
    return re.findall(_FUNCTION_DEF_RE, script.read_text(), flags=re.MULTILINE)


@pytest.mark.parametrize(
    "script",
    sorted([*DEPLOY.glob("*.sh"), *(ROOT / "scripts").glob("*.sh")]),
    ids=lambda p: str(p.relative_to(ROOT)),
)
def test_shell_script_has_valid_syntax(script: Path) -> None:
    result = _run([BASH, "-n", str(script)])
    assert result.returncode == 0, f"bash -n failed for {script}:\n{result.stderr}"


@pytest.mark.parametrize(
    "script",
    sorted([*DEPLOY.glob("*.sh"), *(ROOT / "scripts").glob("*.sh")]),
    ids=lambda p: str(p.relative_to(ROOT)),
)
def test_no_shell_function_is_named_after_a_coreutil_it_could_be_asked_to_run(script: Path) -> None:
    """`install()` + `install -d` is infinite recursion, and `bash -n` cannot see it."""
    shadowed = SHADOW_FORBIDDEN.intersection(_function_names(script))
    assert not shadowed, (
        f"{script.relative_to(ROOT)} defines function(s) {sorted(shadowed)} that shadow an "
        "external command: bash runs the function, so any call meant for the program recurses"
    )


# --- the install path, run for real against a sandboxed root boundary --------------------

#: The commands of `install()` that would touch the machine running pytest. Each one is a
#: PATH stub that records its arguments and succeeds — the same trick `ScriptedNutClient` is
#: for the UPS wire: fake the privileged boundary, run the real logic.
ROOT_BOUNDARY_STUBS = ("id", "getent", "useradd", "usermod", "chown", "systemctl")

#: The three absolute destinations the installer writes to, rewritten into a temp directory
#: so the test exercises the body instead of the operator's `/usr` and `/etc`.
SANDBOX_REWRITES = {
    'AGENT_DIR="/usr/local/lib/dashboards-agent"': 'AGENT_DIR="{root}/lib"',
    'TOKEN_FILE="/etc/dashboards-agent.token"': 'TOKEN_FILE="{root}/token"',
    'UNIT="/etc/systemd/system/$SERVICE.service"': 'UNIT="{root}/dashboards-agent.service"',
}


def _sandbox_installer(tmp_path: Path) -> tuple[Path, Path, dict[str, str]]:
    """A copy of the installer whose writes land in `tmp_path`, plus a stubbed PATH.

    Returns (script, stub call log, env). `agent/host_agent.py` is stubbed next to the copy
    because the script resolves the agent as `../agent/host_agent.py`.
    """
    text = INSTALLER.read_text()
    target = tmp_path / "target"
    for old, new in SANDBOX_REWRITES.items():
        assert old in text, f"the installer moved {old!r} — update SANDBOX_REWRITES with it"
        text = text.replace(old, new.format(root=target))

    script = tmp_path / "deploy" / "install-agent.sh"
    script.parent.mkdir(parents=True)
    script.write_text(text)
    (tmp_path / "agent").mkdir()
    (tmp_path / "agent" / "host_agent.py").write_text("# stub agent source\n", encoding="utf-8")

    stubs = tmp_path / "stubs"
    stubs.mkdir()
    call_log = tmp_path / "stub-calls.log"
    for name in ROOT_BOUNDARY_STUBS:
        stub = stubs / name
        # `id -u` must answer 0 or need_root() refuses before the body runs; every other call
        # just records itself, which is what the assertions below read.
        body = ['#!/bin/sh', f'printf "%s %s\\n" "{name}" "$*" >> "$STUB_LOG"']
        if name == "id":
            body.append('[ "$1" = "-u" ] && echo 0')
        body.append("exit 0")
        stub.write_text("\n".join(body) + "\n", encoding="utf-8")
        stub.chmod(0o755)

    env = {**os.environ, "PATH": f"{stubs}{os.pathsep}{os.environ['PATH']}", "STUB_LOG": str(call_log)}
    return script, call_log, env


@pytest.mark.skipif(hasattr(os, "geteuid") and os.geteuid() == 0, reason="running as root")
def test_install_path_installs_into_its_own_paths_and_prints_the_token_once(tmp_path: Path) -> None:
    """The install body end to end: files, unit, token, and no recursion.

    The exit code is the regression test for the `install()` shadow: while the function was
    named after `/usr/bin/install`, this never returned at all.
    """
    script, call_log, env = _sandbox_installer(tmp_path)
    try:
        result = subprocess.run(
            [BASH, str(script), "--bind", "10.0.0.5", "--port", "9120"],
            capture_output=True,
            text=True,
            env=env,
            timeout=60,
            check=False,
        )
    except subprocess.TimeoutExpired as exc:
        pytest.fail(f"the installer did not return (recursion? partial output: {exc.stdout!r})")
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"

    target = tmp_path / "target"
    assert (target / "lib" / "host_agent.py").is_file(), "install -m never landed the agent"
    unit = (target / "dashboards-agent.service").read_text()
    token_path = target / "token"
    token = token_path.read_text().strip()

    assert "User=dashboards-agent" in unit
    assert f"--token-file {token_path}" in unit, "the unit must use the file, never --token"
    assert "--host 10.0.0.5 --port 9120" in unit
    # DevicePolicy=strict denies every node not listed, and systemd opens /dev/null for stdin
    # before it execs python: without this line the unit dies at spawn (status 208/STDIN).
    assert "DevicePolicy=strict" in unit
    assert "DeviceAllow=/dev/null" in unit, "strict device policy needs /dev/null or the unit cannot start"
    assert "DeviceAllow=/dev/nvme*" in unit, "SMART reads need the disk nodes back"
    assert token and len(token) >= 32, "a generated token is expected"
    assert token in result.stdout, "first install prints the token, and never again"
    assert f'HOST_TARGETS="hostname=http://{token}@10.0.0.5:9120"' in result.stdout

    calls = call_log.read_text()
    assert "systemctl daemon-reload" in calls
    assert "systemctl enable --now dashboards-agent" in calls
    assert "chown root:dashboards-agent" in calls, "the token file must become group-readable"


def test_token_file_is_readable_by_the_service_user_and_nobody_else(tmp_path: Path) -> None:
    """`--token-file` is a path the *agent* opens, as its own user.

    systemd does not read the secret for the service, so a 0600 root-owned file makes the
    agent exit before it binds. 0640 root:dashboards-agent is the smallest mode that works,
    and it is still unreadable by `ops` (which is why deploy/agent-sudoers does not grant it).
    """
    script, _call_log, env = _sandbox_installer(tmp_path)
    result = subprocess.run(
        [BASH, str(script), "--bind", "10.0.0.5"],
        capture_output=True,
        text=True,
        env=env,
        timeout=60,
        check=False,
    )
    assert result.returncode == 0, f"install failed:\n{result.stdout}\n{result.stderr}"
    mode = stat.S_IMODE((tmp_path / "target" / "token").stat().st_mode)
    assert mode == 0o640, f"token file mode is {oct(mode)}, expected 0o640 (group = the agent user)"
    assert not mode & 0o007, "the token file must never be world-readable"