From 1513e0a9f5a5bd557bb4b668cd665cf0e7132e40 Mon Sep 17 00:00:00 2001 From: Raymond Ginger Date: Tue, 11 Aug 2026 10:41:49 +0800 Subject: [PATCH 1/4] fix(security): restrict Windows private files with fail-safe ACL ordering --- core/private_storage.py | 74 ++++++++++++++++++++++++++++++++++++++--- 1 file changed, 70 insertions(+), 4 deletions(-) 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) From 6584b33bd11395ab2c49abb74a4e9da8598ea99f Mon Sep 17 00:00:00 2001 From: Raymond Ginger Date: Tue, 11 Aug 2026 10:41:56 +0800 Subject: [PATCH 2/4] test(security): cover Windows private-file ACL restriction --- tests/test_private_storage_windows.py | 122 ++++++++++++++++++++++++++ 1 file changed, 122 insertions(+) create mode 100644 tests/test_private_storage_windows.py diff --git a/tests/test_private_storage_windows.py b/tests/test_private_storage_windows.py new file mode 100644 index 00000000..8a22a1e1 --- /dev/null +++ b/tests/test_private_storage_windows.py @@ -0,0 +1,122 @@ +"""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) From 08572579ca90bc8351a06895493d833312f313f7 Mon Sep 17 00:00:00 2001 From: Raymond Ginger Date: Tue, 11 Aug 2026 10:42:41 +0800 Subject: [PATCH 3/4] ci(windows): run Windows private-storage ACL tests in windows-lifecycle --- .github/workflows/python-ci.yml | 9 +++++---- 1 file changed, 5 insertions(+), 4 deletions(-) diff --git a/.github/workflows/python-ci.yml b/.github/workflows/python-ci.yml index 41f2a63a..96dea64e 100644 --- a/.github/workflows/python-ci.yml +++ b/.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 From cf12926365a7d66aea06b7eb81edc72e64ce937b Mon Sep 17 00:00:00 2001 From: DeepCode Date: Sun, 16 Aug 2026 09:13:05 +0800 Subject: [PATCH 4/4] fix(security): apply Windows ACL only at file creation, never per open Addresses maintainer feedback on the earlier ACL PR (#148): re-running icacls on every open costs two subprocesses per call for no change. - open_private_file now restricts a file's ACL only when it was just created (target did not exist before os.open); opening an existing private file never re-runs _restrict_windows_acl. - harden_private_tree keeps forcing the restriction (it repairs legacy trees whose ACLs may be absent), so its semantics are unchanged. - new cross-platform tests (mock _restrict_windows_acl) assert: new file restricted exactly once, existing file never re-restricted, read-only open of an existing file never restricts. --- core/private_storage.py | 17 ++++-- tests/test_private_storage_acl_once.py | 75 ++++++++++++++++++++++++++ tests/test_private_storage_windows.py | 33 ++++++++++++ 3 files changed, 121 insertions(+), 4 deletions(-) create mode 100644 tests/test_private_storage_acl_once.py diff --git a/core/private_storage.py b/core/private_storage.py index da7808a9..e90b4990 100644 --- a/core/private_storage.py +++ b/core/private_storage.py @@ -100,10 +100,10 @@ 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 @@ -111,6 +111,11 @@ 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, @@ -128,7 +133,7 @@ def open_private_file(path: Path | str, flags: int) -> int: ) if os.name != "nt": os.fchmod(descriptor, PRIVATE_FILE_MODE) - else: + elif created: _restrict_windows_acl(target) return descriptor except BaseException: @@ -200,8 +205,12 @@ 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: diff --git a/tests/test_private_storage_acl_once.py b/tests/test_private_storage_acl_once.py new file mode 100644 index 00000000..72840d8d --- /dev/null +++ b/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" + ) diff --git a/tests/test_private_storage_windows.py b/tests/test_private_storage_windows.py index 8a22a1e1..5393f97c 100644 --- a/tests/test_private_storage_windows.py +++ b/tests/test_private_storage_windows.py @@ -120,3 +120,36 @@ def test_windows_harden_private_tree_restricts_every_entry(tmp_path: Path) -> No 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"