left4me/deploy/tests/test_example_units.py

"""Lockdown tests for the curated examples kept under `deploy/files/`.

`deploy/` is reference material. The production units are emitted by
ckn-bw's `systemd_units` reactor in `bundles/left4me/metadata.py`;
when reactor output drifts intentionally, update these examples to match.
"""
from pathlib import Path


ROOT = Path(__file__).resolve().parents[2]
DEPLOY = ROOT / "deploy"


WEB_UNIT = DEPLOY / "files/usr/local/lib/systemd/system/left4me-web.service"
SERVER_UNIT = DEPLOY / "files/usr/local/lib/systemd/system/left4me-server@.service"
GAME_SLICE = DEPLOY / "files/usr/local/lib/systemd/system/l4d2-game.slice"
BUILD_SLICE = DEPLOY / "files/usr/local/lib/systemd/system/l4d2-build.slice"
SYSCTL_CONF = DEPLOY / "files/etc/sysctl.d/99-left4me.conf"
SANDBOX_RESOLV_CONF = DEPLOY / "files/etc/left4me/sandbox-resolv.conf"
HOST_ENV = DEPLOY / "templates/etc/left4me/host.env"
WEB_ENV_TEMPLATE = DEPLOY / "templates/etc/left4me/web.env.template"


def test_global_unit_files_exist_at_product_level_paths():
    assert WEB_UNIT.is_file()
    assert SERVER_UNIT.is_file()


def test_web_unit_contains_required_runtime_contract():
    unit = WEB_UNIT.read_text()

    assert "User=left4me" in unit
    assert "Group=left4me" in unit
    assert "WorkingDirectory=/opt/left4me" in unit
    assert "Environment=PATH=/opt/left4me/.venv/bin:" in unit
    assert "EnvironmentFile=/etc/left4me/host.env" in unit
    assert "EnvironmentFile=/etc/left4me/web.env" in unit
    assert "ExecStart=/opt/left4me/.venv/bin/gunicorn" in unit
    assert "--workers 1" in unit
    assert "--threads 32" in unit
    # NoNewPrivileges must remain unset because sudo (used by the overlay,
    # systemctl and journalctl helpers) is setuid.
    assert "NoNewPrivileges=true" not in unit
    # Restored now that fuse-overlayfs propagation is no longer the mechanism.
    assert "PrivateTmp=true" in unit
    assert "ProtectSystem=full" in unit
    assert "ReadWritePaths=/var/lib/left4me" in unit
    # Mounts now happen in PID 1's namespace via the left4me-overlay helper,
    # so MountFlags propagation is irrelevant — and the previous assumption
    # that MountFlags=shared made it work was incorrect.
    assert "MountFlags=" not in unit


def test_server_unit_contains_required_runtime_contract():
    unit = SERVER_UNIT.read_text()

    assert "User=left4me" in unit
    assert "Group=left4me" in unit
    assert "EnvironmentFile=/etc/left4me/host.env" in unit
    assert "EnvironmentFile=/var/lib/left4me/instances/%i/instance.env" in unit
    # `-` prefix: chdir failure is non-fatal so ExecStartPre can run the
    # mount helper before the merged dir exists. ExecStart re-applies and
    # finds the dir once the mount has landed.
    assert "WorkingDirectory=-/var/lib/left4me/runtime/%i/merged/left4dead2" in unit
    # ExecStart must invoke srcds_run from the *merged* overlay tree, not
    # from installation/. srcds_run cds to its own dirname; if we point at
    # installation/, the engine reads gameinfo.txt and addons from the lower
    # layer and never sees overlay plugins (Metamod/SourceMod) or cfgs.
    assert "ExecStart=/var/lib/left4me/runtime/%i/merged/srcds_run" in unit
    assert "$L4D2_ARGS" in unit
    assert "${L4D2_ARGS}" not in unit
    assert "NoNewPrivileges=true" in unit
    assert "PrivateTmp=true" in unit
    assert "PrivateDevices=true" in unit
    assert "ProtectHome=true" in unit
    assert "ProtectSystem=strict" in unit
    assert "ReadOnlyPaths=/var/lib/left4me/installation /var/lib/left4me/overlays" in unit
    assert "ReadWritePaths=/var/lib/left4me/runtime/%i" in unit
    assert "RestrictSUIDSGID=true" in unit
    assert "LockPersonality=true" in unit


def test_server_unit_mounts_overlay_via_exec_start_pre():
    """At boot, systemd auto-starts enabled units before the web app gets a
    chance to run start_instance's pre-start mount. The unit itself must
    re-mount the overlay so reboots are transparent. Pairs with the helper's
    idempotency check (test_overlay_helper_mount_is_idempotent_when_mounted).

    The unit-level `nsenter --mount=/proc/1/ns/mnt --` is what makes
    umount fast: without it, the helper Python process would inherit
    the unit's per-service mount namespace and pin it alive, blocking
    PID 1's umount until the helper exited. Wrapping with nsenter at
    the Exec line puts the helper itself in PID 1's namespace.
    """
    unit = SERVER_UNIT.read_text()
    # `+` prefix: runs as PID 1 (root, no sandbox). Required because
    # the unit has NoNewPrivileges=true, which blocks sudo's setuid
    # escalation — and the helper needs root for the mount syscall.
    assert (
        "ExecStartPre=+/usr/bin/nsenter --mount=/proc/1/ns/mnt -- "
        "/usr/local/libexec/left4me/left4me-overlay mount %i"
        in unit
    )
    # Bound the restart loop; without these, a CHDIR-failure (or any other
    # pre-start error) spins indefinitely.
    assert "StartLimitBurst=5" in unit
    assert "StartLimitIntervalSec=60s" in unit


