Fix native reader permissions during benchmark CLI activation

benchmark/edgebench/cli_source_probe.py

@@ -108,6 +108,33 @@ def _copy(container, observer, remote: str, immutable_roots=()) -> None:
             raise RuntimeError("Observer source transport was not confirmed")
 
 
+def _prepare_observer_native_home(observer, *, python: str, home: str, user: str) -> bool:
+    """Provision the native reader's writable root in the disposable observer."""
+    identity = observer.exec_run(
+        [python, "-I", "-B", "-c", "import os; print(os.geteuid(), os.getegid())"], user=user,
+    )
+    owner = re.fullmatch(rb"([0-9]+) ([0-9]+)", identity.output.strip()) if identity.exit_code == 0 else None
+    if not owner:
+        return False
+    # Only approved store members are copied later. Creating their parent as
+    # root via _copy would leave the native reader unable to initialize.
+    prepare = (
+        "import os, sys\nfrom pathlib import Path\n"
+        "home = Path(sys.argv[1])\n"
+        "if home.resolve() != home: raise ValueError('Observer native home must be physical')\n"
+        "home.mkdir(mode=0o700, parents=True, exist_ok=True)\n"
+        "if home.resolve() != home: raise ValueError('Observer native home must be physical')\n"
+        "os.chown(home, int(sys.argv[2]), int(sys.argv[3]), follow_symlinks=False)\n"
+        "home.chmod(0o700)\n"
+    )
+    result = observer.exec_run(
+        [python, "-I", "-B", "-c", prepare, str(_guest_path(home)),
+         owner[1].decode("ascii"), owner[2].decode("ascii")],
+        user="root",
+    )
+    return result.exit_code == 0
+
+
 def _file_bytes(container, path):
     with _archive_response(container, path) as (chunks, _), tempfile.TemporaryFile() as spool:
         for block in chunks:
@@ -180,7 +207,7 @@ def __init__(self, client, *, declaration: dict, components: list[dict],
             _guest_path(row["source_path"])
         for path in immutable_roots:
             _guest_path(path)
-        home = PurePosixPath(declaration["native_home"])
+        home = _guest_path(declaration["native_home"])
         # Copy only the native owner's declared stores/config/hooks, never auth
         # or the entire credential home, even if the caller supplied that root.
         paths = [PurePosixPath(r["source_path"]) for r in components] + list(map(PurePosixPath, immutable_roots))
@@ -278,6 +305,12 @@ def __call__(self, container, plan):
         try:
             for path in self.immutable_roots:
                 _copy(container, observer, path, self.immutable_roots)
+            if not _prepare_observer_native_home(
+                observer, python=self.declaration["python"], home=self.declaration["native_home"], user=self.user,
+            ):
+                facts["observer_error_kind"] = "ObserverNativeHomeUnavailable"
+                facts["source_witness_after"] = self.witness(container, plan)
+                return facts
             reads = [self.declaration[key] for key in ("registry", "runtime_root", "scheduler_state")]
             reads += self.declaration["authority_paths"]
             reads += [self.declaration["scheduler_state"] + ".host-attempt",

benchmark/tests/test_cli_source_activation.py

@@ -18,14 +18,41 @@
 import pytest
 
 from benchmark.edgebench import cli_source_activation as activation
-from benchmark.edgebench.cli_source_probe import archive_witness, _archive_response, _path_stat
+from benchmark.edgebench.cli_source_probe import (
+    archive_witness, _archive_response, _path_stat, _prepare_observer_native_home,
+)
 from benchmark.edgebench.cli_source_release import StagedCliRelease
 
 
 pytestmark = pytest.mark.skipif(os.environ.get("LOOPX_CLI_ACTIVATION_DOCKER_SMOKE") != "1",
     reason="Requires explicit isolated Docker CLI-activation qualification")
 
 
+def test_native_observer_reader_can_initialize_its_private_home():
+    import docker
+
+    client = docker.from_env(timeout=15)
+    image = os.environ.get("LOOPX_CHECKPOINT_WRITER_IMAGE", "python:3.12-slim")
+    client.images.get(image)
+    box = client.containers.run(image, ["sleep", "180"], detach=True, network_disabled=True,
+        mem_limit="64m", pids_limit=32, labels={"loopx.cli-activation-test": uuid.uuid4().hex})
+    python, home, reader = "/usr/local/bin/python", "/profile/native", "65534:65534"
+    try:
+        assert box.exec_run(["mkdir", "-p", home], user="root").exit_code == 0
+        command = [python, "-I", "-B", "-c", "from pathlib import Path; import sys; "
+                   "home = Path(sys.argv[1]); assert not (home / 'auth.json').exists(); "
+                   "(home / 'native-reader-state').write_bytes(b'initialized')", home]
+        assert box.exec_run(command, user=reader).exit_code != 0
+        assert _prepare_observer_native_home(box, python=python, home=home, user=reader)
+        assert box.exec_run(command, user=reader).exit_code == 0
+        assert box.exec_run(["cat", home + "/native-reader-state"], user=reader).output == b"initialized"
+        other = box.exec_run(["ls", home], user="65533:65533")
+        assert other.exit_code != 0
+    finally:
+        box.remove(force=True)
+        client.close()
+
+
 @pytest.fixture
 def source(tmp_path):
     import docker