fix(security): restrict Windows private files with fail-safe ACL ordering

.github/workflows/python-ci.yml

@@ -64,12 +64,13 @@ jobs:
           tests/application/test_session_deletion_service.py
           tests/application/test_execution_coordinator.py
 
-      # These suites carry Windows-gated cases (the Job Object backend) that
-      # the ubuntu job can only skip. This is the sole place they actually
-      # execute.
-      - name: Verify Job Object sandbox
+      # These suites carry Windows-gated cases (NTFS ACLs, the Job Object
+      # backend) that the ubuntu job can only skip. This is the sole place
+      # they actually execute.
+      - name: Verify Windows ACLs and Job Object sandbox
         run: >-
           python -m pytest -q
+          tests/test_private_storage_windows.py
           tests/test_harness_sandbox.py
           tests/test_exec_sandbox_wiring.py
 

core/private_storage.py

@@ -5,14 +5,17 @@
 the process umask, which is commonly permissive on desktop systems.
 
 POSIX permissions are repaired to ``0700`` for directories and ``0600`` for
-regular files.  Windows access control is inherited from the user's profile;
-the mode arguments are still supplied at creation time where supported.
+regular files.  On Windows the current user is granted full control and the
+inherited access entries are then stripped; the restriction is applied in a
+fail-safe order so a failed grant leaves the inherited ACLs untouched and the
+path stays accessible.
 """
 
 from __future__ import annotations
 
 import os
 import stat
+import subprocess
 from pathlib import Path
 
 PRIVATE_DIRECTORY_MODE = 0o700
@@ -23,6 +26,68 @@ class UnsafePrivateFileError(OSError):
     """A private-state path is not a regular file owned by this path entry."""
 
 
+def _windows_identity() -> str | None:
+    """Return the fully-qualified current user (``DOMAIN\\user``) on Windows."""
+
+    if os.name != "nt":
+        return None
+    try:
+        completed = subprocess.run(
+            ["whoami"],
+            capture_output=True,
+            text=True,
+            encoding="mbcs",
+            errors="replace",
+            timeout=5,
+            check=True,
+        )
+    except (OSError, subprocess.SubprocessError):
+        return None
+    principal = (completed.stdout or "").strip()
+    return principal or None
+
+
+def _restrict_windows_acl(path: Path) -> None:
+    """Restrict ``path`` to the current user, failing safe.
+
+    The current user is granted full control **before** inherited access
+    entries are stripped.  If the grant fails (service account, transient
+    timeout, ...) the inherited ACLs are left untouched so the path stays
+    accessible to the caller; the previous strip-first order could leave a
+    path with no usable ACE and make it unopenable.
+    """
+
+    identity = _windows_identity()
+    if identity is None:
+        return
+    try:
+        subprocess.run(
+            ["icacls", os.fspath(path), "/grant:r", f"{identity}:F"],
+            capture_output=True,
+            text=True,
+            encoding="mbcs",
+            errors="replace",
+            timeout=15,
+            check=True,
+        )
+    except (OSError, subprocess.SubprocessError):
+        # Fail safe: keep the inherited ACLs; the path stays accessible.
+        return
+    try:
+        subprocess.run(
+            ["icacls", os.fspath(path), "/inheritance:r"],
+            capture_output=True,
+            text=True,
+            encoding="mbcs",
+            errors="replace",
+            timeout=15,
+            check=True,
+        )
+    except (OSError, subprocess.SubprocessError):
+        # Strip failed: the path is merely less restricted, still usable.
+        pass
+
+
 def ensure_private_directory(path: Path | str) -> Path:
     """Create ``path`` and make every newly created component user-private."""
 
@@ -35,17 +100,26 @@ def ensure_private_directory(path: Path | str) -> Path:
 
     for component in reversed(missing):
         component.mkdir(mode=PRIVATE_DIRECTORY_MODE, exist_ok=True)
-        _chmod(component, PRIVATE_DIRECTORY_MODE)
+        _chmod(component, PRIVATE_DIRECTORY_MODE, force=True)
 
+    was_missing = not directory.exists()
     directory.mkdir(parents=True, exist_ok=True, mode=PRIVATE_DIRECTORY_MODE)
-    _chmod(directory, PRIVATE_DIRECTORY_MODE)
+    # Restrict only what this call actually created. An existing directory was
+    # restricted at its own creation; re-running icacls on it on every open
+    # costs two subprocesses without changing the ACL.
+    _chmod(directory, PRIVATE_DIRECTORY_MODE, force=was_missing)
     return directory
 
 
 def open_private_file(path: Path | str, flags: int) -> int:
     """Open a private regular file without following a final symlink."""
 
     target = Path(path)
+    # Only restrict a *newly created* file. An existing file was already
+    # restricted at creation; re-running icacls on every open costs two
+    # subprocesses per call (and a full tree walk many times over) without
+    # changing the ACL (maintainer feedback on the earlier ACL PR).
+    created = not target.exists()
     ensure_private_directory(target.parent)
     descriptor = os.open(
         target,
@@ -63,6 +137,8 @@ def open_private_file(path: Path | str, flags: int) -> int:
             )
         if os.name != "nt":
             os.fchmod(descriptor, PRIVATE_FILE_MODE)
+        if created:
+            _restrict_windows_acl(target)
         return descriptor
     except BaseException:
         os.close(descriptor)
@@ -112,31 +188,35 @@ def ensure_private_file(path: Path | str) -> None:
     except OSError:
         return
     if stat.S_ISREG(metadata.st_mode):
-        _chmod(target, PRIVATE_FILE_MODE)
+        _chmod(target, PRIVATE_FILE_MODE, force=True)
 
 
 def harden_private_tree(root: Path | str) -> Path:
     """Repair a DeepCode-owned tree while refusing to traverse symlinks."""
 
     base = ensure_private_directory(root)
-    if os.name == "nt":
-        return base
 
     for current, directories, files in os.walk(base, followlinks=False):
         current_path = Path(current)
-        _chmod(current_path, PRIVATE_DIRECTORY_MODE)
+        _chmod(current_path, PRIVATE_DIRECTORY_MODE, force=True)
         directories[:] = [
             name for name in directories if not (current_path / name).is_symlink()
         ]
         for name in directories:
-            _chmod(current_path / name, PRIVATE_DIRECTORY_MODE)
+            _chmod(current_path / name, PRIVATE_DIRECTORY_MODE, force=True)
         for name in files:
             ensure_private_file(current_path / name)
     return base
 
 
-def _chmod(path: Path, mode: int) -> None:
+def _chmod(path: Path, mode: int, *, force: bool = False) -> None:
     if os.name == "nt":
+        # harden_private_tree (force=True) deliberately re-applies the
+        # restriction even to existing paths (it repairs legacy trees whose
+        # ACLs may be absent or permissive). Default callers pass force=False
+        # so an already-restricted path is not re-churned on every open.
+        if force:
+            _restrict_windows_acl(path)
         return
     try:
         os.chmod(path, mode, follow_symlinks=False)

tests/test_private_storage_acl_once.py

@@ -0,0 +1,75 @@
+"""Cross-platform tests for the per-open ACL optimization in private_storage.
+
+The Windows ACL restriction is applied at file *creation*; opening an
+existing private file must not re-run icacls (maintainer feedback on the
+earlier ACL PR: "open_private_file() currently calls it on each call; once at
+creation is enough"). These tests mock `_restrict_windows_acl` to count calls,
+so they run on any platform.
+"""
+
+from __future__ import annotations
+
+import os
+import sys
+from pathlib import Path
+
+ROOT = Path(__file__).resolve().parents[1]
+if str(ROOT) not in sys.path:
+    sys.path.insert(0, str(ROOT))
+
+from core.private_storage import open_private_file
+
+
+def _file_calls(calls, target: Path) -> int:
+    """Count restrictions applied to the target file itself (excludes the
+    parent-directory restriction that ensure_private_directory performs)."""
+    return sum(1 for p in calls if Path(p) == target)
+
+
+def test_open_existing_file_does_not_rerun_acl(monkeypatch, tmp_path: Path) -> None:
+    import core.private_storage as ps
+
+    calls = []
+    monkeypatch.setattr(ps, "_restrict_windows_acl", lambda p: calls.append(p))
+
+    target = tmp_path / "existing.jsonl"
+    # First open: file does not exist → new → restrict once.
+    fd = open_private_file(target, os.O_CREAT | os.O_RDWR)
+    os.close(fd)
+    assert _file_calls(calls, target) == 1, "new file must be restricted exactly once"
+
+    # Second open: file exists → must NOT re-run the ACL restriction.
+    fd = open_private_file(target, os.O_RDWR)
+    os.close(fd)
+    assert _file_calls(calls, target) == 1, (
+        "existing file must not re-run the ACL restriction"
+    )
+
+
+def test_open_created_file_restricts_once(monkeypatch, tmp_path: Path) -> None:
+    import core.private_storage as ps
+
+    calls = []
+    monkeypatch.setattr(ps, "_restrict_windows_acl", lambda p: calls.append(p))
+
+    target = tmp_path / "fresh.jsonl"
+    for _ in range(3):
+        fd = open_private_file(target, os.O_CREAT | os.O_RDWR)
+        os.close(fd)
+    assert _file_calls(calls, target) == 1, "created once, restricted once, never again"
+
+
+def test_open_without_creat_never_restricts(monkeypatch, tmp_path: Path) -> None:
+    import core.private_storage as ps
+
+    calls = []
+    monkeypatch.setattr(ps, "_restrict_windows_acl", lambda p: calls.append(p))
+
+    target = tmp_path / "pre.jsonl"
+    target.write_text("x", encoding="utf-8")
+    # O_RDONLY (no O_CREAT) on an existing file → no new file → no restriction.
+    fd = open_private_file(target, os.O_RDONLY)
+    os.close(fd)
+    assert _file_calls(calls, target) == 0, (
+        "read-only open of an existing file must not restrict"
+    )