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

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."""
 
@@ -63,6 +128,8 @@ def open_private_file(path: Path | str, flags: int) -> int:
             )
         if os.name != "nt":
             os.fchmod(descriptor, PRIVATE_FILE_MODE)
+        else:
+            _restrict_windows_acl(target)
         return descriptor
     except BaseException:
         os.close(descriptor)
@@ -119,8 +186,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)
@@ -137,6 +202,7 @@ def harden_private_tree(root: Path | str) -> Path:
 
 def _chmod(path: Path, mode: int) -> None:
     if os.name == "nt":
+        _restrict_windows_acl(path)
         return
     try:
         os.chmod(path, mode, follow_symlinks=False)

tests/test_private_storage_windows.py

@@ -0,0 +1,124 @@
+"""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)