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,22 @@ 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)
 
     directory.mkdir(parents=True, exist_ok=True, mode=PRIVATE_DIRECTORY_MODE)
-    _chmod(directory, PRIVATE_DIRECTORY_MODE)
+    _chmod(directory, PRIVATE_DIRECTORY_MODE, force=True)
     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 +133,8 @@ def open_private_file(path: Path | str, flags: int) -> int:
             )
         if os.name != "nt":
             os.fchmod(descriptor, PRIVATE_FILE_MODE)
+        elif created:
+            _restrict_windows_acl(target)
         return descriptor
     except BaseException:
         os.close(descriptor)
@@ -119,8 +191,6 @@ 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)
@@ -135,8 +205,13 @@ def harden_private_tree(root: Path | str) -> Path:
     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 deliberately re-applies the restriction even to
+        # existing paths (it repairs legacy trees whose ACLs may be absent or
+        # permissive), so _chmod restricts unconditionally. The per-open cost
+        # is avoided in open_private_file by only restricting new files.
+        _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"
+    )

tests/test_private_storage_windows.py

@@ -0,0 +1,155 @@
+"""Windows NTFS ACL restriction tests for core.private_storage.
+
+These tests assert that private directories and files are restricted to the
+current user with full control and that dangerous well-known ACEs (Everyone,
+Authenticated Users, BUILTIN\\Users) are removed after the restriction runs.
+They are skipped on non-Windows platforms.
+"""
+
+from __future__ import annotations
+
+import os
+import subprocess
+from pathlib import Path
+
+import pytest
+
+from core.private_storage import (
+    ensure_private_directory,
+    harden_private_tree,
+    open_private_file,
+)
+
+pytestmark = pytest.mark.skipif(
+    os.name != "nt",
+    reason="NTFS ACL restriction applies on Windows only",
+)
+
+_DANGEROUS_ACES = ("Authenticated Users", "BUILTIN\\Users", "Everyone")
+
+
+def _windows_identity() -> str:
+    completed = subprocess.run(
+        ["whoami"],
+        capture_output=True,
+        text=True,
+        encoding="mbcs",
+        errors="replace",
+        timeout=5,
+        check=True,
+    )
+    return (completed.stdout or "").strip()
+
+
+def _acl_lines(path: Path) -> list[str]:
+    completed = subprocess.run(
+        ["icacls", os.fspath(path)],
+        capture_output=True,
+        text=True,
+        encoding="mbcs",
+        errors="replace",
+        timeout=15,
+        check=True,
+    )
+    return [
+        line.strip() for line in (completed.stdout or "").splitlines() if ":" in line
+    ]
+
+
+def _assert_no_dangerous_aces(path: Path) -> None:
+    lines = _acl_lines(path)
+    joined = "\n".join(lines).lower()
+    for ace in _DANGEROUS_ACES:
+        assert ace.lower() not in joined, (
+            f"{path} still exposes dangerous ACE {ace!r}:\n{joined}"
+        )
+
+
+def _assert_current_user_has_full_control(path: Path) -> None:
+    identity = _windows_identity().lower()
+    # ``whoami`` may return either ``domain\user`` or a bare ``user`` depending
+    # on which binary is on PATH, while ``icacls`` always prints the fully
+    # qualified principal.  Compare the last path segment so both match.
+    short_name = identity.rsplit("\\", 1)[-1]
+    lines = _acl_lines(path)
+    for line in lines:
+        # Split from the right: icacls lines start with a Windows path that
+        # contains a drive-letter colon (``C:\\...``), so the first colon is
+        # not the principal/rights separator.
+        principal, rights = line.rsplit(":", 1)
+        principal_short = principal.strip().lower().rsplit("\\", 1)[-1]
+        if principal_short == short_name:
+            assert "(f)" in rights.lower(), (
+                f"{path} does not grant the current user full control:\n{line}"
+            )
+            return
+    raise AssertionError(
+        f"{path} has no ACE for the current user {identity!r}:\n" + "\n".join(lines)
+    )
+
+
+def test_windows_private_directory_is_restricted(tmp_path: Path) -> None:
+    directory = ensure_private_directory(tmp_path / "private" / "nested")
+
+    _assert_no_dangerous_aces(directory)
+    _assert_current_user_has_full_control(directory)
+
+
+def test_windows_private_file_is_restricted(tmp_path: Path) -> None:
+    target = tmp_path / "private" / "credentials.json"
+    descriptor = open_private_file(target, os.O_WRONLY | os.O_CREAT)
+    try:
+        os.write(descriptor, b"secret")
+    finally:
+        os.close(descriptor)
+
+    _assert_no_dangerous_aces(target)
+    _assert_current_user_has_full_control(target)
+    assert target.read_bytes() == b"secret"
+
+
+def test_windows_harden_private_tree_restricts_every_entry(tmp_path: Path) -> None:
+    root = tmp_path / "legacy-private"
+    session = root / "session-1"
+    session.mkdir(parents=True)
+    (session / "session.jsonl").write_text("legacy\n", encoding="utf-8")
+    (root / "settings.json").write_text("{}", encoding="utf-8")
+
+    harden_private_tree(root)
+
+    for path in (root, session, session / "session.jsonl", root / "settings.json"):
+        _assert_no_dangerous_aces(path)
+        _assert_current_user_has_full_control(path)
+
+
+def test_windows_open_existing_file_does_not_rerun_acl(monkeypatch, tmp_path: Path) -> None:
+    """The per-open ACL re-run is gone: opening an existing private file does
+    not call _restrict_windows_acl again (the ACL was applied at creation)."""
+    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 = ps.open_private_file(target, os.O_CREAT | os.O_RDWR)
+    os.close(fd)
+    assert len(calls) == 1, "new file must be restricted exactly once"
+
+    # Second open: file exists → must NOT re-run the ACL restriction.
+    fd = ps.open_private_file(target, os.O_RDWR)
+    os.close(fd)
+    assert len(calls) == 1, "existing file must not re-run the ACL restriction"
+
+
+def test_windows_open_created_file_restricts_once(monkeypatch, tmp_path: Path) -> None:
+    """A file created via open_private_file is restricted exactly once."""
+    import core.private_storage as ps
+
+    calls = []
+    monkeypatch.setattr(ps, "_restrict_windows_acl", lambda p: calls.append(p))
+
+    for _ in range(3):
+        fd = ps.open_private_file(tmp_path / "fresh.jsonl", os.O_CREAT | os.O_RDWR)
+        os.close(fd)
+    assert len(calls) == 1, "created once, restricted once, never again"