Publish curated Unraid-AppOps-Gateway source from e854334923cca574482d22b8424ba3ce46ec6aa1

backend/tests/test_controller_hub_release_volume.py

@@ -5,6 +5,7 @@ import importlib.util
 import io
 import json
 import os
+import stat
 import subprocess
 import sys
 import tarfile
@@ -717,7 +718,8 @@ def test_mirror_updates_stay_on_the_open_directory_after_path_swap(tmp_path, mon
 
 @pytest.mark.skipif(os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0,
                     reason="private /tmp ownership validation requires root on POSIX")
-def test_private_stage_rejects_symlinked_source_directory():
+@pytest.mark.parametrize("tmp_mode", [0o700, 0o1777])
+def test_private_stage_accepts_root_owned_tmp_with_only_approved_modes(monkeypatch, tmp_mode):
     import shutil
     import uuid
 
@@ -727,6 +729,101 @@ def test_private_stage_rejects_symlinked_source_directory():
     source.chmod(0o700)
     work.chmod(0o700)
     try:
+        tmp_info = os.stat("/tmp", follow_symlinks=False)
+        original_fstat = os.fstat
+
+        def fstat_with_tmp_mode(descriptor):
+            info = original_fstat(descriptor)
+            if (info.st_dev, info.st_ino) == (tmp_info.st_dev, tmp_info.st_ino):
+                return types.SimpleNamespace(
+                    st_uid=info.st_uid, st_gid=info.st_gid,
+                    st_mode=stat.S_IFDIR | tmp_mode, st_dev=info.st_dev,
+                )
+            return info
+
+        monkeypatch.setattr(os, "fstat", fstat_with_tmp_mode)
+        module._validate_private_stage(source)
+    finally:
+        monkeypatch.undo()
+        shutil.rmtree(work)
+
+
+@pytest.mark.skipif(os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0,
+                    reason="private /tmp ownership validation requires root on POSIX")
+@pytest.mark.parametrize(
+    ("target_name", "changes"),
+    [
+        ("tmp", {"st_mode": stat.S_IFDIR | 0o777}),
+        ("tmp", {"st_uid": 1000}),
+        ("tmp", {"st_gid": 1000}),
+        ("work", {"st_mode": stat.S_IFDIR | 0o755}),
+        ("source", {"st_mode": stat.S_IFDIR | 0o755}),
+        ("work", {"st_uid": 1000}),
+        ("work", {"st_gid": 1000}),
+        ("source", {"st_uid": 1000}),
+        ("source", {"st_gid": 1000}),
+        ("source", {"st_dev": 123456}),
+    ],
+)
+def test_private_stage_rejects_unsafe_tmp_and_child_metadata(monkeypatch, target_name, changes):
+    import shutil
+    import uuid
+
+    work = Path("/tmp") / f"itworx-controller-upgrade-{uuid.uuid4().hex[:12]}.Ab3xY7"
+    source = work / "source"
+    source.mkdir(parents=True, mode=0o700)
+    source.chmod(0o700)
+    work.chmod(0o700)
+    targets = {"tmp": Path("/tmp"), "work": work, "source": source}
+    target_info = os.stat(targets[target_name], follow_symlinks=False)
+    original_fstat = os.fstat
+
+    def fstat_with_unsafe_metadata(descriptor):
+        info = original_fstat(descriptor)
+        if (info.st_dev, info.st_ino) == (target_info.st_dev, target_info.st_ino):
+            values = {
+                "st_uid": info.st_uid, "st_gid": info.st_gid,
+                "st_mode": info.st_mode, "st_dev": info.st_dev,
+            }
+            values.update(changes)
+            return types.SimpleNamespace(**values)
+        return info
+
+    try:
+        monkeypatch.setattr(os, "fstat", fstat_with_unsafe_metadata)
+        with pytest.raises(module.VolumeTransitionError, match="private_stage_invalid"):
+            module._validate_private_stage(source)
+    finally:
+        monkeypatch.undo()
+        shutil.rmtree(work)
+
+
+@pytest.mark.skipif(os.name == "nt" or not hasattr(os, "geteuid") or os.geteuid() != 0,
+                    reason="private /tmp ownership validation requires root on POSIX")
+def test_private_stage_rejects_acl_and_symlinked_source_directory(monkeypatch):
+    import shutil
+    import uuid
+
+    work = Path("/tmp") / f"itworx-controller-upgrade-{uuid.uuid4().hex[:12]}.Ab3xY7"
+    source = work / "source"
+    source.mkdir(parents=True, mode=0o700)
+    source.chmod(0o700)
+    work.chmod(0o700)
+    source_info = os.stat(source, follow_symlinks=False)
+    original_listxattr = os.listxattr
+
+    def listxattr_with_acl(descriptor):
+        info = os.fstat(descriptor)
+        values = original_listxattr(descriptor)
+        if (info.st_dev, info.st_ino) == (source_info.st_dev, source_info.st_ino):
+            return [*values, "system.posix_acl_access"]
+        return values
+
+    try:
+        monkeypatch.setattr(os, "listxattr", listxattr_with_acl)
+        with pytest.raises(module.VolumeTransitionError, match="acl_unsafe"):
+            module._validate_private_stage(source)
+        monkeypatch.setattr(os, "listxattr", original_listxattr)
         module._validate_private_stage(source)
         source.rename(work / "source-real")
         source.symlink_to(work / "source-real", target_is_directory=True)