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>
170 lines
6 KiB
Python
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
|