186 lines
7.2 KiB
Python
186 lines
7.2 KiB
Python
from __future__ import annotations
|
|
|
|
import os
|
|
import select
|
|
import shutil
|
|
import stat
|
|
import tempfile
|
|
from pathlib import Path
|
|
|
|
import pytest
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="POSIX non-root runtime lock regression")
|
|
def test_forge_oidc_lock_works_for_nonroot_owner_of_preprovisioned_runtime_dirs():
|
|
"""Exercise the real file lock after image setup, as the app UID."""
|
|
from backend.app.services import forge_oidc_provisioning
|
|
|
|
was_root = os.geteuid() == 0
|
|
runtime_uid, runtime_gid = (100, 101) if was_root else (os.geteuid(), os.getegid())
|
|
fixture_root = Path(tempfile.mkdtemp(prefix="forge-oidc-runtime-lock-", dir="/tmp"))
|
|
runtime_parent = fixture_root / "itworx-appops"
|
|
runtime_root = runtime_parent / "forge-oidc-provision"
|
|
read_fd, write_fd = os.pipe()
|
|
child_pid: int | None = None
|
|
|
|
try:
|
|
fixture_root.chmod(0o755)
|
|
runtime_parent.mkdir(mode=0o700)
|
|
runtime_root.mkdir(mode=0o700)
|
|
if was_root:
|
|
os.chown(runtime_parent, runtime_uid, runtime_gid)
|
|
os.chown(runtime_root, runtime_uid, runtime_gid)
|
|
|
|
for path in (runtime_parent, runtime_root):
|
|
info = path.stat(follow_symlinks=False)
|
|
assert (info.st_uid, info.st_gid, stat.S_IMODE(info.st_mode)) == (
|
|
runtime_uid,
|
|
runtime_gid,
|
|
0o700,
|
|
)
|
|
|
|
# This private test seam mirrors the fixed /run/... path without
|
|
# creating or changing anything under the host's real /run.
|
|
monkeypatch_lock_root = forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT
|
|
forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT = runtime_root
|
|
child_pid = os.fork()
|
|
if child_pid != 0:
|
|
forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT = monkeypatch_lock_root
|
|
|
|
if child_pid == 0:
|
|
os.close(read_fd)
|
|
try:
|
|
if was_root:
|
|
os.setgroups([])
|
|
os.setgid(runtime_gid)
|
|
os.setuid(runtime_uid)
|
|
if os.geteuid() == 0:
|
|
raise AssertionError("worker remained root")
|
|
with forge_oidc_provisioning._forge_oidc_process_lock():
|
|
os.write(
|
|
write_fd,
|
|
f"uid={os.geteuid()};gid={os.getegid()};lock=acquired".encode(),
|
|
)
|
|
os._exit(0)
|
|
except BaseException as exc: # noqa: BLE001 - send only a safe failure class.
|
|
os.write(write_fd, f"worker_error={type(exc).__name__}".encode())
|
|
os._exit(1)
|
|
|
|
os.close(write_fd)
|
|
write_fd = -1
|
|
ready, _, _ = select.select([read_fd], [], [], 5.0)
|
|
assert ready, "unprivileged lock attempt did not finish"
|
|
result = os.read(read_fd, 256).decode("ascii", "strict")
|
|
waited_pid, status = os.waitpid(child_pid, 0)
|
|
child_pid = None
|
|
|
|
assert waited_pid > 0
|
|
assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, result
|
|
assert result == f"uid={runtime_uid};gid={runtime_gid};lock=acquired"
|
|
|
|
lock_path = runtime_root / forge_oidc_provisioning._PROVISION_LOCK_FILENAME
|
|
lock_stat = lock_path.stat(follow_symlinks=False)
|
|
assert stat.S_ISREG(lock_stat.st_mode)
|
|
assert lock_stat.st_nlink == 1
|
|
assert (lock_stat.st_uid, lock_stat.st_gid, stat.S_IMODE(lock_stat.st_mode)) == (
|
|
runtime_uid,
|
|
runtime_gid,
|
|
0o600,
|
|
)
|
|
finally:
|
|
if child_pid is not None:
|
|
try:
|
|
os.kill(child_pid, 9)
|
|
except ProcessLookupError:
|
|
pass
|
|
os.waitpid(child_pid, 0)
|
|
for fd in (read_fd, write_fd):
|
|
if fd >= 0:
|
|
os.close(fd)
|
|
shutil.rmtree(fixture_root)
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="POSIX private runtime path regression")
|
|
def test_forge_oidc_lock_does_not_create_missing_runtime_dirs_under_root_run_parent():
|
|
"""A missing image-initialized tree still fails closed for the app UID."""
|
|
from backend.app.services import forge_oidc_provisioning
|
|
|
|
was_root = os.geteuid() == 0
|
|
runtime_uid, runtime_gid = (100, 101) if was_root else (os.geteuid(), os.getegid())
|
|
fixture_root = Path(tempfile.mkdtemp(prefix="forge-oidc-runtime-missing-", dir="/tmp"))
|
|
runtime_root = fixture_root / "itworx-appops" / "forge-oidc-provision"
|
|
read_fd, write_fd = os.pipe()
|
|
child_pid: int | None = None
|
|
|
|
try:
|
|
fixture_root.chmod(0o755)
|
|
monkeypatch_lock_root = forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT
|
|
forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT = runtime_root
|
|
child_pid = os.fork()
|
|
if child_pid != 0:
|
|
forge_oidc_provisioning._FORGE_OIDC_LOCK_ROOT = monkeypatch_lock_root
|
|
|
|
if child_pid == 0:
|
|
os.close(read_fd)
|
|
try:
|
|
if was_root:
|
|
os.setgroups([])
|
|
os.setgid(runtime_gid)
|
|
os.setuid(runtime_uid)
|
|
if os.geteuid() == 0:
|
|
raise AssertionError("worker remained root")
|
|
with forge_oidc_provisioning._forge_oidc_process_lock():
|
|
os.write(write_fd, b"unexpectedly_acquired")
|
|
os._exit(0)
|
|
except forge_oidc_provisioning._ProvisionLockUnavailable:
|
|
os.write(write_fd, b"unavailable")
|
|
os._exit(0)
|
|
except BaseException as exc: # noqa: BLE001 - send only a safe failure class.
|
|
os.write(write_fd, f"worker_error={type(exc).__name__}".encode())
|
|
os._exit(1)
|
|
|
|
os.close(write_fd)
|
|
write_fd = -1
|
|
ready, _, _ = select.select([read_fd], [], [], 5.0)
|
|
assert ready, "unprivileged lock attempt did not finish"
|
|
result = os.read(read_fd, 256).decode("ascii", "strict")
|
|
waited_pid, status = os.waitpid(child_pid, 0)
|
|
child_pid = None
|
|
|
|
assert waited_pid > 0
|
|
assert os.WIFEXITED(status) and os.WEXITSTATUS(status) == 0, result
|
|
assert result == "unavailable"
|
|
assert not runtime_root.parent.exists()
|
|
assert not runtime_root.exists()
|
|
finally:
|
|
if child_pid is not None:
|
|
try:
|
|
os.kill(child_pid, 9)
|
|
except ProcessLookupError:
|
|
pass
|
|
os.waitpid(child_pid, 0)
|
|
for fd in (read_fd, write_fd):
|
|
if fd >= 0:
|
|
os.close(fd)
|
|
shutil.rmtree(fixture_root)
|
|
|
|
|
|
@pytest.mark.skipif(os.name == "nt", reason="POSIX private runtime path regression")
|
|
def test_forge_oidc_lock_rejects_symlink_runtime_root_without_touching_target(
|
|
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
|
|
):
|
|
"""The fixed runtime path is checked without following a symlink."""
|
|
from backend.app.services import forge_oidc_provisioning
|
|
|
|
target = tmp_path / "private-target"
|
|
target.mkdir(mode=0o700)
|
|
link = tmp_path / "runtime-link"
|
|
link.symlink_to(target, target_is_directory=True)
|
|
monkeypatch.setattr(forge_oidc_provisioning, "_FORGE_OIDC_LOCK_ROOT", link)
|
|
|
|
with pytest.raises(forge_oidc_provisioning._ProvisionLockUnavailable):
|
|
with forge_oidc_provisioning._forge_oidc_process_lock():
|
|
pytest.fail("symlinked runtime root must not be used")
|
|
|
|
assert list(target.iterdir()) == []
|