left4me/l4d2host/tests/test_overlay_helper.py
mwiegand 5eac51a93e
fix(deploy): wrap overlay helper with nsenter so it doesn't pin the unit's mount namespace
systemd's `+` Exec prefix removes sandbox/credentials but does NOT
detach from the unit's per-service mount namespace (created by
PrivateTmp/Protect*). The Python interpreter for the helper was
launched inside that namespace, and even though the helper internally
nsenter'd into PID 1 for the umount syscall, the calling Python
process itself never left the unit's namespace. Its existence pinned
the namespace alive, which kept the slave mount tree alive, which
made PID 1's umount return EBUSY for the entire duration of the
helper's run. The mount became unmountable the moment the helper
exited — empirically verified by polling /proc/*/ns/mnt during stop:
the only PID holding the dying namespace was the helper itself.

Wrap both ExecStartPre and ExecStopPost with `/usr/bin/nsenter
--mount=/proc/1/ns/mnt --` so the helper Python interpreter runs in
PID 1's mount namespace from the start. With the helper out of the
unit's namespace, umount succeeds first try once the cgroup empties.
Reset went from ~25 s with retry/lazy-fallback workarounds to ~0.5 s
clean.

Knock-on cleanups:
- Helper drops internal nsenter for the syscalls (already in PID 1's
  namespace), and drops the eager-retry loop + lazy-umount fallback +
  inner work_inner retry (no race left to ride out).
- Revert TimeoutStopSec=60s back to 15s.
- Tests updated to expect the new argv shapes.

Co-Authored-By: Claude Opus 4.7 (1M context) <noreply@anthropic.com>
2026-05-09 15:13:59 +02:00

170 lines
6 KiB
Python

import os
import shlex
import subprocess
import sys
from pathlib import Path
import pytest
HELPER_SOURCE = (
Path(__file__).resolve().parents[2]
/ "deploy"
/ "files"
/ "usr"
/ "local"
/ "libexec"
/ "left4me"
/ "left4me-overlay"
)
def _setup_instance(root: Path, name: str = "alpha", lowerdirs: list[str] | None = None) -> None:
"""Create the on-disk shape the helper expects."""
(root / "installation").mkdir(parents=True, exist_ok=True)
(root / "overlays" / "workshop").mkdir(parents=True, exist_ok=True)
if lowerdirs is None:
lowerdirs = [str(root / "overlays" / "workshop"), str(root / "installation")]
inst_dir = root / "instances" / name
inst_dir.mkdir(parents=True)
(inst_dir / "instance.env").write_text(
f"L4D2_PORT=27015\nL4D2_ARGS=-tickrate 100\nL4D2_LOWERDIRS={':'.join(lowerdirs)}\n"
)
runtime = root / "runtime" / name
(runtime / "upper").mkdir(parents=True)
(runtime / "work").mkdir(parents=True)
(runtime / "merged").mkdir(parents=True)
def _run(args: list[str], root: Path, extra_env: dict[str, str] | None = None) -> subprocess.CompletedProcess:
env = {
**os.environ,
"LEFT4ME_ROOT": str(root),
"LEFT4ME_OVERLAY_PRINT_ONLY": "1",
}
if extra_env:
env.update(extra_env)
return subprocess.run(
[sys.executable, str(HELPER_SOURCE), *args],
env=env,
capture_output=True,
text=True,
)
def test_mount_prints_expected_command(tmp_path: Path) -> None:
"""The helper invokes /bin/mount directly. nsenter into PID 1's
mount namespace happens at the systemd Exec line (see the unit
file), so by the time the helper runs, the syscall already lands
in the right namespace.
"""
_setup_instance(tmp_path)
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode == 0, result.stderr
parts = shlex.split(result.stdout.strip())
assert parts[0] == "/bin/mount"
assert parts[1:3] == ["-t", "overlay"]
assert parts[3] == "overlay"
assert parts[4] == "-o"
options = parts[5]
assert f"upperdir={tmp_path}/runtime/alpha/upper" in options
assert f"workdir={tmp_path}/runtime/alpha/work" in options
assert f"lowerdir={tmp_path}/overlays/workshop:{tmp_path}/installation" in options
assert parts[6] == str(tmp_path / "runtime" / "alpha" / "merged")
def test_umount_prints_expected_command(tmp_path: Path) -> None:
"""Same as the mount path: helper invokes /bin/umount directly,
relying on the unit-level nsenter to put it in PID 1's namespace.
"""
_setup_instance(tmp_path)
result = _run(["umount", "alpha"], tmp_path)
assert result.returncode == 0, result.stderr
parts = shlex.split(result.stdout.strip())
assert parts == [
"/bin/umount",
str(tmp_path / "runtime" / "alpha" / "merged"),
]
@pytest.mark.parametrize("bad_name", ["..", "../escape", "FOO", "foo bar", "foo/bar", ""])
def test_rejects_bad_instance_name(tmp_path: Path, bad_name: str) -> None:
result = _run(["mount", bad_name], tmp_path)
assert result.returncode != 0
assert "invalid instance name" in result.stderr or "usage:" in result.stderr
def test_rejects_lowerdir_outside_allowlist(tmp_path: Path) -> None:
_setup_instance(tmp_path, lowerdirs=["/etc"])
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "outside the permitted roots" in result.stderr
def test_rejects_lowerdir_traversal(tmp_path: Path) -> None:
# An overlay subdirectory whose path uses .. to escape the overlays root.
_setup_instance(tmp_path, lowerdirs=[str(tmp_path / "overlays" / "..") + "/etc"])
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "outside the permitted roots" in result.stderr or "path does not exist" in result.stderr
def test_rejects_lowerdir_symlink_escape(tmp_path: Path) -> None:
_setup_instance(tmp_path)
sneaky = tmp_path / "overlays" / "sneaky"
os.symlink("/etc", sneaky)
# rewrite instance.env to point at the symlink
inst_env = tmp_path / "instances" / "alpha" / "instance.env"
inst_env.write_text(f"L4D2_LOWERDIRS={sneaky}\n")
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "outside the permitted roots" in result.stderr
def test_rejects_missing_instance_env(tmp_path: Path) -> None:
(tmp_path / "instances" / "alpha").mkdir(parents=True)
runtime = tmp_path / "runtime" / "alpha"
(runtime / "upper").mkdir(parents=True)
(runtime / "work").mkdir(parents=True)
(runtime / "merged").mkdir(parents=True)
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "instance.env not found" in result.stderr
def test_rejects_lowerdir_count_over_cap(tmp_path: Path) -> None:
(tmp_path / "installation").mkdir()
many = [str(tmp_path / "installation")] * 501
_setup_instance(tmp_path, lowerdirs=many)
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "501 entries" in result.stderr
def test_rejects_empty_lowerdir_entry(tmp_path: Path) -> None:
(tmp_path / "installation").mkdir()
_setup_instance(
tmp_path,
lowerdirs=[str(tmp_path / "installation"), "", str(tmp_path / "installation")],
)
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "empty entry" in result.stderr
@pytest.mark.skipif(sys.platform != "linux", reason="user.* xattrs are Linux-only")
def test_rejects_upperdir_with_fuseoverlayfs_xattr(tmp_path: Path) -> None:
_setup_instance(tmp_path)
tainted = tmp_path / "runtime" / "alpha" / "upper" / "deleted-thing"
tainted.write_bytes(b"")
try:
os.setxattr(tainted, "user.fuseoverlayfs.opaque", b"y")
except OSError:
pytest.skip("filesystem doesn't support user.* xattrs")
result = _run(["mount", "alpha"], tmp_path)
assert result.returncode != 0
assert "fuse-overlayfs xattr" in result.stderr