Publish curated Unraid-AppOps-Gateway source from a480f280f26f4aa48c61394fa841d2233067c208

backend/tests/test_forge_oidc_runtime_lock.py

@@ -0,0 +1,185 @@
+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()) == []