nisse/tests/deploy/test_agent_host.py

"""The agent host's vault and SilverBullet (deploy/agent-host; L10, M5.4):
the operator writes the vault, never the agents' memory; nothing but the
vault and SilverBullet's own data is mounted; the image changes only in a
PR; only the operator's Caddy reaches it."""

import re
import shutil
import subprocess
from pathlib import Path
from typing import Any

import pytest
import yaml

HOST = Path(__file__).resolve().parents[2] / "deploy" / "agent-host"
VAULT = "/srv/nisse/vault"


def silverbullet() -> dict[str, Any]:
    compose = yaml.safe_load((HOST / "silverbullet" / "compose.yaml").read_text())
    return compose["services"]["silverbullet"]


def mounts() -> list[tuple[str, str, str]]:
    out = []
    for volume in silverbullet()["volumes"]:
        source, target, *mode = volume.split(":")
        out.append((source, target, mode[0] if mode else "rw"))
    return out


def setting(name: str) -> str:
    match = re.search(rf"^{name}=(\S+)", (HOST / "setup.sh").read_text(), re.MULTILINE)
    assert match, name
    return match.group(1)


def test_the_agents_memory_is_read_only_and_the_rest_of_the_vault_is_not() -> None:
    by_source = {source: (target, mode) for source, target, mode in mounts()}
    vault_target, vault_mode = by_source[VAULT]
    assert vault_mode == "rw"
    # memory/ is mounted over the same place in the space, read-only.
    assert by_source[f"{VAULT}/memory"] == (f"{vault_target}/memory", "ro")
    # Nothing else of the vault is mounted, so nothing gets around that.
    others = [s for s in by_source if s.startswith(VAULT)]
    assert sorted(others) == [VAULT, f"{VAULT}/memory"]


def test_it_sees_nothing_of_the_host_but_the_vault_and_its_own_data() -> None:
    assert sorted(source for source, _, _ in mounts()) == [
        "/srv/nisse/silverbullet",
        VAULT,
        f"{VAULT}/memory",
    ]


def test_the_image_is_pinned_to_a_release() -> None:
    # The docker role updates images nightly; a pinned release changes only here.
    image = silverbullet()["image"]
    assert re.fullmatch(
        r"ghcr\.io/silverbulletmd/silverbullet:\d+\.\d+\.\d+(-slim)?", image
    )


def test_no_secrets_in_the_compose_file() -> None:
    environment = silverbullet().get("environment", {})
    assert not [k for k in environment if re.search(r"USER|TOKEN|PASS|SECRET|KEY", k)]


def test_its_port_caddys_and_the_firewalls_agree() -> None:
    port = setting("SB_PORT")
    assert silverbullet()["ports"] == [f"0.0.0.0:{port}:3000"]  # IPv4 only
    caddy = (HOST / "caddy" / "vault.dongar.net.caddy").read_text()
    assert f"reverse_proxy nisse.garrett-hall.com:{port}" in caddy
    assert "@outside not remote_ip private_ranges" in caddy  # the LAN only


def test_logs_go_to_the_hosts_journal() -> None:
    assert silverbullet()["logging"] == {"driver": "journald"}


def _can_use_a_network_namespace() -> bool:
    if not (shutil.which("unshare") and shutil.which("iptables")):
        return False
    probe = subprocess.run(
        ["unshare", "-n", "iptables", "-S"], capture_output=True, check=False
    )
    return probe.returncode == 0


@pytest.mark.skipif(
    not _can_use_a_network_namespace(),
    reason="needs root, unshare and iptables to try the rule in a namespace",
)
def test_the_firewall_lets_only_one_address_in_and_replaces_its_own_rule() -> None:
    script = HOST / "nisse-firewall"
    steps = (
        "iptables -N DOCKER-USER; iptables -A DOCKER-USER -j RETURN; "
        f"{script} 192.168.6.13 3000; {script} 192.168.6.13 3000; "
        f"{script} 192.168.6.14 3000; iptables -S DOCKER-USER"
    )
    run = subprocess.run(
        ["unshare", "-n", "sh", "-ec", steps],
        capture_output=True,
        text=True,
        check=True,
    )
    rules = [r for r in run.stdout.splitlines() if "nisse-firewall" in r]
    assert len(rules) == 1, run.stdout  # run again: still one; changed: replaced
    [rule] = rules
    assert "! -s 192.168.6.14/32" in rule
    assert "--ctorigdstport 3000" in rule
    assert rule.endswith("-j DROP")
    # Ahead of Docker's own return, so it's checked first.
    assert run.stdout.splitlines().index(rule) < run.stdout.splitlines().index(
        "-A DOCKER-USER -j RETURN"
    )


SCRIPT = HOST.parents[1] / "scripts" / "agent-host.sh"


def _fake_host(tmp_path: Path, *, memory_writable: bool) -> dict[str, str]:
    """Commands standing in for the host's: ssh runs the command here, sudo
    just runs it, and docker reads all of stdin, as docker compose exec does."""
    bin_ = tmp_path / "bin"
    bin_.mkdir()
    stubs = {
        "ssh": 'for last; do :; done; exec bash -c "$last"',
        "sudo": 'exec "$@"',
        "curl": "echo 200",
        "iptables": "echo '-A DOCKER-USER -m comment --comment nisse-firewall -j DROP'",
        "docker": (
            "cat > /dev/null\n"
            'case "$*" in\n'
            '  *"ps --status running"*) echo abc123 ;;\n'
            f'  *"/vault/memory/"*) exit {0 if memory_writable else 1} ;;\n'
            "esac"
        ),
    }
    for name, body in stubs.items():
        path = bin_ / name
        path.write_text(f"#!/usr/bin/env bash\n{body}\n")
        path.chmod(0o755)
    stack = tmp_path / "stack"
    stack.mkdir()
    return {"PATH": f"{bin_}:/usr/bin:/bin", "STACK": str(stack), "HOME": str(tmp_path)}


def _check(env: dict[str, str]) -> subprocess.CompletedProcess[str]:
    return subprocess.run(
        ["bash", str(SCRIPT), "check"],
        env=env,
        capture_output=True,
        text=True,
        timeout=30,
        check=False,
    )


def test_the_check_runs_every_step_even_though_docker_reads_stdin(
    tmp_path: Path,
) -> None:
    # Fed to the host on stdin, the first docker compose exec swallowed the
    # rest of the check, which then passed without the last two steps.
    result = _check(_fake_host(tmp_path, memory_writable=False))
    assert result.returncode == 0, result.stdout + result.stderr
    assert result.stdout.splitlines() == [
        "ok:   SilverBullet is running",
        "ok:   it answers locally (200)",
        "ok:   the operator's folders are writable",
        "ok:   memory/ is read-only to SilverBullet",
        "ok:   the firewall rule is in place",
        "all checks passed",
    ]


def test_a_writable_memory_folder_fails_the_check(tmp_path: Path) -> None:
    result = _check(_fake_host(tmp_path, memory_writable=True))
    assert result.returncode == 1
    assert (
        "FAIL: memory/ is writable from SilverBullet, even as root in it"
        in result.stdout
    )
    assert "ok:   the firewall rule is in place" in result.stdout  # still checked
    assert "all checks passed" not in result.stdout