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()) == []