From a976c2c80f188b64f511c59adb9cca16e8e16f1d Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Tue, 18 Aug 2026 13:23:26 +0530 Subject: [PATCH 1/2] Harden history directory and append writes --- lib/python/base_cli/history.py | 43 ++++++++++++++++++++++++++++++---- tests/test_history.py | 25 ++++++++++++++++++++ 2 files changed, 64 insertions(+), 4 deletions(-) diff --git a/lib/python/base_cli/history.py b/lib/python/base_cli/history.py index 3dd851e..8de9cdf 100644 --- a/lib/python/base_cli/history.py +++ b/lib/python/base_cli/history.py @@ -17,7 +17,7 @@ except ImportError: # pragma: no cover - msvcrt is unavailable outside Windows. _msvcrt = None # type: ignore[assignment] -from ._private_files import restrict_file, write_private_json +from ._private_files import _open_parent_directory, restrict_directory, restrict_file, write_private_json from .exit_codes import ExitCode from .redaction import REDACTED, is_secret_key, option_name_to_parameter, redact_argv, redact_text_value @@ -164,9 +164,16 @@ def write_primary_record( def write_history_record(path: Path, record: dict[str, Any]) -> None: """Append one serialized record to a consumer-selected history path.""" + missing: list[Path] = [] + candidate = path.parent + while not candidate.exists(): + missing.append(candidate) + candidate = candidate.parent path.parent.mkdir(parents=True, exist_ok=True) + if os.name != "nt": + for directory in [path.parent, *missing]: + restrict_directory(directory) append_history_line(path, f"{json.dumps(record, sort_keys=True)}\n") - restrict_file(path) def update_run_metadata(run_root: Path, record: dict[str, Any]) -> None: @@ -206,10 +213,28 @@ def update_run_metadata(run_root: Path, record: dict[str, Any]) -> None: def append_history_line(path: Path, line: str) -> None: binary_flag = getattr(os, "O_BINARY", 0) - fd = os.open(path, os.O_WRONLY | os.O_CREAT | os.O_APPEND | binary_flag, 0o600) + open_flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND | binary_flag | getattr(os, "O_NOFOLLOW", 0) + parent_fd: int | None = None + if ( + os.name != "nt" + and hasattr(os, "O_NOFOLLOW") + and hasattr(os, "O_DIRECTORY") + and os.open in os.supports_dir_fd + ): + opened_parent_fd = _open_parent_directory(path.parent) + assert opened_parent_fd is not None + parent_fd = opened_parent_fd + try: + fd = os.open(path.name, open_flags, 0o600, dir_fd=opened_parent_fd) + except BaseException: + os.close(opened_parent_fd) + raise + else: + fd = os.open(path, open_flags, 0o600) lock_fd = fd sidecar_fd: int | None = None try: + _restrict_open_file(fd, path) if _fcntl is None and _msvcrt is not None: sidecar_path = path.with_name(f".{path.name}.lock") sidecar_fd = os.open(sidecar_path, os.O_RDWR | os.O_CREAT | binary_flag, 0o600) @@ -222,7 +247,7 @@ def append_history_line(path: Path, line: str) -> None: # for the subsequent blocking msvcrt lock; do not turn # that expected initialization race into a command error. pass - restrict_file(sidecar_path) + _restrict_open_file(sidecar_fd, sidecar_path) lock_fd = sidecar_fd lock_history_file(lock_fd) try: @@ -233,6 +258,16 @@ def append_history_line(path: Path, line: str) -> None: if sidecar_fd is not None: os.close(sidecar_fd) os.close(fd) + if parent_fd is not None: + os.close(parent_fd) + + +def _restrict_open_file(fd: int, path: Path) -> None: + fchmod = getattr(os, "fchmod", None) + if os.name != "nt" and fchmod is not None: + fchmod(fd, 0o600) + else: + restrict_file(path) def lock_history_file(fd: int) -> None: diff --git a/tests/test_history.py b/tests/test_history.py index 0289642..2645a9b 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -1,6 +1,8 @@ from __future__ import annotations import json +import os +import stat import tempfile import unittest from concurrent.futures import ThreadPoolExecutor @@ -22,6 +24,29 @@ def locking(self, _fd: int, mode: int, size: int) -> None: class HistoryAppendTests(unittest.TestCase): + def test_history_directories_and_file_are_private(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + path = Path(tmpdir) / "cache" / "app" / "history" / "runs.jsonl" + history.write_history_record(path, {"run": 1}) + + for directory in (path.parent, path.parent.parent, path.parent.parent.parent): + self.assertEqual(stat.S_IMODE(directory.stat().st_mode), 0o700) + self.assertEqual(stat.S_IMODE(path.stat().st_mode), 0o600) + + @unittest.skipUnless(os.name != "nt" and hasattr(os, "O_NOFOLLOW"), "requires POSIX no-follow support") + def test_append_refuses_symlink_destination(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + victim = root / "victim.jsonl" + victim.write_text("original\n", encoding="utf-8") + history_path = root / "history.jsonl" + history_path.symlink_to(victim) + + with self.assertRaises(OSError): + history.append_history_line(history_path, "attacker\n") + + self.assertEqual(victim.read_text(encoding="utf-8"), "original\n") + def test_current_shell_falls_back_to_comspec(self) -> None: with mock.patch.dict("os.environ", {"COMSPEC": r"C:\Windows\System32\cmd.exe"}, clear=True): self.assertEqual(history.current_shell(), r"C:\Windows\System32\cmd.exe") From bc9b1c9b4485b0030cc1636953b567e7fd58d352 Mon Sep 17 00:00:00 2001 From: Ramesh Padmanabhaiah <22363102+codeforester@users.noreply.github.com> Date: Tue, 18 Aug 2026 13:28:15 +0530 Subject: [PATCH 2/2] Handle concurrent history creation across platforms --- lib/python/base_cli/history.py | 15 ++++++++------- tests/test_history.py | 1 + 2 files changed, 9 insertions(+), 7 deletions(-) diff --git a/lib/python/base_cli/history.py b/lib/python/base_cli/history.py index 8de9cdf..058f5a3 100644 --- a/lib/python/base_cli/history.py +++ b/lib/python/base_cli/history.py @@ -215,17 +215,18 @@ def append_history_line(path: Path, line: str) -> None: binary_flag = getattr(os, "O_BINARY", 0) open_flags = os.O_WRONLY | os.O_CREAT | os.O_APPEND | binary_flag | getattr(os, "O_NOFOLLOW", 0) parent_fd: int | None = None - if ( - os.name != "nt" - and hasattr(os, "O_NOFOLLOW") - and hasattr(os, "O_DIRECTORY") - and os.open in os.supports_dir_fd - ): + if os.name != "nt" and hasattr(os, "O_NOFOLLOW") and hasattr(os, "O_DIRECTORY") and os.open in os.supports_dir_fd: opened_parent_fd = _open_parent_directory(path.parent) assert opened_parent_fd is not None parent_fd = opened_parent_fd try: - fd = os.open(path.name, open_flags, 0o600, dir_fd=opened_parent_fd) + try: + fd = os.open(path.name, open_flags, 0o600, dir_fd=opened_parent_fd) + except FileNotFoundError: + # macOS can report ENOENT for a concurrent first creation via + # a directory descriptor. Retry the same no-follow open by + # path; the final-component symlink guard remains in force. + fd = os.open(path, open_flags, 0o600) except BaseException: os.close(opened_parent_fd) raise diff --git a/tests/test_history.py b/tests/test_history.py index 2645a9b..caddc5d 100644 --- a/tests/test_history.py +++ b/tests/test_history.py @@ -24,6 +24,7 @@ def locking(self, _fd: int, mode: int, size: int) -> None: class HistoryAppendTests(unittest.TestCase): + @unittest.skipUnless(os.name != "nt", "POSIX directory mode bits are unavailable on Windows") def test_history_directories_and_file_are_private(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: path = Path(tmpdir) / "cache" / "app" / "history" / "runs.jsonl"