diff --git a/core/private_storage.py b/core/private_storage.py index e4e662ac..da7808a9 100644 --- a/core/private_storage.py +++ b/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) diff --git a/tests/test_private_storage_windows.py b/tests/test_private_storage_windows.py new file mode 100644 index 00000000..2c96c1d0 --- /dev/null +++ b/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)