def test_server_unit_unmounts_overlay_via_exec_stop_post():
    """Single source of truth for unmount, mirroring the mount path.
    ExecStopPost (not ExecStop) so it runs after srcds has fully exited
    and the cgroup is cleared.

    Same nsenter-at-Exec-line wrapping as ExecStartPre — without it,
    the helper process would itself hold a reference to the unit's
    per-service mount namespace, and umount in PID 1 would loop on
    EBUSY until the helper gave up. With it, umount succeeds first try.
    """
    unit = SERVER_UNIT.read_text()
    assert (
        "ExecStopPost=+/usr/bin/nsenter --mount=/proc/1/ns/mnt -- "
        "/usr/local/libexec/left4me/left4me-overlay umount %i"
        in unit
    )


def test_server_unit_contains_perf_baseline_directives():
    unit = SERVER_UNIT.read_text()

    # Slice membership.
    assert "Slice=l4d2-game.slice" in unit

    # CFS priority bump (no SCHED_FIFO).
    assert "Nice=-5" in unit
    assert "CPUSchedulingPolicy=" not in unit

    # I/O priority.
    assert "IOSchedulingClass=best-effort" in unit
    assert "IOSchedulingPriority=4" in unit

    # OOM ordering: game servers survive, sandbox dies first.
    assert "OOMScoreAdjust=-200" in unit

    # Memory caps with headroom for map-load spikes.
    assert "MemoryHigh=1.5G" in unit
    assert "MemoryMax=2G" in unit

    # Bounded fork surface.
    assert "TasksMax=256" in unit

    # Plenty of fds for plugin-heavy setups.
    assert "LimitNOFILE=65536" in unit

    # srcds clean shutdown via SIGINT, with time to flush. With the
    # helper running in PID 1's mount namespace (via the unit-level
    # nsenter on ExecStopPost), umount has no race window and the
    # default 15 s is plenty for the whole stop transition.
    assert "KillSignal=SIGINT" in unit
    assert "TimeoutStopSec=15s" in unit

    # Per-unit override of journald rate limiting (default drops srcds output).
    assert "LogRateLimitIntervalSec=0" in unit


def test_l4d2_game_slice_exists_with_high_weights():
    assert GAME_SLICE.is_file()
    text = GAME_SLICE.read_text()
    assert "[Slice]" in text
    assert "CPUWeight=1000" in text
    assert "IOWeight=1000" in text


def test_l4d2_build_slice_exists_with_low_weights():
    assert BUILD_SLICE.is_file()
    text = BUILD_SLICE.read_text()
    assert "[Slice]" in text
    assert "CPUWeight=10" in text
    assert "IOWeight=10" in text


def test_sysctl_conf_present_with_perf_settings():
    assert SYSCTL_CONF.is_file()
    text = SYSCTL_CONF.read_text()
    for line in (
        "net.core.rmem_max = 8388608",
        "net.core.wmem_max = 8388608",
        "net.core.rmem_default = 524288",
        "net.core.wmem_default = 524288",
        "net.core.netdev_max_backlog = 5000",
        "net.core.netdev_budget = 600",
        "vm.swappiness = 10",
        "net.ipv4.udp_rmem_min = 16384",
        "net.ipv4.udp_wmem_min = 16384",
        "net.core.default_qdisc = fq_codel",
        "net.ipv4.tcp_congestion_control = bbr",
    ):
        assert line in text, f"missing {line!r} in 99-left4me.conf"


def test_env_templates_contain_required_defaults():
    host_env = HOST_ENV.read_text()
    assert "Deployment units use fixed /var/lib/left4me paths" in host_env
    assert host_env.endswith("LEFT4ME_ROOT=/var/lib/left4me\n")
    web_env = WEB_ENV_TEMPLATE.read_text()
    assert web_env.startswith(
        "DATABASE_URL=sqlite:////var/lib/left4me/left4me.db\n"
        "SECRET_KEY=replace-with-generated-secret\n"
        "JOB_WORKER_THREADS=4\n"
    )
    assert web_env.rstrip().endswith("STEAM_WEB_API_KEY=")


def test_sandbox_resolv_conf_exists():
    assert SANDBOX_RESOLV_CONF.is_file()
    text = SANDBOX_RESOLV_CONF.read_text()
    nameservers = [
        line.split()[1]
        for line in text.splitlines()
        if line.startswith("nameserver ")
    ]
    assert len(nameservers) >= 2, "expected at least two nameservers for redundancy"
    # Sanity: the resolvers must be public (not RFC1918 / loopback). We don't
    # pin the exact IPs — Cloudflare/Google/Quad9 are all acceptable.
    for ns in nameservers:
        assert not ns.startswith("127."), ns
        assert not ns.startswith("10."), ns
        assert not ns.startswith("192.168."), ns
        first_octet = int(ns.split(".")[0])
        # Reject 172.16.0.0/12.
        if first_octet == 172:
            second_octet = int(ns.split(".")[1])
            assert not (16 <= second_octet <= 31), ns