diff --git a/CHANGELOG.md b/CHANGELOG.md index ab36bd7..c0a2611 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -26,6 +26,14 @@ and versions are tracked in the repo-root `VERSION` file. ### Fixed +- Redact sensitive option values across every declared alias and Click value + form, redact sensitive positional arguments, and protect conventional secret + parameter names automatically before argv reaches logs or history writers. +- Claim invocation temp leaves exclusively and refuse recursive cleanup unless + ownership, strict run-root containment, the run-ID marker, and a symlink-free + path can all be proven; content erasure uses a retained directory handle and + intentionally leaves the empty directory skeleton instead of reopening a + pathname-removal race. - Finalize core-owned run metadata for successful, failed, aborted, interrupted, and unexpected command outcomes without letting secondary persistence failures replace the command result. diff --git a/README.md b/README.md index 5b85df4..040eefd 100644 --- a/README.md +++ b/README.md @@ -220,8 +220,8 @@ def main(ctx: base_cli.Context, project: str, workspace: str | None) -> None: ... ``` -Use `sensitive=True` for options whose values should not appear in invocation -logs: +Use `sensitive=True` for options or arguments whose values must not reach +invocation logs or history writers: ```python @base_cli.option("--token", sensitive=True, required=True) @@ -229,8 +229,23 @@ def main(ctx: base_cli.Context, token: str) -> None: ... ``` -Both `--token secret` and `--token=secret` are accepted and redacted in debug -logs. +All aliases declared for a sensitive option are protected, including short and +alternate long forms. Spaced values, equals forms, and attached short-option +values are redacted. Sensitive positional arguments are redacted according to +the Click command schema: + +```python +@app.command() +@base_cli.argument("credential", sensitive=True) +def login(ctx: base_cli.Context, credential: str) -> None: + ... +``` + +Parameters whose names contain `token`, `password`, `secret`, `api-key` +(`api_key`), or `authorization` are protected automatically. Use +`sensitive=True` for domain-specific secret names. Custom history writers +receive already-redacted argv, so raw secret-bearing argv never crosses the +framework's persistence boundary. Use `dry_run=True` when a nonstandard option should drive `ctx.dry_run` and the lifecycle's default durable-write suppression: @@ -501,9 +516,31 @@ owns a bundle and therefore do not create one. Neither do inherited runtimes, `log_to_file=False`, or dry-run invocations; an explicit log path can still receive diagnostics in the latter two modes. Context startup is transactional: if directory creation, logger setup, or retention fails, base-cli closes -partially installed handlers and removes new bundle-local temp/log artifacts -and empty directories. Pre-existing content, persistent component caches, and -parent-runtime data are preserved. +partially installed handlers and erases new bundle-local temp files through the +same retained handle used by normal teardown. It retains log files and empty +directory boundaries rather than attempting race-prone pathname removal. +Pre-existing content, persistent component caches, and parent-runtime data are +preserved. + +Recursive temp cleanup is fail-closed. Base-cli erases contents only when +its leaf was claimed exclusively for the invocation, its retained directory +handle and creation-time filesystem identity still match, and the path remains +a strict, run-ID-marked descendant of the selected run root with no symlinked +component. Filesystem roots, the run root itself, replaced directories, +traversal paths, mounted targets, external paths, and paths whose ownership or +mount identity cannot be proven are kept and reported as cleanup warnings. +Content erasure is descriptor-relative. Empty directory nodes—including the +leaf and its ancestors—are intentionally retained because portable POSIX APIs +cannot atomically remove an already-verified open directory; avoiding pathname +`rmdir` closes the final replacement race. Platforms without the required +handle operations retain files too and warn. `--keep-temp` preserves both the +directory tree and files. + +This boundary assumes the per-user runtime tree is not maliciously mutated by +another process running with the same account while ownership is acquired or +cleanup runs. Processes with the same filesystem authority can otherwise +rename or replace any user-owned runtime path; base-cli still verifies the +retained handle against the published path before erasing contents. On POSIX, base-cli enforces owner-only `0600`/`0700` modes. On Windows, the default user-local cache root relies on inherited user-profile ACLs; consumers @@ -525,7 +562,7 @@ def close_connection() -> None: ctx.on_cleanup(close_connection) ``` -Cleanup hooks run before temp directory removal. Hook failures are logged as +Cleanup hooks run before temp-content erasure. Hook failures are logged as warnings and do not prevent later hooks from running. ## Testing diff --git a/docs/cache-ownership-and-layout.md b/docs/cache-ownership-and-layout.md index 320a6be..8af1451 100644 --- a/docs/cache-ownership-and-layout.md +++ b/docs/cache-ownership-and-layout.md @@ -44,9 +44,32 @@ The ownership boundary intentionally excludes parser failures, help and version requests, inherited runtime bindings, `log_to_file=False`, and dry-run mode. Those invocations do not create or finalize a bundle. If context construction fails after creating artifacts, rollback closes partial logging handlers and -removes new bundle-local temp/log artifacts and empty directories. It does not -delete pre-existing content, persistent component caches, paths outside the -selected run root, or a parent runtime's metadata. +erases new bundle-local temp files through the retained ownership handle. It +retains log files and empty directory boundaries rather than reopening pathname +replacement races, and does not delete pre-existing content, persistent +component caches, paths outside the selected run root, or a parent runtime's +metadata. + +Temp cleanup uses the same fail-closed ownership proof during normal teardown +and startup rollback. The final leaf is claimed exclusively through a stable +parent handle, retained for the invocation, and checked against its captured +filesystem identity. Its path must remain a strict lexical and resolved +descendant of the selected run root, carry the invocation's run ID as its final +component, and contain no symlinked component. Cleanup refuses roots, the run +root itself, replaced directories, traversal paths, mounted targets, external +paths, pre-existing directories, missing Linux mount identities, and anything +it cannot inspect safely. + +Files and symlinks are erased relative to the retained directory handle and +cleanup refuses cross-device descendants. All empty directory nodes are +retained: portable POSIX APIs cannot atomically bind `rmdir` to an +already-verified open directory, so pathname removal would reopen a replacement +race at every depth. If the host lacks the required handle operations, files +are retained with a warning. `--keep-temp` also retains files. A refusal never +replaces the command's primary result. The ownership claim assumes the private +per-user runtime tree is not maliciously mutated by another process with the +same account while ownership is acquired or cleanup runs; it is not a +cryptographic proof against a hostile same-account process. Persistent component caches live under the owner's `cache/components/` path. On POSIX systems, runtime directories are owner-only (`0700`) and runtime files diff --git a/docs/platform-support.md b/docs/platform-support.md index b91beb4..4366e51 100644 --- a/docs/platform-support.md +++ b/docs/platform-support.md @@ -20,6 +20,15 @@ Base or `basectl` natively Windows-compatible; those consumers have their own Unix-tooling and shell boundaries. The package does not provide package-manager integration, shell startup management, or WSL/Windows path translation. +Recursive invocation-temp content erasure requires descriptor-relative, +no-follow directory operations. Linux, macOS, and WSL2 provide those +primitives; the empty leaf is retained on every platform because portable +POSIX has no identity-bound `rmdir`. Empty nested directories and ancestors are +retained for the same reason. Linux additionally requires readable mount IDs +and fails closed if they are unavailable. Native Windows currently uses the +secure fallback: it retains both directories and files and emits a cleanup +warning rather than perform race-prone pathname recursion. + The supported Python range is Python 3.10 through 3.14. Bug reports should include the operating system, distribution or WSL version when relevant, Python version, and whether paths live on the native filesystem or a mounted diff --git a/lib/python/base_cli/_cleanup.py b/lib/python/base_cli/_cleanup.py new file mode 100644 index 0000000..3492c17 --- /dev/null +++ b/lib/python/base_cli/_cleanup.py @@ -0,0 +1,341 @@ +from __future__ import annotations + +import errno +import os +import stat +import sys +from collections.abc import Callable +from dataclasses import dataclass +from pathlib import Path, PureWindowsPath + + +class UnsafeCleanupPathError(RuntimeError): + """Raised when a recursive cleanup target cannot be proven safe.""" + + +@dataclass(frozen=True) +class _ValidatedCleanup: + root: Path + relative: Path + root_identity: tuple[int, int] + target_identity: tuple[int, int] + + +def remove_owned_temp_directory( + temp_dir: Path, + run_root: Path | None, + run_id: str, + *, + expected_identity: tuple[int, int], + owned_descriptor: int | None, + before_remove: Callable[[], None] | None = None, +) -> None: + """Erase an invocation-owned temp tree through its retained directory handle. + + The empty leaf itself is intentionally retained: portable POSIX APIs cannot + atomically bind a directory removal to an already-verified inode. + """ + + cleanup = _validated_cleanup_paths(temp_dir, run_root, run_id, expected_identity) + if cleanup is None: + return + if before_remove is not None: + before_remove() + if not _supports_fd_relative_cleanup(): + raise UnsafeCleanupPathError( + "platform does not provide the directory-handle operations required for safe recursive cleanup" + ) + if owned_descriptor is None: + raise UnsafeCleanupPathError("the invocation-owned temp directory handle is unavailable") + _remove_with_directory_handles(cleanup, owned_descriptor) + + +def _validated_cleanup_paths( + temp_dir: Path, + run_root: Path | None, + run_id: str, + expected_identity: tuple[int, int], +) -> _ValidatedCleanup | None: + if run_root is None: + raise UnsafeCleanupPathError("run root is unavailable") + + target_path = Path(temp_dir) + root_path = Path(run_root) + if _contains_parent_reference(target_path) or _contains_parent_reference(root_path): + raise UnsafeCleanupPathError("path traversal is not allowed") + if _is_root_like(target_path) or _is_root_like(root_path): + raise UnsafeCleanupPathError("filesystem-root cleanup targets are not allowed") + if not _is_single_path_component(run_id): + raise UnsafeCleanupPathError("run ID is not a valid ownership marker") + if target_path.name != run_id: + raise UnsafeCleanupPathError("temp directory does not carry the run ID ownership marker") + + target = _lexical_absolute(target_path) + root = _lexical_absolute(root_path) + if _is_root_like(target) or _is_root_like(root): + raise UnsafeCleanupPathError("filesystem-root cleanup targets are not allowed") + + relative = _strict_relative_path(target, root, resolved=False) + if not _inspect_directory_chain(root, relative): + return None + + try: + resolved_root = root.resolve(strict=True) + resolved_target = target.resolve(strict=True) + except OSError as exc: + raise UnsafeCleanupPathError("cleanup path could not be resolved safely") from exc + resolved_relative = _strict_relative_path(resolved_target, resolved_root, resolved=True) + if resolved_relative.parts != relative.parts: + raise UnsafeCleanupPathError("resolved cleanup path does not match its lexical run-root path") + if _is_root_like(resolved_target) or _is_root_like(resolved_root): + raise UnsafeCleanupPathError("resolved filesystem-root cleanup targets are not allowed") + if _is_mount_target(target): + raise UnsafeCleanupPathError("mounted temp directories are not recursive cleanup targets") + + root_stat = os.stat(resolved_root, follow_symlinks=False) + target_stat = os.stat(resolved_target, follow_symlinks=False) + _require_identity( + target_stat, + expected_identity, + "temp directory no longer matches the invocation-owned directory", + ) + return _ValidatedCleanup( + root=resolved_root, + relative=relative, + root_identity=_identity(root_stat), + target_identity=expected_identity, + ) + + +def _supports_fd_relative_cleanup() -> bool: + return bool( + hasattr(os, "O_DIRECTORY") + and hasattr(os, "O_NOFOLLOW") + and os.open in os.supports_dir_fd + and os.stat in os.supports_dir_fd + and os.stat in os.supports_follow_symlinks + and os.unlink in os.supports_dir_fd + and os.scandir in os.supports_fd + ) + + +def _remove_with_directory_handles(cleanup: _ValidatedCleanup, owned_descriptor: int) -> None: + root_fd = _open_absolute_directory(cleanup.root) + descriptors = [root_fd] + try: + root_stat = os.fstat(root_fd) + _require_identity(root_stat, cleanup.root_identity, "run root changed during cleanup validation") + root_device = root_stat.st_dev + root_mount = _required_mount_identity(root_fd) + + for component in cleanup.relative.parts[:-1]: + child_fd = _open_child_directory(descriptors[-1], component) + try: + child_stat = os.fstat(child_fd) + if _crosses_mount(child_fd, child_stat, root_device, root_mount): + raise UnsafeCleanupPathError("cleanup path crosses a mounted filesystem") + except BaseException: + os.close(child_fd) + raise + descriptors.append(child_fd) + + target_name = cleanup.relative.parts[-1] + owned_stat = os.fstat(owned_descriptor) + _require_identity( + owned_stat, + cleanup.target_identity, + "retained temp directory handle no longer matches invocation ownership", + ) + _require_current_entry(descriptors[-1], target_name, owned_stat) + if _crosses_mount(owned_descriptor, owned_stat, root_device, root_mount): + raise UnsafeCleanupPathError("cleanup target crosses a mounted filesystem") + _remove_directory_contents(owned_descriptor, root_device, root_mount) + _require_identity( + os.fstat(owned_descriptor), + cleanup.target_identity, + "retained temp directory handle changed during cleanup", + ) + finally: + for descriptor in reversed(descriptors): + try: + os.close(descriptor) + except OSError: + pass + + +def _open_absolute_directory(path: Path) -> int: + anchor = Path(path.anchor) + descriptor = os.open(anchor, _directory_open_flags()) + try: + for component in path.relative_to(anchor).parts: + next_descriptor = _open_child_directory(descriptor, component) + os.close(descriptor) + descriptor = next_descriptor + return descriptor + except BaseException: + os.close(descriptor) + raise + + +def _open_child_directory(parent_fd: int, name: str) -> int: + try: + return os.open(name, _directory_open_flags(), dir_fd=parent_fd) + except OSError as exc: + if exc.errno in {errno.ELOOP, errno.ENOTDIR}: + raise UnsafeCleanupPathError("symlink cleanup targets are not allowed") from exc + raise + + +def _directory_open_flags() -> int: + return os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + + +def _remove_directory_contents( + directory_fd: int, + root_device: int, + root_mount: str | None, +) -> None: + with os.scandir(directory_fd) as entries: + names = [entry.name for entry in entries] + for name in names: + try: + child_fd = _open_child_directory(directory_fd, name) + except UnsafeCleanupPathError: + entry_stat = os.stat(name, dir_fd=directory_fd, follow_symlinks=False) + if stat.S_ISDIR(entry_stat.st_mode): + raise + os.unlink(name, dir_fd=directory_fd) + continue + + try: + child_stat = os.fstat(child_fd) + if _crosses_mount(child_fd, child_stat, root_device, root_mount): + raise UnsafeCleanupPathError("cleanup tree contains a mounted filesystem") + _remove_directory_contents(child_fd, root_device, root_mount) + finally: + os.close(child_fd) + + +def _require_current_entry(parent_fd: int, name: str, opened_stat: os.stat_result) -> None: + try: + current_stat = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + except OSError as exc: + raise UnsafeCleanupPathError("cleanup directory changed while it was open") from exc + if not stat.S_ISDIR(current_stat.st_mode) or _identity(current_stat) != _identity(opened_stat): + raise UnsafeCleanupPathError("cleanup directory changed while it was open") + + +def _identity(value: os.stat_result) -> tuple[int, int]: + return value.st_dev, value.st_ino + + +def _require_identity(value: os.stat_result, expected: tuple[int, int], message: str) -> None: + if _identity(value) != expected or not stat.S_ISDIR(value.st_mode): + raise UnsafeCleanupPathError(message) + + +def _crosses_mount( + descriptor: int, + value: os.stat_result, + root_device: int, + root_mount: str | None, +) -> bool: + if value.st_dev != root_device: + return True + mount = _mount_identity(descriptor) + if _requires_mount_identity() and (root_mount is None or mount is None): + raise UnsafeCleanupPathError("Linux mount identity is unavailable; refusing recursive cleanup") + return root_mount is not None and mount is not None and mount != root_mount + + +def _required_mount_identity(descriptor: int) -> str | None: + mount = _mount_identity(descriptor) + if _requires_mount_identity() and mount is None: + raise UnsafeCleanupPathError("Linux mount identity is unavailable; refusing recursive cleanup") + return mount + + +def _requires_mount_identity() -> bool: + return sys.platform.startswith("linux") + + +def _mount_identity(descriptor: int) -> str | None: + """Return Linux's stable mount ID for an open directory when available.""" + + try: + with open(f"/proc/self/fdinfo/{descriptor}", encoding="utf-8") as handle: + for line in handle: + key, separator, value = line.partition(":") + if separator and key == "mnt_id": + return value.strip() or None + except OSError: + return None + return None + + +def _contains_parent_reference(path: Path) -> bool: + return ".." in path.parts or ".." in PureWindowsPath(os.fspath(path)).parts + + +def _is_single_path_component(value: str) -> bool: + if not value or value in {".", ".."}: + return False + native_path = Path(value) + windows_path = PureWindowsPath(value) + return ( + not native_path.anchor + and not windows_path.anchor + and native_path.parts == (value,) + and windows_path.parts == (value,) + ) + + +def _lexical_absolute(path: Path) -> Path: + try: + return Path(os.path.abspath(os.fspath(path))) + except (OSError, TypeError, ValueError) as exc: + raise UnsafeCleanupPathError("cleanup path could not be normalized") from exc + + +def _is_root_like(path: Path) -> bool: + if path.anchor and path == Path(path.anchor): + return True + windows_path = PureWindowsPath(os.fspath(path)) + return bool(windows_path.anchor) and windows_path == PureWindowsPath(windows_path.anchor) + + +def _strict_relative_path(target: Path, root: Path, *, resolved: bool) -> Path: + try: + relative = target.relative_to(root) + except ValueError as exc: + qualifier = "resolved " if resolved else "" + raise UnsafeCleanupPathError(f"temp directory is outside the {qualifier}run root") from exc + if relative == Path("."): + qualifier = "resolved " if resolved else "" + raise UnsafeCleanupPathError(f"temp directory equals the {qualifier}run root") + return relative + + +def _inspect_directory_chain(root: Path, relative: Path) -> bool: + current = root + for index, component in enumerate(("", *relative.parts)): + if index: + current /= component + try: + mode = os.lstat(current).st_mode + except FileNotFoundError: + return False + except OSError as exc: + raise UnsafeCleanupPathError("cleanup path could not be inspected safely") from exc + if stat.S_ISLNK(mode): + raise UnsafeCleanupPathError("symlink cleanup targets are not allowed") + if not stat.S_ISDIR(mode): + raise UnsafeCleanupPathError("cleanup target path contains a non-directory") + return True + + +def _is_mount_target(path: Path) -> bool: + try: + return os.path.ismount(path) + except (OSError, ValueError) as exc: + raise UnsafeCleanupPathError("cleanup target mount status could not be inspected") from exc diff --git a/lib/python/base_cli/_runtime.py b/lib/python/base_cli/_runtime.py index 1c94b9f..3fe1537 100644 --- a/lib/python/base_cli/_runtime.py +++ b/lib/python/base_cli/_runtime.py @@ -3,10 +3,11 @@ import json import logging import os +import stat from dataclasses import dataclass from pathlib import Path -from ._private_files import restrict_directory, write_private_json +from ._private_files import PRIVATE_DIRECTORY_MODE, restrict_directory, write_private_json from .paths import runtime_run_directory_name, runtime_slug @@ -69,6 +70,133 @@ def create_runtime_directory(path: Path, cache_root: Path) -> None: raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc +def create_owned_runtime_directory( + path: Path, + cache_root: Path, +) -> tuple[tuple[int, int], int | None]: + """Exclusively create one invocation-owned leaf and retain its stable handle.""" + + create_runtime_directory(path.parent, cache_root) + if not _supports_secure_owned_directory_creation(): + return _create_owned_runtime_directory_portable(path, cache_root) + + try: + parent_fd = _open_absolute_directory(path.parent) + except OSError as exc: + raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc + + leaf_fd: int | None = None + try: + try: + os.mkdir(path.name, PRIVATE_DIRECTORY_MODE, dir_fd=parent_fd) + except FileExistsError as exc: + raise RuntimeDirectoryError(_owned_directory_collision_error(path)) from exc + + leaf_fd = os.open(path.name, _directory_open_flags(), dir_fd=parent_fd) + created_stat = os.fstat(leaf_fd) + _require_current_owned_entry(parent_fd, path.name, created_stat, path) + if _is_within(path, cache_root) and os.name != "nt": + os.fchmod(leaf_fd, PRIVATE_DIRECTORY_MODE) + _require_current_owned_entry(parent_fd, path.name, created_stat, path) + return (created_stat.st_dev, created_stat.st_ino), leaf_fd + except RuntimeDirectoryError: + if leaf_fd is not None: + os.close(leaf_fd) + raise + except OSError as exc: + if leaf_fd is not None: + os.close(leaf_fd) + raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc + except BaseException: + if leaf_fd is not None: + os.close(leaf_fd) + raise + finally: + os.close(parent_fd) + + +def _supports_secure_owned_directory_creation() -> bool: + return bool( + hasattr(os, "O_DIRECTORY") + and hasattr(os, "O_NOFOLLOW") + and os.mkdir in os.supports_dir_fd + and os.open in os.supports_dir_fd + and os.stat in os.supports_dir_fd + and os.stat in os.supports_follow_symlinks + and hasattr(os, "fchmod") + ) + + +def _create_owned_runtime_directory_portable( + path: Path, + cache_root: Path, +) -> tuple[tuple[int, int], int | None]: + """Use the strongest available binding where directory-relative APIs are absent.""" + + try: + path.mkdir(mode=PRIVATE_DIRECTORY_MODE) + except FileExistsError as exc: + raise RuntimeDirectoryError(_owned_directory_collision_error(path)) from exc + except OSError as exc: + raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc + + try: + created_stat = os.stat(path, follow_symlinks=False) + if not stat.S_ISDIR(created_stat.st_mode): + raise RuntimeDirectoryError( + f"Unable to claim invocation temp directory '{path}': it changed during creation." + ) + if _is_within(path, cache_root) and os.name != "nt": + restrict_directory(path) + return (created_stat.st_dev, created_stat.st_ino), None + except RuntimeDirectoryError: + raise + except OSError as exc: + raise RuntimeDirectoryError(_runtime_directory_error(path, cache_root, exc)) from exc + + +def _open_absolute_directory(path: Path) -> int: + absolute = path.resolve(strict=True) + anchor = Path(absolute.anchor) + descriptor = os.open(anchor, _directory_open_flags()) + try: + for component in absolute.relative_to(anchor).parts: + next_descriptor = os.open(component, _directory_open_flags(), dir_fd=descriptor) + os.close(descriptor) + descriptor = next_descriptor + return descriptor + except BaseException: + os.close(descriptor) + raise + + +def _directory_open_flags() -> int: + return os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW | getattr(os, "O_CLOEXEC", 0) + + +def _require_current_owned_entry( + parent_fd: int, + name: str, + created_stat: os.stat_result, + path: Path, +) -> None: + current_stat = os.stat(name, dir_fd=parent_fd, follow_symlinks=False) + if ( + not stat.S_ISDIR(current_stat.st_mode) + or (current_stat.st_dev, current_stat.st_ino) != (created_stat.st_dev, created_stat.st_ino) + ): + raise RuntimeDirectoryError( + f"Unable to claim invocation temp directory '{path}': it changed during creation." + ) + + +def _owned_directory_collision_error(path: Path) -> str: + return ( + f"Unable to claim invocation temp directory '{path}': it appeared concurrently. " + "Refusing to treat pre-existing content as framework-owned." + ) + + def runtime_namespace_root(cache_root: Path, namespace: str) -> Path: """Return an application-owned runtime namespace without product assumptions.""" return cache_root / runtime_slug(namespace, fallback="application") diff --git a/lib/python/base_cli/app.py b/lib/python/base_cli/app.py index dfda1df..cf90bec 100644 --- a/lib/python/base_cli/app.py +++ b/lib/python/base_cli/app.py @@ -3,7 +3,6 @@ import functools import logging import os -import shutil import sys import time import traceback @@ -11,6 +10,7 @@ from dataclasses import dataclass from datetime import datetime from pathlib import Path +from threading import Lock from typing import Any, Callable from ._lifecycle import ( @@ -21,7 +21,12 @@ system_exit_code, ) from ._private_files import write_private_json -from ._runtime import RuntimeDirectoryError, create_runtime_directory, prune_log_files +from ._runtime import ( + RuntimeDirectoryError, + create_owned_runtime_directory, + create_runtime_directory, + prune_log_files, +) from .context import Context, recover_current_context, reset_current_context, set_current_context from .errors import ConfigurationError from .exit_codes import ExitCode @@ -32,7 +37,7 @@ normalize_cli_name, ) from .profile import CliProfile -from .redaction import parameter_name_from_decls +from .redaction import RedactionPlan, compile_redaction_plan, parameter_name_from_decls, redact_argv _STANDARD_OPTION_KEYS = ("debug", "quiet", "environment", "config", "keep_temp", "log_file") _GROUP_STANDARD_OPTIONS_KEY = "base_cli_standard_options" @@ -198,6 +203,8 @@ def __init__( # explicit profile. self.profile = profile or CliProfile.generic() self._click_command = None + self._redaction_plan: RedactionPlan | None = None + self._click_command_lock = Lock() self._command_func: Callable[..., Any] | None = None self._command_args: tuple[Any, ...] = () self._command_kwargs: dict[str, Any] = {} @@ -239,9 +246,20 @@ def __call__(self, *args: Any, **kwargs: Any) -> Any: @property def click_command(self) -> Any: - if self._click_command is None: - self._click_command = self._build_click_command() - return self._click_command + command = self._click_command + if command is not None: + return command + + with self._click_command_lock: + command = self._click_command + if command is None: + command = self._build_click_command() + redaction_plan = compile_redaction_plan(command) + # Publish the command last so another thread can never invoke + # its wrapper before the corresponding plan is available. + self._redaction_plan = redaction_plan + self._click_command = command + return command def _build_click_command(self) -> Any: if self._command_func is None and not self._subcommands: @@ -268,7 +286,6 @@ def _build_command_wrapper( func: Callable[..., Any], include_version: bool, ) -> Callable[..., Any]: - sensitive_options = set(getattr(func, "__base_cli_sensitive_options__", set())) dry_run_parameter = getattr(func, "__base_cli_dry_run_parameter__", "dry_run") @functools.wraps(func) @@ -285,12 +302,14 @@ def wrapper(**kwargs: Any): recorder: RunRecorder | None = None outcome = outcome_from_exit_code(ExitCode.SUCCESS) invocation_argv: list[str] = [] + redaction_plan = self._redaction_plan + if redaction_plan is None: + raise RuntimeError("Command redaction plan was not initialized.") token = None try: try: context = self._create_context( standard, - sensitive_options, dry_run=bool(kwargs.get(dry_run_parameter)), ) except ConfigurationError as exc: @@ -301,9 +320,9 @@ def wrapper(**kwargs: Any): recorder = RunRecorder(context, started_at, started_monotonic_ns) token = set_current_context(context) _capture_invocation_context(context) - invocation_argv = _current_invocation_argv() + invocation_argv = redact_argv(_current_invocation_argv(), redaction_plan) _start_run_recorder(recorder) - log_invocation(context.log, invocation_argv, sensitive_options) + log_invocation(context.log, invocation_argv, None) if context.project_root is not None: context.log.debug("project_root=%s", context.project_root) if context.manifest_path is not None: @@ -335,7 +354,7 @@ def wrapper(**kwargs: Any): self.profile.history_writer( context, invocation_argv, - sensitive_options, + set(redaction_plan), started_at, outcome.exit_code, ) @@ -363,16 +382,21 @@ def wrapper(**kwargs: Any): if token is not None: _reset_active_context(context, token) - for kind, param_decls, attrs in getattr(func, "__base_cli_param_specs__", []): + for spec in getattr(func, "__base_cli_param_specs__", []): + kind, param_decls, attrs, *metadata = spec + sensitive = bool(metadata[0]) if metadata else False if kind == "option": wrapper = click.option(*param_decls, **attrs)(wrapper) elif kind == "argument": wrapper = click.argument(*param_decls, **attrs)(wrapper) + if sensitive: + click_parameters = getattr(wrapper, "__click_params__", ()) + if click_parameters: + click_parameters[-1]._base_cli_sensitive = True wrapper = _decorate_standard_options(click, wrapper, self.version if include_version else None) return wrapper - def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], dry_run: bool = False) -> Context: - del sensitive_options + def _create_context(self, standard: dict[str, Any], dry_run: bool = False) -> Context: project = self.profile.discover_project(current_working_dir()) manifest_path = project.manifest if project is not None else None explicit_config = Path(standard["config"]).expanduser() if standard.get("config") else None @@ -402,19 +426,7 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], owns_run_metadata = inherited_path is None and not dry_run and self.log_to_file run_metadata_path = layout.run_root / "run.json" if owns_run_metadata else None - run_root_was_new = not layout.run_root.exists() temp_dir_was_new = not layout.temp_dir.exists() - rollback_empty_directories = tuple( - directory - for directory in ( - layout.temp_dir.parent, - layout.temp_dir.parent.parent, - layout.log_dir, - layout.run_root, - ) - if not directory.exists() - ) - log_file_existed = log_file.exists() if log_file is not None else False logger = logging.getLogger(f"base_cli.{self.name}") context = Context( cli_name=self.name, @@ -454,8 +466,15 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], if log_file is not None: create_runtime_directory(log_file.parent, cache_root) else: - for directory in (layout.log_dir, layout.cache_dir, layout.temp_dir): + for directory in (layout.log_dir, layout.cache_dir): create_runtime_directory(directory, cache_root) + if temp_dir_was_new: + owned_identity, owned_descriptor = create_owned_runtime_directory(layout.temp_dir, cache_root) + context._owned_temp_descriptor = owned_descriptor + context._owned_temp_identity = owned_identity + context._owns_temp_dir = True + else: + create_runtime_directory(layout.temp_dir, cache_root) if log_file is not None: create_runtime_directory(log_file.parent, cache_root) @@ -488,22 +507,9 @@ def _create_context(self, standard: dict[str, Any], sensitive_options: set[str], pass return context except BaseException: - remove_owned_log = bool( - uses_default_log_file - and log_file is not None - and not log_file_existed - and run_root_was_new - and _path_is_within(log_file, layout.run_root) - ) _rollback_context_creation( context, logger_activation_started=logger_activation_started, - remove_owned_log=remove_owned_log, - remove_new_temp=( - temp_dir_was_new - and _path_is_within(layout.temp_dir, layout.run_root, strict=True) - ), - empty_directories=rollback_empty_directories, ) raise @@ -512,52 +518,24 @@ def _rollback_context_creation( context: Context, *, logger_activation_started: bool, - remove_owned_log: bool, - remove_new_temp: bool, - empty_directories: tuple[Path, ...], ) -> None: if logger_activation_started: keep_temp = context.keep_temp context.keep_temp = True try: try: - context.cleanup() + context._cleanup_preserving_temp_ownership() except BaseException: # pylint: disable=broad-exception-caught pass finally: context.keep_temp = keep_temp - if remove_new_temp: - _remove_new_temp_directory(context.temp_dir) - - if remove_owned_log and context.log_file is not None: - try: - context.log_file.unlink() - except BaseException: # pylint: disable=broad-exception-caught - pass - for directory in sorted(set(empty_directories), key=lambda path: len(path.parts), reverse=True): - try: - directory.rmdir() - except BaseException: # pylint: disable=broad-exception-caught - pass - - -def _remove_new_temp_directory(temp_dir: Path) -> None: try: - if temp_dir.exists(): - shutil.rmtree(temp_dir) + context._cleanup_owned_temp_dir() except BaseException: # pylint: disable=broad-exception-caught pass -def _path_is_within(path: Path, root: Path, *, strict: bool = False) -> bool: - try: - relative = path.resolve().relative_to(root.resolve()) - except BaseException: # pylint: disable=broad-exception-caught - return False - return not strict or relative != Path(".") - - def run_app(app: App, argv: list[str] | None = None, *, reraise_unexpected: bool = False) -> int: """Run an :class:`App` and return its normalized process exit code.""" @@ -694,12 +672,8 @@ def command(*args: Any, **kwargs: Any): def option(*param_decls: str, sensitive: bool = False, dry_run: bool = False, **attrs: Any): def decorator(func: Callable[..., Any]): specs = list(getattr(func, "__base_cli_param_specs__", [])) - specs.append(("option", param_decls, attrs)) + specs.append(("option", param_decls, attrs, sensitive)) func.__base_cli_param_specs__ = specs - if sensitive: - options = set(getattr(func, "__base_cli_sensitive_options__", set())) - options.add(parameter_name_from_decls(param_decls)) - func.__base_cli_sensitive_options__ = options if dry_run: dry_run_parameter = parameter_name_from_decls(param_decls) existing_dry_run_parameter = getattr(func, "__base_cli_dry_run_parameter__", None) @@ -714,10 +688,10 @@ def decorator(func: Callable[..., Any]): return decorator -def argument(*param_decls: str, **attrs: Any): +def argument(*param_decls: str, sensitive: bool = False, **attrs: Any): def decorator(func: Callable[..., Any]): specs = list(getattr(func, "__base_cli_param_specs__", [])) - specs.append(("argument", param_decls, attrs)) + specs.append(("argument", param_decls, attrs, sensitive)) func.__base_cli_param_specs__ = specs return func diff --git a/lib/python/base_cli/context.py b/lib/python/base_cli/context.py index 1efdd7f..3897713 100644 --- a/lib/python/base_cli/context.py +++ b/lib/python/base_cli/context.py @@ -2,11 +2,13 @@ import contextvars import logging -import shutil +import os from dataclasses import dataclass, field from pathlib import Path from typing import Any, Callable +from ._cleanup import remove_owned_temp_directory + _current_context: contextvars.ContextVar[Context | None] = contextvars.ContextVar( "base_cli_current_context", @@ -50,6 +52,9 @@ class Context: owner_root: Path | None = None run_root: Path | None = None _run_metadata_path: Path | None = field(default=None, init=False, repr=False, compare=False) + _owns_temp_dir: bool = field(default=False, init=False, repr=False, compare=False) + _owned_temp_identity: tuple[int, int] | None = field(default=None, init=False, repr=False, compare=False) + _owned_temp_descriptor: int | None = field(default=None, init=False, repr=False, compare=False) def on_cleanup(self, hook: Callable[[], None]) -> None: self.cleanup_hooks.append(hook) @@ -66,22 +71,61 @@ def _warn_cleanup_failure(self, message: str, *args: object) -> None: except BaseException: # pylint: disable=broad-exception-caught pass + def _cleanup_owned_temp_dir(self) -> None: + if not self._owns_temp_dir: + self._close_owned_temp_descriptor() + return + expected_identity = self._owned_temp_identity + owned_descriptor = self._owned_temp_descriptor + self._owns_temp_dir = False + self._owned_temp_identity = None + self._owned_temp_descriptor = None + try: + if expected_identity is None: + raise RuntimeError("temp directory ownership identity is unavailable") + remove_owned_temp_directory( + self.temp_dir, + self.run_root, + self.run_id, + expected_identity=expected_identity, + owned_descriptor=owned_descriptor, + ) + except BaseException as exc: # pylint: disable=broad-exception-caught + self._warn_cleanup_failure("Temp directory cleanup failed for '%s': %s", self.temp_dir, exc) + finally: + if owned_descriptor is not None: + try: + os.close(owned_descriptor) + except OSError as exc: + self._warn_cleanup_failure("Temp directory handle close failed: %s", exc) + + def _close_owned_temp_descriptor(self) -> None: + owned_descriptor = self._owned_temp_descriptor + self._owned_temp_descriptor = None + self._owned_temp_identity = None + self._owns_temp_dir = False + if owned_descriptor is not None: + try: + os.close(owned_descriptor) + except OSError as exc: + self._warn_cleanup_failure("Temp directory handle close failed: %s", exc) + def cleanup(self) -> None: + self._cleanup_resources(preserve_temp_ownership=False) + + def _cleanup_preserving_temp_ownership(self) -> None: + self._cleanup_resources(preserve_temp_ownership=True) + + def _cleanup_resources(self, *, preserve_temp_ownership: bool) -> None: for hook in self.cleanup_hooks: try: hook() except BaseException as exc: # pylint: disable=broad-exception-caught self._warn_cleanup_failure("Cleanup hook failed: %s", exc) - if not self.keep_temp and self.temp_dir.exists(): - try: - shutil.rmtree(self.temp_dir) - for parent in (self.temp_dir.parent, self.temp_dir.parent.parent): - try: - parent.rmdir() - except OSError: - break - except BaseException as exc: # pylint: disable=broad-exception-caught - self._warn_cleanup_failure("Temp directory cleanup failed for '%s': %s", self.temp_dir, exc) + if not self.keep_temp: + self._cleanup_owned_temp_dir() + elif not preserve_temp_ownership: + self._close_owned_temp_descriptor() for handler in list(self.log.handlers): try: handler.flush() diff --git a/lib/python/base_cli/history.py b/lib/python/base_cli/history.py index 86b2da6..bf146e0 100644 --- a/lib/python/base_cli/history.py +++ b/lib/python/base_cli/history.py @@ -62,13 +62,14 @@ def build_finished_record( exit_code: int, ) -> dict[str, Any]: ended_at = utc_now() + safe_argv = redact_history_argv(argv, sensitive_options) record: dict[str, Any] = { "schema_version": SCHEMA_VERSION, "run_id": context.run_id, "event": "finished", - "command": context.history_display_command(context.cli_name, argv), + "command": redact_history_text(context.history_display_command(context.cli_name, safe_argv)), "raw_command": context.cli_name, - "argv": redact_history_argv(argv, sensitive_options), + "argv": safe_argv, "started_at": format_timestamp(started_at), "ended_at": format_timestamp(ended_at), "duration_ms": duration_ms(started_at, ended_at), @@ -116,7 +117,7 @@ def write_primary_record( "schema_version": SCHEMA_VERSION, "run_id": run_id, "event": "finished", - "command": command, + "command": redact_history_text(command), "raw_command": raw_command, "argv": redact_history_argv(argv, sensitive_options=set()), "started_at": format_timestamp(started_at), @@ -293,31 +294,12 @@ def current_shell() -> str | None: def redact_history_argv(argv: list[str], sensitive_options: set[str]) -> list[str]: - redacted = redact_argv(argv, sensitive_options) - result: list[str] = [] - redact_next = False - for arg in redacted: - if redact_next: - result.append(REDACTED) - redact_next = False - continue - - option, separator, _value = arg.partition("=") - normalized = option_name_to_parameter(option) if option.startswith("--") else option - if option.startswith("--") and is_secret_key(normalized): - if separator: - result.append(f"{option}={REDACTED}") - else: - result.append(option) - redact_next = True - continue - result.append(redact_history_text(arg)) - return result + return [redact_history_text(arg) for arg in redact_argv(argv, sensitive_options)] def redact_history_text(value: str) -> str: key, separator, _value = value.partition("=") - if separator and is_secret_key(key): + if separator and is_secret_key(option_name_to_parameter(key)): return f"{key}={REDACTED}" return compact_home_text(redact_text_value(value)) diff --git a/lib/python/base_cli/logging.py b/lib/python/base_cli/logging.py index 52b7105..f0ccefa 100644 --- a/lib/python/base_cli/logging.py +++ b/lib/python/base_cli/logging.py @@ -171,8 +171,13 @@ def _active_application_home() -> Path | None: return context.application_home -def log_invocation(logger: logging.Logger, argv: list[str], sensitive_options: set[str]) -> None: - logger.debug("argv=%s", redact_argv(argv, sensitive_options)) +def log_invocation( + logger: logging.Logger, + argv: list[str], + sensitive_options: set[str] | None, +) -> None: + safe_argv = list(argv) if sensitive_options is None else redact_argv(argv, sensitive_options) + logger.debug("argv=%s", safe_argv) logger.debug("platform=%s %s", platform.system(), platform.machine()) logger.debug("python=%s", sys.version.replace("\n", " ")) diff --git a/lib/python/base_cli/redaction.py b/lib/python/base_cli/redaction.py index 5091968..660589b 100644 --- a/lib/python/base_cli/redaction.py +++ b/lib/python/base_cli/redaction.py @@ -1,45 +1,129 @@ from __future__ import annotations import re +from collections.abc import Callable, Iterable, Mapping, Sequence +from dataclasses import dataclass +from typing import Any REDACTED = "[REDACTED]" SECRET_KEY_RE = re.compile(r"(token|password|secret|api[-_]?key|authorization)", re.IGNORECASE) URL_CREDENTIALS_RE = re.compile(r"(?P[a-zA-Z][a-zA-Z0-9+.-]*://)[^/@\s]+@") +@dataclass(frozen=True) +class _OptionSpec: + name: str + aliases: tuple[str, ...] + sensitive: bool + takes_value: bool + nargs: int + + +@dataclass(frozen=True) +class _ArgumentSpec: + name: str + sensitive: bool + nargs: int + + +@dataclass(frozen=True) +class _CommandSpec: + options: tuple[_OptionSpec, ...] + option_aliases: Mapping[str, _OptionSpec] + arguments: tuple[_ArgumentSpec, ...] + subcommands: Mapping[str, "_CommandSpec"] + allow_interspersed_args: bool + ignore_unknown_options: bool + token_normalize_func: Callable[[str], str] | None + + +@dataclass(frozen=True) +class _OptionMatch: + option: _OptionSpec + attached_prefix: str | None = None + has_unknown: bool = False + + +class RedactionPlan(set[str]): + """A set-compatible collection with a private Click-aware scan plan. + + Consumer history writers have always received ``set[str]`` sensitive-name + collections. Subclassing ``set`` keeps that contract while allowing core + logging and history sinks to retain the option and argument schema needed + to redact values by their argv position. + """ + + def __init__(self, values: Iterable[str] = (), *, root: _CommandSpec | None = None) -> None: + super().__init__(values) + self._root = root + + def option_name_to_parameter(param_decl: str) -> str: - name = param_decl.lstrip("-") - return name.replace("-", "_") + _prefix, name = _split_option(param_decl) + return name.replace("-", "_").lower() + + +def option_aliases_from_decls(param_decls: Sequence[str]) -> tuple[str, ...]: + """Return raw Click option aliases, including both split flag forms.""" + + aliases: list[str] = [] + for decl in param_decls: + if decl.isidentifier(): + continue + split_char = ";" if decl.startswith("/") else "/" + first, separator, second = decl.partition(split_char) + for alias in (first.rstrip(), second.lstrip() if separator else ""): + if alias and _split_option(alias)[0]: + aliases.append(alias) + return tuple(aliases) def parameter_name_from_decls(param_decls: tuple[str, ...]) -> str: - options = [decl for decl in param_decls if decl.startswith("--")] - if options: - return option_name_to_parameter(options[0]) - return option_name_to_parameter(param_decls[0]) + """Resolve Click's destination name from raw option declarations.""" + for decl in param_decls: + if decl.isidentifier(): + return decl -def redact_argv(argv: list[str], sensitive_options: set[str]) -> list[str]: - redacted: list[str] = [] - skip_next = False - for arg in argv: - if skip_next: - redacted.append(REDACTED) - skip_next = False - continue + possible_names: list[tuple[str, str]] = [] + for decl in param_decls: + split_char = ";" if decl.startswith("/") else "/" + primary = decl.partition(split_char)[0].rstrip() + if _split_option(primary)[0]: + possible_names.append(_split_option(primary)) + if not possible_names: + return option_name_to_parameter(param_decls[0]) + possible_names.sort(key=lambda value: -len(value[0])) + return possible_names[0][1].replace("-", "_").lower() - option, separator, _value = arg.partition("=") - normalized = option_name_to_parameter(option) if option.startswith("--") else option - if option.startswith("--") and normalized in sensitive_options: - if separator: - redacted.append(f"{option}={REDACTED}") - else: - redacted.append(option) - skip_next = True - continue - redacted.append(arg) - return redacted +def compile_redaction_plan(command: Any, sensitive_names: Iterable[str] = ()) -> RedactionPlan: + """Compile a recursively Click-aware redaction plan for ``command``. + + Parameters may be marked with the private ``_base_cli_sensitive`` flag. + ``sensitive_names`` remains useful for adapters that already have a set of + destination names or raw aliases. Secret-looking parameter names are + protected automatically as defense in depth. + """ + + explicit = _sensitive_forms(sensitive_names) + compatible_names: set[str] = set(sensitive_names) + root = _compile_command( + command, + explicit, + compatible_names, + active=set(), + inherited_token_normalize_func=None, + ) + return RedactionPlan(compatible_names, root=root) + + +def redact_argv(argv: list[str], sensitive_options: set[str]) -> list[str]: + if isinstance(sensitive_options, RedactionPlan) and sensitive_options._root is not None: + redacted = _redact_with_plan(argv, sensitive_options._root) + else: + redacted = _redact_without_schema(argv, sensitive_options) + return [_redact_inline_text(value) for value in redacted] def is_secret_key(value: str) -> bool: @@ -48,3 +132,422 @@ def is_secret_key(value: str) -> bool: def redact_text_value(value: str) -> str: return URL_CREDENTIALS_RE.sub(lambda match: f"{match.group('prefix')}{REDACTED}@", value) + + +def _compile_command( + command: Any, + explicit: set[str], + compatible_names: set[str], + *, + active: set[int], + inherited_token_normalize_func: Callable[[str], str] | None, +) -> _CommandSpec: + identity = id(command) + if identity in active: + return _CommandSpec( + options=(), + option_aliases={}, + arguments=(), + subcommands={}, + allow_interspersed_args=True, + ignore_unknown_options=False, + token_normalize_func=inherited_token_normalize_func, + ) + active.add(identity) + try: + context_settings = dict(getattr(command, "context_settings", None) or {}) + token_normalize_func = context_settings.get("token_normalize_func") + if token_normalize_func is None: + token_normalize_func = inherited_token_normalize_func + options: list[_OptionSpec] = [] + arguments: list[_ArgumentSpec] = [] + for parameter in tuple(getattr(command, "params", ())): + name = str(getattr(parameter, "name", "") or "") + is_option = getattr(parameter, "param_type_name", None) == "option" + raw_aliases = ( + tuple( + str(alias) + for alias in ( + *tuple(getattr(parameter, "opts", ())), + *tuple(getattr(parameter, "secondary_opts", ())), + ) + if alias + ) + if is_option + else () + ) + sensitive = bool(getattr(parameter, "_base_cli_sensitive", False)) or _parameter_is_sensitive( + name, + raw_aliases, + explicit, + ) + nargs = _parameter_nargs(parameter) + if raw_aliases: + is_flag = bool(getattr(parameter, "is_flag", False) or getattr(parameter, "count", False)) + options.append( + _OptionSpec( + name=name, + aliases=raw_aliases, + sensitive=sensitive, + takes_value=not is_flag, + nargs=max(1, nargs), + ) + ) + if sensitive: + _add_compatible_names(compatible_names, name, raw_aliases) + else: + arguments.append(_ArgumentSpec(name=name, sensitive=sensitive, nargs=nargs)) + if sensitive and name: + compatible_names.add(name) + + subcommands: dict[str, _CommandSpec] = {} + raw_subcommands = getattr(command, "commands", None) + if isinstance(raw_subcommands, Mapping): + for command_name, child in raw_subcommands.items(): + if child is not None: + subcommands[str(command_name)] = _compile_command( + child, + explicit, + compatible_names, + active=active, + inherited_token_normalize_func=token_normalize_func, + ) + + allow_interspersed_args = context_settings.get("allow_interspersed_args") + if allow_interspersed_args is None: + allow_interspersed_args = getattr(command, "allow_interspersed_args", True) + ignore_unknown_options = context_settings.get("ignore_unknown_options") + if ignore_unknown_options is None: + ignore_unknown_options = getattr(command, "ignore_unknown_options", False) + return _CommandSpec( + options=tuple(options), + option_aliases=_build_option_alias_map(tuple(options), token_normalize_func), + arguments=tuple(arguments), + subcommands=subcommands, + allow_interspersed_args=bool(allow_interspersed_args), + ignore_unknown_options=bool(ignore_unknown_options), + token_normalize_func=token_normalize_func, + ) + finally: + active.remove(identity) + + +def _parameter_nargs(parameter: Any) -> int: + value = getattr(parameter, "nargs", 1) + return value if isinstance(value, int) else 1 + + +def _build_option_alias_map( + options: tuple[_OptionSpec, ...], + normalize: Callable[[str], str] | None, +) -> Mapping[str, _OptionSpec]: + alias_groups: dict[str, list[int]] = {} + for index, option in enumerate(options): + for alias in option.aliases: + normalized = _normalize_option_token(alias, normalize) + alias_groups.setdefault(normalized, []).append(index) + + effective_sensitive = [option.sensitive for option in options] + changed = True + while changed: + changed = False + for indices in alias_groups.values(): + if any(effective_sensitive[index] for index in indices): + for index in indices: + if not effective_sensitive[index]: + effective_sensitive[index] = True + changed = True + + aliases: dict[str, _OptionSpec] = {} + for index, option in enumerate(options): + effective = option + if effective_sensitive[index] and not option.sensitive: + effective = _OptionSpec( + name=option.name, + aliases=option.aliases, + sensitive=True, + takes_value=option.takes_value, + nargs=option.nargs, + ) + for alias in option.aliases: + aliases[_normalize_option_token(alias, normalize)] = effective + return aliases + + +def _parameter_is_sensitive(name: str, aliases: tuple[str, ...], explicit: set[str]) -> bool: + identifiers = {name, option_name_to_parameter(name)} if name else set() + for alias in aliases: + identifiers.add(alias) + identifiers.add(option_name_to_parameter(alias)) + return bool(identifiers & explicit) or any(is_secret_key(identifier) for identifier in identifiers) + + +def _add_compatible_names(values: set[str], name: str, aliases: tuple[str, ...]) -> None: + if name: + values.add(name) + for alias in aliases: + values.add(alias) + values.add(option_name_to_parameter(alias)) + + +def _sensitive_forms(values: Iterable[str]) -> set[str]: + result: set[str] = set() + for value in values: + result.add(value) + result.add(option_name_to_parameter(value)) + return result + + +def _redact_with_plan(argv: list[str], root: _CommandSpec) -> list[str]: + result = list(argv) + if not argv: + return result + + # Invocation argv always includes the display/program token at index zero. + # It is metadata, not part of the Click grammar. + _redact_command(argv, list(range(1, len(argv))), root, result) + + return result + + +def _redact_command( + argv: list[str], + token_indices: list[int], + command: _CommandSpec, + result: list[str], +) -> None: + positional_indices = _scan_options(argv, token_indices, command, result) + argument_spans, extra_indices = _allocate_argument_spans(positional_indices, command.arguments) + for argument, span in zip(command.arguments, argument_spans): + if argument.sensitive: + for index in span: + result[index] = REDACTED + + if not command.subcommands or not extra_indices: + return + command_index = extra_indices[0] + command_name = argv[command_index] + child = command.subcommands.get(command_name) + if child is None and command.token_normalize_func is not None: + child = command.subcommands.get(command.token_normalize_func(command_name)) + if child is not None: + _redact_command(argv, extra_indices[1:], child, result) + + +def _scan_options( + argv: list[str], + token_indices: list[int], + command: _CommandSpec, + result: list[str], +) -> list[int]: + positional: list[int] = [] + offset = 0 + while offset < len(token_indices): + index = token_indices[offset] + value = argv[index] + if value == "--": + positional.extend(token_indices[offset + 1 :]) + break + + match = _match_option(value, command) + if match is not None: + values_needed = match.option.nargs if match.option.takes_value else 0 + if match.attached_prefix is not None: + values_needed -= 1 + if match.option.sensitive: + result[index] = f"{match.attached_prefix}{REDACTED}" + + if match.has_unknown: + positional.append(index) + + offset += 1 + while values_needed > 0 and offset < len(token_indices): + value_index = token_indices[offset] + if match.option.sensitive: + result[value_index] = REDACTED + offset += 1 + values_needed -= 1 + continue + + if _looks_like_option(value, command): + if command.ignore_unknown_options: + positional.append(index) + offset += 1 + continue + + if not command.allow_interspersed_args: + positional.extend(token_indices[offset:]) + break + positional.append(index) + offset += 1 + return positional + + +def _allocate_argument_spans( + positional_indices: list[int], + arguments: tuple[_ArgumentSpec, ...], +) -> tuple[list[list[int]], list[int]]: + spans: list[list[int]] = [[] for _ in arguments] + wildcard = next((index for index, argument in enumerate(arguments) if argument.nargs < 0), None) + left = 0 + right = len(positional_indices) + + prefix_end = wildcard if wildcard is not None else len(arguments) + for argument_index in range(prefix_end): + count = max(1, arguments[argument_index].nargs) + end = min(left + count, right) + spans[argument_index] = positional_indices[left:end] + left = end + + if wildcard is not None: + for argument_index in range(len(arguments) - 1, wildcard, -1): + count = max(1, arguments[argument_index].nargs) + start = max(left, right - count) + spans[argument_index] = positional_indices[start:right] + right = start + spans[wildcard] = positional_indices[left:right] + return spans, [] + + return spans, positional_indices[left:] + + +def _match_option(value: str, command: _CommandSpec) -> _OptionMatch | None: + aliases = command.option_aliases + option_name, separator, _attached = value.partition("=") + exact = aliases.get(_normalize_option_token(option_name, command.token_normalize_func)) + if separator: + if exact is not None and exact.takes_value: + return _OptionMatch(exact, f"{option_name}=") + if exact is not None: + return None + elif exact is not None: + return _OptionMatch(exact) + + prefix = value[:1] + short_prefixes = { + alias[:1] + for option in command.options + for alias in option.aliases + if len(alias) == 2 and len(_split_option(alias)[0]) == 1 + } + if prefix not in short_prefixes or value[1:2] == prefix or len(value) <= 2: + return None + last_option: _OptionSpec | None = None + has_unknown = False + for position, character in enumerate(value[1:]): + short_option = aliases.get( + _normalize_option_token(f"{prefix}{character}", command.token_normalize_func) + ) + if short_option is None: + if command.ignore_unknown_options: + has_unknown = True + continue + return None + last_option = short_option + if short_option.takes_value: + attached = value[position + 2 :] + prefix = value[: position + 2] if attached else None + if prefix is not None and attached.startswith("="): + prefix += "=" + return _OptionMatch(short_option, prefix, has_unknown) + return _OptionMatch(last_option, has_unknown=has_unknown) if last_option is not None else None + + +def _normalize_option_token(value: str, normalize: Callable[[str], str] | None) -> str: + if normalize is None: + return value + prefix, name = _split_option(value) + if not prefix: + return value + return f"{prefix}{normalize(name)}" + + +def _looks_like_option(value: str, command: _CommandSpec) -> bool: + if value == "--" or _match_option(value, command) is not None: + return True + prefix, _name = _split_option(value) + prefixes: set[str] = set() + for option in command.options: + for alias in option.aliases: + alias_prefix, _alias_name = _split_option(alias) + if alias_prefix: + prefixes.update((alias_prefix, alias_prefix[:1])) + return bool(prefix) and prefix in prefixes + + +def _redact_without_schema(argv: list[str], sensitive_options: set[str]) -> list[str]: + result = list(argv) + sensitive = _sensitive_forms(sensitive_options) + short_aliases = _legacy_short_aliases(sensitive_options) + option_parsing = True + index = 0 + while index < len(argv): + value = argv[index] + if option_parsing and value == "--": + option_parsing = False + index += 1 + continue + if not option_parsing: + index += 1 + continue + + option_name, separator, _attached = value.partition("=") + normalized = option_name_to_parameter(option_name) + explicitly_sensitive = option_name in sensitive or normalized in sensitive + automatically_sensitive = _is_option_alias(option_name) and is_secret_key(normalized) + if explicitly_sensitive: + if separator: + result[index] = f"{option_name}={REDACTED}" + elif index + 1 < len(argv): + result[index + 1] = REDACTED + index += 1 + index += 1 + continue + + attached_alias = next( + (alias for alias in short_aliases if value.startswith(alias) and len(value) > len(alias)), + None, + ) + if attached_alias is not None: + suffix = value[len(attached_alias) :] + equals = "=" if suffix.startswith("=") else "" + result[index] = f"{attached_alias}{equals}{REDACTED}" + elif automatically_sensitive: + if separator: + result[index] = f"{option_name}={REDACTED}" + elif index + 1 < len(argv): + result[index + 1] = REDACTED + index += 1 + index += 1 + return result + + +def _legacy_short_aliases(sensitive_options: Iterable[str]) -> tuple[str, ...]: + aliases: set[str] = set() + for value in sensitive_options: + prefix, name = _split_option(value) + if len(prefix) == 1 and len(name) == 1: + aliases.add(value) + elif len(value) == 1 and value.isidentifier(): + aliases.add(f"-{value}") + return tuple(sorted(aliases, key=len, reverse=True)) + + +def _redact_inline_text(value: str) -> str: + key, separator, _raw_value = value.partition("=") + if separator and is_secret_key(option_name_to_parameter(key)): + value = f"{key}={REDACTED}" + return redact_text_value(value) + + +def _is_option_alias(value: str) -> bool: + return bool(_split_option(value)[0]) + + +def _split_option(value: str) -> tuple[str, str]: + first = value[:1] + if not first or first.isalnum() or first == "_": + return "", value + if value[1:2] == first: + return value[:2], value[2:] + return first, value[1:] diff --git a/tests/test_app_lifecycle.py b/tests/test_app_lifecycle.py index aa9df99..2d9d9ff 100644 --- a/tests/test_app_lifecycle.py +++ b/tests/test_app_lifecycle.py @@ -62,7 +62,8 @@ def record_cleanup_context() -> None: self.assertEqual(result.exit_code, 0, result.output) self.assertIsNone(result.exception) self.assertIs(seen["cleanup_context"], seen["context"]) - self.assertFalse(Path(seen["temp_dir"]).exists()) + self.assertTrue(Path(seen["temp_dir"]).is_dir()) + self.assertEqual(list(Path(seen["temp_dir"]).iterdir()), []) self.assertEqual(seen["logger"].handlers, []) with self.assertRaisesRegex(RuntimeError, "context is not active"): @@ -125,7 +126,8 @@ def main(ctx: base_cli.Context) -> None: self.assertIsInstance(result.exception, SystemExit) self.assertIsNot(result.exception, primary_failure) self.assertTrue(seen["cleanup_called"]) - self.assertFalse(Path(seen["temp_dir"]).exists()) + self.assertTrue(Path(seen["temp_dir"]).is_dir()) + self.assertEqual(list(Path(seen["temp_dir"]).iterdir()), []) self.assertEqual(seen["logger"].handlers, []) with self.assertRaisesRegex(RuntimeError, "context is not active"): @@ -164,8 +166,8 @@ def main(ctx: base_cli.Context) -> None: with tempfile.TemporaryDirectory() as tmpdir: with mock.patch.object( - context_module.shutil, - "rmtree", + context_module, + "remove_owned_temp_directory", side_effect=RuntimeError("cleanup implementation failed"), ): result = invoke(app, [], home=Path(tmpdir)) diff --git a/tests/test_app_security_boundaries.py b/tests/test_app_security_boundaries.py new file mode 100644 index 0000000..f296aea --- /dev/null +++ b/tests/test_app_security_boundaries.py @@ -0,0 +1,458 @@ +from __future__ import annotations + +import importlib.util +import os +import tempfile +import unittest +from dataclasses import replace +from datetime import datetime +from pathlib import Path +from threading import Event, Thread +from unittest import mock + +import base_cli +import base_cli.app as app_module +import base_cli._runtime as runtime_module +from base_cli._runtime import RuntimeDirectoryError, runtime_layout +from base_cli.redaction import REDACTED +from base_cli.testing import invoke + + +class RuntimeOwnershipBoundaryTests(unittest.TestCase): + def test_created_leaf_swap_is_detected_and_retained_handle_is_closed(self) -> None: + if not runtime_module._supports_secure_owned_directory_creation(): + self.skipTest("secure directory-relative creation is unavailable") + + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + cache_root = root / "cache" + temp_dir = cache_root / "run" / "tmp" / "cli" / "run-123" + replacement = root / "foreign" + replacement.mkdir() + marker = replacement / "preserve.txt" + marker.write_text("preserve", encoding="utf-8") + parked_owned = root / "parked-owned" + opened_leaf_descriptors: list[int] = [] + original_fstat = runtime_module.os.fstat + original_require = runtime_module._require_current_owned_entry + + def capture_fstat(descriptor: int) -> os.stat_result: + opened_leaf_descriptors.append(descriptor) + return original_fstat(descriptor) + + swapped = False + + def replace_before_binding( + parent_fd: int, + name: str, + created_stat: os.stat_result, + path: Path, + ) -> None: + nonlocal swapped + if not swapped: + swapped = True + temp_dir.rename(parked_owned) + replacement.rename(temp_dir) + original_require(parent_fd, name, created_stat, path) + + with mock.patch.object(runtime_module.os, "fstat", side_effect=capture_fstat), mock.patch.object( + runtime_module, + "_require_current_owned_entry", + side_effect=replace_before_binding, + ): + with self.assertRaisesRegex(RuntimeDirectoryError, "changed during creation"): + runtime_module.create_owned_runtime_directory(temp_dir, cache_root) + + self.assertEqual((temp_dir / marker.name).read_text(encoding="utf-8"), "preserve") + self.assertTrue(parked_owned.is_dir()) + self.assertEqual(len(opened_leaf_descriptors), 1) + with self.assertRaises(OSError): + os.fstat(opened_leaf_descriptors[0]) + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class AppRedactionBoundaryTests(unittest.TestCase): + def test_redaction_plan_is_published_before_concurrent_invocation(self) -> None: + compile_started = Event() + release_compile = Event() + invocation_started = Event() + captured: list[list[str]] = [] + received: list[str] = [] + failures: list[BaseException] = [] + statuses: list[int] = [] + secret = "fjord-ember-9274" + + def history_writer( + _ctx: base_cli.Context, + argv: list[str], + _sensitive: set[str], + _started_at: datetime, + _exit_code: int, + ) -> None: + captured.append(list(argv)) + + with tempfile.TemporaryDirectory() as tmpdir: + profile = replace( + base_cli.CliProfile.generic(cache_root=Path(tmpdir)), + history_writer=history_writer, + ) + app = base_cli.App(name="publication-race", profile=profile) + + @app.command() + @base_cli.option("--credential", sensitive=True, required=True) + def main(ctx: base_cli.Context, credential: str) -> None: + del ctx + received.append(credential) + + original_compile = app_module.compile_redaction_plan + + def blocking_compile(command: object): + compile_started.set() + if not release_compile.wait(5): + raise AssertionError("redaction-plan compilation was not released") + return original_compile(command) + + def build_command() -> None: + try: + _ = app.click_command + except BaseException as exc: # pylint: disable=broad-exception-caught + failures.append(exc) + + def invoke_command() -> None: + invocation_started.set() + try: + statuses.append( + app_module.run_app( + app, + ["--credential", secret], + reraise_unexpected=True, + ) + ) + except BaseException as exc: # pylint: disable=broad-exception-caught + failures.append(exc) + + builder = Thread(target=build_command) + invoker = Thread(target=invoke_command) + with mock.patch.object( + app_module, + "compile_redaction_plan", + side_effect=blocking_compile, + ): + builder.start() + try: + self.assertTrue(compile_started.wait(5)) + self.assertIsNone(app._click_command) + self.assertIsNone(app._redaction_plan) + invoker.start() + self.assertTrue(invocation_started.wait(5)) + finally: + release_compile.set() + builder.join(5) + if invoker.ident is not None: + invoker.join(5) + + self.assertFalse(builder.is_alive()) + self.assertFalse(invoker.is_alive()) + self.assertEqual(failures, []) + self.assertEqual(statuses, [0]) + self.assertEqual(received, [secret]) + self.assertEqual( + captured, + [["publication-race", "--credential", REDACTED]], + ) + self.assertNotIn(secret, repr(captured)) + + def test_actual_click_schema_redacts_every_value_before_history_callback(self) -> None: + captured: list[tuple[list[str], set[str], Path]] = [] + + def history_writer( + ctx: base_cli.Context, + argv: list[str], + sensitive: set[str], + _started_at: datetime, + _exit_code: int, + ) -> None: + assert ctx.log_file is not None + captured.append((list(argv), set(sensitive), ctx.log_file)) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="redaction-boundary", profile=profile) + received: list[tuple[str, str, bool]] = [] + + @app.command() + @base_cli.option("-t", "--token", "--auth-token", "credential", sensitive=True, required=True) + @base_cli.option("-v", "--verbose", is_flag=True) + @base_cli.argument("payload", sensitive=True) + def main(ctx: base_cli.Context, credential: str, verbose: bool, payload: str) -> None: + del ctx + received.append((credential, payload, verbose)) + + cases = ( + (["--token", "long-secret", "payload-one"], ("long-secret", "payload-one", False)), + (["--auth-token=equals-secret", "payload-two"], ("equals-secret", "payload-two", False)), + (["-vtattached-secret", "payload-three"], ("attached-secret", "payload-three", True)), + ) + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + for index, (args, expected) in enumerate(cases): + with self.subTest(args=args): + home = root / str(index) + home.mkdir() + result = invoke(app, args, home=home) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(received[-1], expected) + + self.assertEqual(len(captured), len(cases)) + all_secrets = {value for _, expected in cases for value in expected[:2]} + for safe_argv, sensitive, log_file in captured: + rendered = repr(safe_argv) + log_text = log_file.read_text(encoding="utf-8") + self.assertIn(REDACTED, rendered) + self.assertIn(REDACTED, log_text) + for secret in all_secrets: + self.assertNotIn(secret, rendered) + self.assertNotIn(secret, log_text) + self.assertTrue( + {"credential", "-t", "--token", "--auth-token", "payload"}.issubset(sensitive) + ) + + def test_secret_name_heuristics_cover_options_and_arguments(self) -> None: + captured: list[list[str]] = [] + + def history_writer( + _ctx: base_cli.Context, + argv: list[str], + _sensitive: set[str], + _started_at: datetime, + _exit_code: int, + ) -> None: + captured.append(list(argv)) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="automatic-redaction", profile=profile) + + @app.command() + @base_cli.option("--api-key", required=True) + @base_cli.argument("password") + def main(ctx: base_cli.Context, api_key: str, password: str) -> None: + del ctx, api_key, password + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke(app, ["--api-key", "option-secret", "argument-secret"], home=Path(tmpdir)) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual( + captured, + [["automatic-redaction", "--api-key", REDACTED, REDACTED]], + ) + + def test_click_token_normalization_cannot_bypass_redaction(self) -> None: + captured: list[list[str]] = [] + normalized_inputs: list[str] = [] + + def normalize(value: str) -> str: + normalized_inputs.append(value) + if "=" in value: + raise AssertionError("normalizer received attached value bytes") + return value.lower() + + def history_writer( + _ctx: base_cli.Context, + argv: list[str], + _sensitive: set[str], + _started_at: datetime, + _exit_code: int, + ) -> None: + captured.append(list(argv)) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="normalized-redaction", profile=profile) + + @app.command(context_settings={"token_normalize_func": normalize}) + @base_cli.option("--token", sensitive=True, required=True) + def main(ctx: base_cli.Context, token: str) -> None: + del ctx, token + + with tempfile.TemporaryDirectory() as tmpdir: + result = invoke(app, ["--TOKEN=normalized-secret"], home=Path(tmpdir)) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(captured, [["normalized-redaction", f"--TOKEN={REDACTED}"]]) + self.assertTrue(normalized_inputs) + self.assertFalse(any("normalized-secret" in value for value in normalized_inputs)) + + def test_short_prefixes_and_mixed_unknown_clusters_are_redacted(self) -> None: + captured: list[list[str]] = [] + + def history_writer( + _ctx: base_cli.Context, + argv: list[str], + _sensitive: set[str], + _started_at: datetime, + _exit_code: int, + ) -> None: + captured.append(list(argv)) + + profile = replace(base_cli.CliProfile.generic(), history_writer=history_writer) + app = base_cli.App(name="short-redaction", profile=profile) + received: list[str] = [] + + @app.command(context_settings={"ignore_unknown_options": True, "allow_extra_args": True}) + @base_cli.option("-t", "+t", "/t", "credential", sensitive=True, required=True) + @base_cli.option("-v", is_flag=True) + def main(ctx: base_cli.Context, credential: str, v: bool) -> None: + del ctx, v + received.append(credential) + + cases = ( + (["-xtcluster-one"], "cluster-one"), + (["-xvtcluster-two"], "cluster-two"), + (["-xt", "cluster-three"], "cluster-three"), + (["-xvt=cluster-four"], "=cluster-four"), + (["+tplus-secret"], "plus-secret"), + (["+vt=plus-equals"], "=plus-equals"), + (["/tslash-secret"], "slash-secret"), + ) + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + for index, (args, expected) in enumerate(cases): + home = root / str(index) + home.mkdir() + result = invoke(app, args, home=home) + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(received[-1], expected) + + self.assertEqual(len(captured), len(cases)) + for safe_argv in captured: + self.assertIn(REDACTED, repr(safe_argv)) + for _, secret in cases: + self.assertNotIn(secret, repr(safe_argv)) + + +@unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") +class AppCleanupBoundaryTests(unittest.TestCase): + def test_concurrent_temp_leaf_creation_is_never_claimed_or_deleted(self) -> None: + app = base_cli.App(name="cleanup-claim-race") + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run after an ownership race") + + original_create = app_module.create_owned_runtime_directory + raced_marker: list[Path] = [] + + def race_create(path: Path, cache_root: Path) -> None: + path.mkdir(parents=True) + marker = path / "foreign.txt" + marker.write_text("preserve", encoding="utf-8") + raced_marker.append(marker) + original_create(path, cache_root) + + with tempfile.TemporaryDirectory() as tmpdir, mock.patch.object( + app_module, + "create_owned_runtime_directory", + side_effect=race_create, + ): + result = invoke(app, [], home=Path(tmpdir)) + + self.assertEqual(result.exit_code, 1, result.output) + self.assertEqual(len(raced_marker), 1) + self.assertEqual(raced_marker[0].read_text(encoding="utf-8"), "preserve") + self.assertIn("appeared concurrently", result.stderr) + + def test_successful_invocation_refuses_external_profile_temp_tree(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + run_id = "fixed-run" + external_temp = root / "external" / run_id + layout = replace( + runtime_layout(cache_root, "cleanup-boundary", run_id), + temp_dir=external_temp, + ) + + def resolve_runtime( + _cli_name: str, + _project: base_cli.ProjectInfo | None, + ) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="cleanup-boundary", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id=run_id, + ) + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="cleanup-boundary", profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + (ctx.temp_dir / "keep.txt").write_text("preserve", encoding="utf-8") + + result = invoke(app, [], home=home) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual((external_temp / "keep.txt").read_text(encoding="utf-8"), "preserve") + self.assertIn("outside the run root", result.stderr) + + def test_successful_inherited_invocation_preserves_parent_bundle(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + parent_run = cache_root / "parent" / "runs" / "parent-run" + parent_run.mkdir(parents=True) + parent_marker = parent_run / "run.json" + parent_marker.write_text('{"status":"running"}', encoding="utf-8") + run_id = "child-run" + layout = runtime_layout( + cache_root, + "cleanup-child", + run_id, + namespace="parent", + inherited_run_root=parent_run, + ) + + def resolve_runtime( + _cli_name: str, + _project: base_cli.ProjectInfo | None, + ) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="parent", + project_root=None, + project_name=None, + inherited_path=parent_run, + history_parent_run_id="parent-run", + run_id=run_id, + ) + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="cleanup-child", profile=profile) + seen_temp: list[Path] = [] + + @app.command() + def main(ctx: base_cli.Context) -> None: + seen_temp.append(ctx.temp_dir) + (ctx.temp_dir / "temporary.txt").write_text("temporary", encoding="utf-8") + + result = invoke(app, [], home=home) + + self.assertEqual(result.exit_code, 0, result.output) + self.assertEqual(parent_marker.read_text(encoding="utf-8"), '{"status":"running"}') + self.assertTrue(parent_run.is_dir()) + self.assertTrue(seen_temp[0].is_dir()) + self.assertEqual(list(seen_temp[0].iterdir()), []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_app_startup_transaction.py b/tests/test_app_startup_transaction.py index b22ae74..2b96cc6 100644 --- a/tests/test_app_startup_transaction.py +++ b/tests/test_app_startup_transaction.py @@ -28,7 +28,7 @@ def _run(app: base_cli.App, home: Path) -> int: @unittest.skipUnless(importlib.util.find_spec("click"), "Click is not installed") class AppStartupTransactionTests(unittest.TestCase): - def test_retention_failure_rolls_back_new_bundle_and_logger(self) -> None: + def test_retention_failure_retains_partial_log_and_closes_logger(self) -> None: app = base_cli.App(name="startup-retention", max_log_files=1) called: list[None] = [] @@ -49,13 +49,15 @@ def main(ctx: base_cli.Context) -> None: self.assertEqual(status, 1) self.assertEqual(called, []) self.assertEqual(list((home / "cache").glob("**/run.json")), []) - self.assertEqual(list((home / "cache").glob("**/primary.log")), []) + partial_logs = list((home / "cache").glob("**/primary.log")) + self.assertEqual(len(partial_logs), 1) + self.assertTrue(partial_logs[0].is_file()) self.assertEqual(logging.getLogger("base_cli.startup-retention").handlers, []) self.assertEqual(preserved.read_text(encoding="utf-8"), "keep") with self.assertRaisesRegex(RuntimeError, "context is not active"): base_cli.get_current_context() - def test_partial_logger_failure_closes_handlers_and_removes_new_bundle(self) -> None: + def test_partial_logger_failure_closes_handlers_and_retains_partial_log(self) -> None: app = base_cli.App(name="startup-logger") original_configure_logger = app_module.configure_logger @@ -75,11 +77,105 @@ def fail_after_logger_setup(*args: object, **kwargs: object) -> logging.Logger: self.assertEqual(status, 1) self.assertEqual(list((home / "cache").glob("**/run.json")), []) - self.assertEqual(list((home / "cache").glob("**/primary.log")), []) + partial_logs = list((home / "cache").glob("**/primary.log")) + self.assertEqual(len(partial_logs), 1) + self.assertTrue(partial_logs[0].is_file()) self.assertEqual(logging.getLogger("base_cli.startup-logger").handlers, []) with self.assertRaisesRegex(RuntimeError, "context is not active"): base_cli.get_current_context() + def test_startup_rollback_never_unlinks_log_through_a_swapped_symlink_ancestor(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + run_id = "fixed-run" + layout = runtime_layout(cache_root, "startup-log-swap", run_id) + external_log_dir = root / "external-logs" + external_log_dir.mkdir() + victim = external_log_dir / "primary.log" + victim.write_text("preserve", encoding="utf-8") + parked_log_dir = layout.run_root / "logs-parked" + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="startup-log-swap", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id=run_id, + ) + + def fail_after_swapping_log_ancestor(*_args: object) -> None: + layout.log_dir.rename(parked_log_dir) + layout.log_dir.symlink_to(external_log_dir, target_is_directory=True) + raise RuntimeError("retention unavailable") + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="startup-log-swap", max_log_files=1, profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + try: + with mock.patch.object(app_module, "prune_log_files", side_effect=fail_after_swapping_log_ancestor): + status = _run(app, home) + except OSError as exc: + self.skipTest(f"directory symlinks are unavailable: {exc}") + + self.assertEqual(status, 1) + self.assertEqual(victim.read_text(encoding="utf-8"), "preserve") + self.assertTrue(layout.log_dir.is_symlink()) + self.assertTrue((parked_log_dir / "primary.log").is_file()) + self.assertEqual(logging.getLogger("base_cli.startup-log-swap").handlers, []) + + def test_post_logger_failure_erases_owned_temp_through_retained_handle(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + run_id = "fixed-run" + layout = runtime_layout(cache_root, "startup-retained-handle", run_id) + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="startup-retained-handle", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id=run_id, + ) + + def fail_with_temp_payload(*_args: object) -> None: + (layout.temp_dir / "partial-startup.txt").write_text("temporary", encoding="utf-8") + raise RuntimeError("retention unavailable") + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="startup-retained-handle", max_log_files=1, profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + with mock.patch.object(app_module, "prune_log_files", side_effect=fail_with_temp_payload): + status = _run(app, home) + + self.assertEqual(status, 1) + self.assertTrue(layout.temp_dir.is_dir()) + self.assertEqual(list(layout.temp_dir.iterdir()), []) + self.assertEqual(logging.getLogger("base_cli.startup-retained-handle").handlers, []) + def test_inherited_startup_failure_never_finalizes_or_deletes_parent_bundle(self) -> None: with tempfile.TemporaryDirectory() as tmpdir: root = Path(tmpdir) @@ -124,7 +220,9 @@ def main(ctx: base_cli.Context) -> None: self.assertEqual(status, 1) self.assertEqual(json.loads(parent_metadata.read_text(encoding="utf-8")), parent_payload) self.assertTrue(parent_run.is_dir()) - self.assertFalse((parent_run / "tmp" / "startup-inherited" / "child-run").exists()) + retained_temp = parent_run / "tmp" / "startup-inherited" / "child-run" + self.assertTrue(retained_temp.is_dir()) + self.assertEqual(list(retained_temp.iterdir()), []) self.assertEqual(logging.getLogger("base_cli.startup-inherited").handlers, []) def test_startup_rollback_preserves_preexisting_temp_content(self) -> None: @@ -210,6 +308,57 @@ def main(ctx: base_cli.Context) -> None: self.assertEqual(marker.read_text(encoding="utf-8"), "preserve") self.assertEqual(logging.getLogger("base_cli.startup-contained").handlers, []) + def test_startup_rollback_never_prunes_through_a_swapped_symlink_ancestor(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + home = root / "home" + cache_root = home / "cache" + run_id = "fixed-run" + layout = runtime_layout(cache_root, "startup-symlink-swap", run_id) + external = root / "external" + foreign_cli_dir = external / "startup-symlink-swap" + foreign_cli_dir.mkdir(parents=True) + parked_temp_parent = layout.run_root / "tmp-parked" + + def resolve_runtime(_cli_name: str, _project: base_cli.ProjectInfo | None) -> base_cli.RuntimeBinding: + return base_cli.RuntimeBinding( + cache_root=cache_root, + layout=layout, + application_home=None, + runtime_owner="startup-symlink-swap", + project_root=None, + project_name=None, + inherited_path=None, + history_parent_run_id=None, + run_id=run_id, + ) + + def fail_after_swapping_temp_ancestor(*_args: object) -> None: + temp_parent = layout.run_root / "tmp" + temp_parent.rename(parked_temp_parent) + temp_parent.symlink_to(external, target_is_directory=True) + raise RuntimeError("retention unavailable") + + profile = replace(base_cli.CliProfile.generic(), resolve_runtime=resolve_runtime) + app = base_cli.App(name="startup-symlink-swap", max_log_files=1, profile=profile) + + @app.command() + def main(ctx: base_cli.Context) -> None: + del ctx + self.fail("command should not run") + + try: + with mock.patch.object(app_module, "prune_log_files", side_effect=fail_after_swapping_temp_ancestor): + status = _run(app, home) + except OSError as exc: + self.skipTest(f"directory symlinks are unavailable: {exc}") + + self.assertEqual(status, 1) + self.assertTrue(foreign_cli_dir.is_dir()) + self.assertTrue((layout.run_root / "tmp").is_symlink()) + self.assertTrue((parked_temp_parent / "startup-symlink-swap" / run_id).is_dir()) + self.assertEqual(logging.getLogger("base_cli.startup-symlink-swap").handlers, []) + def test_rollback_cleanup_runtime_error_cannot_mask_startup_failure(self) -> None: app = base_cli.App(name="startup-rollback-runtime", max_log_files=1) @@ -225,8 +374,8 @@ def main(ctx: base_cli.Context) -> None: "prune_log_files", side_effect=RuntimeError("primary startup failure"), ), mock.patch.object( - app_module.shutil, - "rmtree", + app_module.Context, + "_cleanup_owned_temp_dir", side_effect=RuntimeError("secondary rollback failure"), ): with self.assertRaisesRegex(RuntimeError, "primary startup failure"): diff --git a/tests/test_app_subcommands.py b/tests/test_app_subcommands.py index ecd9d52..14fb369 100644 --- a/tests/test_app_subcommands.py +++ b/tests/test_app_subcommands.py @@ -84,8 +84,10 @@ def clean(ctx: base_cli.Context, target: str) -> None: self.assertEqual(seen["hello"]["name"], "Ada") self.assertEqual(seen["clean"]["target"], "cache") self.assertNotEqual(seen["hello"]["run_id"], seen["clean"]["run_id"]) - self.assertFalse(seen["hello"]["temp_dir"].exists()) - self.assertFalse(seen["clean"]["temp_dir"].exists()) + self.assertTrue(seen["hello"]["temp_dir"].is_dir()) + self.assertTrue(seen["clean"]["temp_dir"].is_dir()) + self.assertEqual(list(seen["hello"]["temp_dir"].iterdir()), []) + self.assertEqual(list(seen["clean"]["temp_dir"].iterdir()), []) self.assertTrue(seen["hello"]["cache_dir"].is_dir()) self.assertTrue(seen["clean"]["cache_dir"].is_dir()) self.assertTrue(seen["hello"]["log_file"].is_file()) diff --git a/tests/test_cleanup_security.py b/tests/test_cleanup_security.py new file mode 100644 index 0000000..7c2a068 --- /dev/null +++ b/tests/test_cleanup_security.py @@ -0,0 +1,482 @@ +from __future__ import annotations + +import io +import logging +import os +import tempfile +import unittest +from pathlib import Path +from unittest import mock + +import base_cli +import base_cli._cleanup as cleanup_module +from base_cli._cleanup import UnsafeCleanupPathError, _is_root_like, remove_owned_temp_directory + + +class CleanupSecurityTests(unittest.TestCase): + def _context( + self, + root: Path, + temp_dir: Path, + run_root: Path | None, + *, + run_id: str = "run-123", + owned: bool = True, + ) -> tuple[base_cli.Context, io.StringIO]: + stream = io.StringIO() + logger = logging.Logger(f"cleanup-security-{id(stream)}", level=logging.DEBUG) + logger.addHandler(logging.StreamHandler(stream)) + context = base_cli.Context( + cli_name="cleanup-security", + run_id=run_id, + state_dir=root / "state", + log_dir=root / "logs", + cache_dir=root / "cache", + temp_dir=temp_dir, + log_file=None, + config={}, + environment="test", + debug=False, + keep_temp=False, + log=logger, + run_root=run_root, + ) + if owned: + context._owns_temp_dir = True + try: + temp_stat = os.stat(temp_dir, follow_symlinks=False) + context._owned_temp_identity = (temp_stat.st_dev, temp_stat.st_ino) + context._owned_temp_descriptor = os.open( + temp_dir, + os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW, + ) + except OSError: + pass + return context, stream + + def test_owned_temp_contents_are_removed_while_empty_leaf_is_retained(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + (temp_dir / "payload.txt").write_text("temporary", encoding="utf-8") + (run_root / "logs").mkdir() + + context, _ = self._context(root, temp_dir, run_root) + owned_descriptor = context._owned_temp_descriptor + context.cleanup() + + self.assertTrue(temp_dir.is_dir()) + self.assertEqual(list(temp_dir.iterdir()), []) + self.assertTrue((run_root / "tmp" / "cleanup-security").is_dir()) + self.assertTrue((run_root / "tmp").is_dir()) + self.assertTrue(run_root.exists()) + self.assertTrue((run_root / "logs").exists()) + self.assertEqual(context.log.handlers, []) + self.assertIsNotNone(owned_descriptor) + with self.assertRaises(OSError): + os.fstat(owned_descriptor) + + def test_nonempty_ancestor_and_its_content_are_preserved(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + sibling = run_root / "tmp" / "keep.txt" + sibling.write_text("persistent", encoding="utf-8") + + context, _ = self._context(root, temp_dir, run_root) + context.cleanup() + + self.assertTrue(temp_dir.is_dir()) + self.assertEqual(list(temp_dir.iterdir()), []) + self.assertTrue(sibling.is_file()) + self.assertTrue((run_root / "tmp").is_dir()) + self.assertTrue(run_root.is_dir()) + + def test_preexisting_unowned_temp_tree_is_never_removed(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + marker = temp_dir / "preexisting.txt" + marker.write_text("keep", encoding="utf-8") + + context, stream = self._context(root, temp_dir, run_root, owned=False) + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertEqual(stream.getvalue(), "") + self.assertEqual(context.log.handlers, []) + + def test_keep_temp_preserves_contents_and_closes_retained_handle(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + marker = temp_dir / "keep.txt" + marker.write_text("keep", encoding="utf-8") + context, _ = self._context(root, temp_dir, run_root) + context.keep_temp = True + owned_descriptor = context._owned_temp_descriptor + + context.cleanup() + + self.assertEqual(marker.read_text(encoding="utf-8"), "keep") + self.assertFalse(context._owns_temp_dir) + self.assertIsNone(context._owned_temp_descriptor) + self.assertIsNotNone(owned_descriptor) + with self.assertRaises(OSError): + os.fstat(owned_descriptor) + + def test_traversal_target_is_refused_and_handlers_still_close(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + run_root.mkdir() + actual_target = root / "outside" / "run-123" + actual_target.mkdir(parents=True) + marker = actual_target / "keep.txt" + marker.write_text("keep", encoding="utf-8") + traversal_target = run_root / ".." / "outside" / "run-123" + + context, stream = self._context(root, traversal_target, run_root) + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertIn("path traversal is not allowed", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_outside_target_and_missing_run_id_marker_are_refused(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + run_root.mkdir() + cases = ( + (root / "outside" / "run-123", "run-123", "outside"), + (run_root / "tmp" / "wrong-marker", "run-123", "ownership marker"), + ) + for target, run_id, expected_warning in cases: + with self.subTest(target=target): + target.mkdir(parents=True) + marker = target / "keep.txt" + marker.write_text("keep", encoding="utf-8") + context, stream = self._context(root, target, run_root, run_id=run_id) + + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertIn(expected_warning, stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_symlink_target_is_refused_without_touching_its_destination(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + external = root / "external" + external.mkdir() + marker = external / "keep.txt" + marker.write_text("keep", encoding="utf-8") + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.parent.mkdir(parents=True) + try: + temp_dir.symlink_to(external, target_is_directory=True) + except OSError as exc: + self.skipTest(f"directory symlinks are unavailable: {exc}") + + context, stream = self._context(root, temp_dir, run_root) + context.cleanup() + + self.assertTrue(temp_dir.is_symlink()) + self.assertTrue(marker.is_file()) + self.assertIn("symlink cleanup targets are not allowed", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_intermediate_symlink_is_refused_even_when_it_resolves_inside_root(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + real_parent = run_root / "real-temp" + actual_target = real_parent / "cleanup-security" / "run-123" + actual_target.mkdir(parents=True) + marker = actual_target / "keep.txt" + marker.write_text("keep", encoding="utf-8") + linked_parent = run_root / "tmp" + try: + linked_parent.symlink_to(real_parent, target_is_directory=True) + except OSError as exc: + self.skipTest(f"directory symlinks are unavailable: {exc}") + temp_dir = linked_parent / "cleanup-security" / "run-123" + + context, stream = self._context(root, temp_dir, run_root) + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertTrue(linked_parent.is_symlink()) + self.assertIn("symlink cleanup targets are not allowed", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_intermediate_symlink_swap_after_validation_cannot_redirect_deletion(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_parent = run_root / "tmp" + temp_dir = temp_parent / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + owned_marker = temp_dir / "owned.txt" + owned_marker.write_text("owned", encoding="utf-8") + outside_parent = root / "outside" + outside_target = outside_parent / "cleanup-security" / "run-123" + outside_target.mkdir(parents=True) + victim = outside_target / "victim.txt" + victim.write_text("preserve", encoding="utf-8") + parked_parent = run_root / "tmp-parked" + owned_stat = temp_dir.stat() + owned_descriptor = os.open(temp_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + + def swap_intermediate() -> None: + temp_parent.rename(parked_parent) + temp_parent.symlink_to(outside_parent, target_is_directory=True) + + try: + try: + with self.assertRaisesRegex(UnsafeCleanupPathError, "symlink cleanup targets"): + remove_owned_temp_directory( + temp_dir, + run_root, + "run-123", + expected_identity=(owned_stat.st_dev, owned_stat.st_ino), + owned_descriptor=owned_descriptor, + before_remove=swap_intermediate, + ) + finally: + os.close(owned_descriptor) + except OSError as exc: + self.skipTest(f"directory symlinks are unavailable: {exc}") + + self.assertEqual(victim.read_text(encoding="utf-8"), "preserve") + self.assertEqual( + (parked_parent / "cleanup-security" / "run-123" / "owned.txt").read_text(encoding="utf-8"), + "owned", + ) + + def test_foreign_leaf_replacement_before_removal_is_never_deleted(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + (temp_dir / "owned.txt").write_text("owned", encoding="utf-8") + owned_stat = temp_dir.stat() + owned_descriptor = os.open(temp_dir, os.O_RDONLY | os.O_DIRECTORY | os.O_NOFOLLOW) + parked_owned = run_root / "parked-owned" + foreign = root / "foreign" + foreign.mkdir() + + def replace_leaf() -> None: + temp_dir.rename(parked_owned) + foreign.rename(temp_dir) + + try: + with self.assertRaisesRegex(UnsafeCleanupPathError, "changed while it was open"): + remove_owned_temp_directory( + temp_dir, + run_root, + "run-123", + expected_identity=(owned_stat.st_dev, owned_stat.st_ino), + owned_descriptor=owned_descriptor, + before_remove=replace_leaf, + ) + finally: + os.close(owned_descriptor) + + self.assertTrue(temp_dir.is_dir()) + self.assertEqual(list(temp_dir.iterdir()), []) + self.assertEqual((parked_owned / "owned.txt").read_text(encoding="utf-8"), "owned") + + def test_missing_linux_mount_identity_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + marker = temp_dir / "preserve.txt" + marker.write_text("preserve", encoding="utf-8") + context, stream = self._context(root, temp_dir, run_root) + + with mock.patch.object(cleanup_module, "_requires_mount_identity", return_value=True), mock.patch.object( + cleanup_module, + "_mount_identity", + return_value=None, + ): + context.cleanup() + + self.assertEqual(marker.read_text(encoding="utf-8"), "preserve") + self.assertIn("mount identity is unavailable", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_mount_identity_failure_closes_unpublished_child_descriptor(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + marker = temp_dir / "preserve.txt" + marker.write_text("preserve", encoding="utf-8") + context, stream = self._context(root, temp_dir, run_root) + checked_descriptors: list[int] = [] + + def fail_mount_check( + descriptor: int, + _value: os.stat_result, + _root_device: int, + _root_mount: str | None, + ) -> bool: + checked_descriptors.append(descriptor) + raise UnsafeCleanupPathError("mount identity inspection failed") + + with mock.patch.object(cleanup_module, "_crosses_mount", side_effect=fail_mount_check): + context.cleanup() + + self.assertEqual(marker.read_text(encoding="utf-8"), "preserve") + self.assertIn("mount identity inspection failed", stream.getvalue()) + self.assertEqual(len(checked_descriptors), 1) + with self.assertRaises(OSError): + os.fstat(checked_descriptors[0]) + + def test_nested_directory_skeleton_is_retained_without_any_rmdir(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + nested = temp_dir / "nested" / "deep" + nested.mkdir(parents=True) + payload = nested / "payload.txt" + payload.write_text("temporary", encoding="utf-8") + foreign = root / "foreign" + foreign.mkdir() + parked_nested = root / "parked-nested" + context, stream = self._context(root, temp_dir, run_root) + original_rmdir = cleanup_module.os.rmdir + + def destructive_rmdir(name: str, *, dir_fd: int) -> None: + del name, dir_fd + (temp_dir / "nested").rename(parked_nested) + foreign.rename(temp_dir / "nested") + original_rmdir(temp_dir / "nested") + + with mock.patch.object(cleanup_module.os, "rmdir", side_effect=destructive_rmdir) as rmdir_mock: + context.cleanup() + + rmdir_mock.assert_not_called() + self.assertFalse(payload.exists()) + self.assertTrue(nested.is_dir()) + self.assertEqual(list(nested.iterdir()), []) + self.assertTrue(foreign.is_dir()) + self.assertFalse(parked_nested.exists()) + self.assertEqual(stream.getvalue(), "") + + def test_plain_directory_replacement_never_inherits_cleanup_ownership(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + owned_marker = temp_dir / "owned.txt" + owned_marker.write_text("owned", encoding="utf-8") + context, stream = self._context(root, temp_dir, run_root) + + parked_owned = run_root / "parked-owned" + temp_dir.rename(parked_owned) + replacement = root / "replacement" + replacement.mkdir() + victim = replacement / "do-not-delete.txt" + victim.write_text("preserve", encoding="utf-8") + replacement.rename(temp_dir) + + context.cleanup() + + self.assertEqual((temp_dir / "do-not-delete.txt").read_text(encoding="utf-8"), "preserve") + self.assertEqual((parked_owned / "owned.txt").read_text(encoding="utf-8"), "owned") + self.assertIn("no longer matches the invocation-owned directory", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_target_equal_to_run_root_is_refused(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run-123" + run_root.mkdir() + marker = run_root / "keep.txt" + marker.write_text("keep", encoding="utf-8") + + context, stream = self._context(root, run_root, run_root) + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertIn("equals the run root", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_filesystem_root_and_cross_platform_root_like_paths_are_refused(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + target = root / "run-123" + target.mkdir() + marker = target / "keep.txt" + marker.write_text("keep", encoding="utf-8") + filesystem_root = Path(target.anchor) + + context, stream = self._context(root, target, filesystem_root) + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertIn("filesystem-root cleanup targets", stream.getvalue()) + self.assertTrue(_is_root_like(Path("/"))) + self.assertTrue(_is_root_like(Path("C:\\"))) + self.assertTrue(_is_root_like(Path("\\\\server\\share\\"))) + self.assertEqual(context.log.handlers, []) + + def test_mount_target_is_refused(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + marker = temp_dir / "keep.txt" + marker.write_text("keep", encoding="utf-8") + original_is_mount = cleanup_module.os.path.ismount + + def is_mount(path: Path) -> bool: + return path == temp_dir or original_is_mount(path) + + context, stream = self._context(root, temp_dir, run_root) + with mock.patch.object(cleanup_module.os.path, "ismount", side_effect=is_mount): + context.cleanup() + + self.assertTrue(marker.is_file()) + self.assertIn("mounted temp directories", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + def test_platform_without_safe_directory_handles_fails_closed(self) -> None: + with tempfile.TemporaryDirectory() as tmpdir: + root = Path(tmpdir) + run_root = root / "run" + temp_dir = run_root / "tmp" / "cleanup-security" / "run-123" + temp_dir.mkdir(parents=True) + marker = temp_dir / "keep.txt" + marker.write_text("preserve", encoding="utf-8") + context, stream = self._context(root, temp_dir, run_root) + + with mock.patch.object(cleanup_module, "_supports_fd_relative_cleanup", return_value=False): + context.cleanup() + + self.assertEqual(marker.read_text(encoding="utf-8"), "preserve") + self.assertIn("directory-handle operations", stream.getvalue()) + self.assertEqual(context.log.handlers, []) + + +if __name__ == "__main__": + unittest.main() diff --git a/tests/test_redaction_security.py b/tests/test_redaction_security.py new file mode 100644 index 0000000..8a32bc7 --- /dev/null +++ b/tests/test_redaction_security.py @@ -0,0 +1,267 @@ +from __future__ import annotations + +import unittest + +import click + +from base_cli.history import redact_history_argv +from base_cli.redaction import ( + REDACTED, + RedactionPlan, + compile_redaction_plan, + option_aliases_from_decls, + parameter_name_from_decls, + redact_argv, +) + + +def _sensitive(parameter: object) -> object: + setattr(parameter, "_base_cli_sensitive", True) + return parameter + + +def _command_tree() -> click.Group: + root_token = _sensitive(click.Option(["--root-token"])) + verbose = click.Option(["-v", "--verbose"], is_flag=True) + token = _sensitive(click.Option(["-t", "--token", "credential"])) + pair = _sensitive(click.Option(["--credential-pair"], nargs=2)) + api_key = click.Option(["--api-key"]) + split_flag = _sensitive(click.Option(["--auth/--no-auth"], is_flag=True)) + payload = _sensitive(click.Argument(["payload"])) + label = click.Argument(["label"], required=False) + push = click.Command( + "push", + params=[verbose, token, pair, api_key, split_flag, payload, label], + ) + return click.Group("tool", params=[root_token], commands={"push": push}) + + +class DeclarationTests(unittest.TestCase): + def test_explicit_destination_precedes_alias_derived_name(self) -> None: + self.assertEqual( + parameter_name_from_decls(("-t", "--token", "credential")), + "credential", + ) + + def test_alias_collection_includes_both_boolean_forms(self) -> None: + self.assertEqual( + option_aliases_from_decls(("-a", "--auth/--no-auth", "authorization_mode")), + ("-a", "--auth", "--no-auth"), + ) + + def test_destination_derivation_ignores_secondary_boolean_alias(self) -> None: + self.assertEqual(parameter_name_from_decls(("-x/--no-foo",)), "x") + + +class LegacySetRedactionTests(unittest.TestCase): + def test_raw_aliases_and_destination_names_are_supported(self) -> None: + cases = ( + (["tool", "--token", "long"], {"token"}, ["tool", "--token", REDACTED]), + (["tool", "--credential=long"], {"--credential"}, ["tool", f"--credential={REDACTED}"]), + (["tool", "-p", "short"], {"-p"}, ["tool", "-p", REDACTED]), + (["tool", "-pshort"], {"p"}, ["tool", f"-p{REDACTED}"]), + (["tool", "-p=short"], {"-p"}, ["tool", f"-p={REDACTED}"]), + (["tool", "+pplus"], {"+p"}, ["tool", f"+p{REDACTED}"]), + (["tool", "/pslash"], {"/p"}, ["tool", f"/p{REDACTED}"]), + ) + for argv, sensitive, expected in cases: + with self.subTest(argv=argv, sensitive=sensitive): + self.assertEqual(redact_argv(argv, sensitive), expected) + + def test_secret_name_heuristics_apply_without_registration(self) -> None: + argv = [ + "tool", + "--password", + "hunter2", + "API_TOKEN=value", + "https://user:pass@example.test/path", + ] + expected = [ + "tool", + "--password", + REDACTED, + f"API_TOKEN={REDACTED}", + f"https://{REDACTED}@example.test/path", + ] + + self.assertEqual(redact_argv(argv, set()), expected) + self.assertEqual(redact_history_argv(argv, set()), expected) + + def test_option_looking_values_follow_click_consumption(self) -> None: + self.assertEqual( + redact_argv(["tool", "--token", "--verbose"], {"token"}), + ["tool", "--token", REDACTED], + ) + + +class RedactionPlanTests(unittest.TestCase): + def setUp(self) -> None: + self.plan = compile_redaction_plan(_command_tree()) + + def test_plan_is_set_compatible_and_contains_destinations_and_aliases(self) -> None: + self.assertIsInstance(self.plan, set) + self.assertIsInstance(self.plan, RedactionPlan) + self.assertTrue( + {"credential", "-t", "--token", "root_token", "--root-token"}.issubset(self.plan) + ) + self.assertTrue({"--auth", "--no-auth"}.issubset(self.plan)) + + def test_recursive_plan_redacts_option_forms_and_positional_span(self) -> None: + cases = ( + ( + ["tool", "--root-token=root", "push", "--token", "value", "payload", "visible"], + ["tool", f"--root-token={REDACTED}", "push", "--token", REDACTED, REDACTED, "visible"], + ), + ( + ["tool", "push", "-vtattached", "payload", "visible"], + ["tool", "push", f"-vt{REDACTED}", REDACTED, "visible"], + ), + ( + ["tool", "push", "--credential-pair=first", "second", "payload"], + ["tool", "push", f"--credential-pair={REDACTED}", REDACTED, REDACTED], + ), + ( + ["tool", "push", "--api-key", "automatic", "payload"], + ["tool", "push", "--api-key", REDACTED, REDACTED], + ), + ( + ["tool", "push", "--", "--literal-payload", "visible"], + ["tool", "push", "--", REDACTED, "visible"], + ), + ) + for argv, expected in cases: + with self.subTest(argv=argv): + self.assertEqual(redact_argv(argv, self.plan), expected) + self.assertEqual(redact_history_argv(argv, self.plan), expected) + + def test_sensitive_positional_is_redacted_by_span_not_value_matching(self) -> None: + self.assertEqual( + redact_argv(["tool", "push", "same", "same"], self.plan), + ["tool", "push", REDACTED, "same"], + ) + + def test_option_looking_sensitive_value_follows_click_consumption(self) -> None: + self.assertEqual( + redact_argv(["tool", "push", "--token", "--verbose", "payload"], self.plan), + ["tool", "push", "--token", REDACTED, REDACTED], + ) + + def test_malformed_argv_table_is_total_and_never_drops_tokens(self) -> None: + cases = ( + (["tool", "push", "--token"], ["tool", "push", "--token"]), + (["tool", "push", "--token", "--verbose"], ["tool", "push", "--token", REDACTED]), + (["tool", "push", "--token", "--unknown"], ["tool", "push", "--token", REDACTED]), + (["tool", "push", "--token", "--"], ["tool", "push", "--token", REDACTED]), + (["tool", "push", "--token="], ["tool", "push", f"--token={REDACTED}"]), + (["tool", "push", "-t"], ["tool", "push", "-t"]), + (["tool", "push", "-t", "-v"], ["tool", "push", "-t", REDACTED]), + (["tool", "push", "-tvalue"], ["tool", "push", f"-t{REDACTED}"]), + (["tool", "push", "--", "--token", "raw"], ["tool", "push", "--", REDACTED, "raw"]), + (["tool", "unknown", "--token", "raw"], ["tool", "unknown", "--token", "raw"]), + ) + for argv, expected in cases: + with self.subTest(argv=argv): + actual = redact_argv(argv, self.plan) + self.assertEqual(actual, expected) + self.assertEqual(len(actual), len(argv)) + + def test_token_normalization_applies_to_commands_and_option_forms(self) -> None: + token = _sensitive(click.Option(["-t", "--token"])) + push = click.Command("push", params=[token]) + root = click.Group( + "tool", + commands={"push": push}, + context_settings={"token_normalize_func": str.lower}, + ) + plan = compile_redaction_plan(root) + + self.assertEqual( + redact_argv(["tool", "PUSH", "--TOKEN", "long"], plan), + ["tool", "PUSH", "--TOKEN", REDACTED], + ) + self.assertEqual( + redact_argv(["tool", "PUSH", "-Tattached"], plan), + ["tool", "PUSH", f"-T{REDACTED}"], + ) + + def test_context_parser_settings_control_sensitive_positional_spans(self) -> None: + password = _sensitive(click.Argument(["password"])) + unknown_command = click.Command( + "probe", + params=[password], + context_settings={"ignore_unknown_options": True, "allow_extra_args": True}, + ) + self.assertEqual( + redact_argv(["probe", "-hunter2"], compile_redaction_plan(unknown_command)), + ["probe", REDACTED], + ) + + verbose = click.Option(["-v", "+v"], is_flag=True) + mixed_command = click.Command( + "probe", + params=[verbose, password], + context_settings={"ignore_unknown_options": True, "allow_extra_args": True}, + ) + for cluster in ("-xv", "-vx", "-xyv", "+xv"): + with self.subTest(cluster=cluster): + self.assertEqual( + redact_argv(["probe", cluster], compile_redaction_plan(mixed_command)), + ["probe", REDACTED], + ) + + verbose = click.Option(["--verbose"], is_flag=True) + first = click.Argument(["first"]) + stopped_command = click.Command( + "probe", + params=[verbose, first, password], + context_settings={"allow_interspersed_args": False}, + ) + self.assertEqual( + redact_argv(["probe", "visible", "--verbose"], compile_redaction_plan(stopped_command)), + ["probe", "visible", REDACTED], + ) + + bang_verbose = click.Option(["!v"], is_flag=True) + punctuation_command = click.Command( + "probe", + params=[bang_verbose, first, password], + context_settings={ + "ignore_unknown_options": True, + "allow_extra_args": True, + "allow_interspersed_args": False, + }, + ) + self.assertEqual( + redact_argv( + ["probe", "!x", "!v", "supersecret"], + compile_redaction_plan(punctuation_command), + ), + ["probe", "!x", "!v", REDACTED], + ) + + def test_variadic_argument_backfill_protects_following_sensitive_argument(self) -> None: + sources = click.Argument(["sources"], nargs=-1) + password = _sensitive(click.Argument(["password"])) + command = click.Command("probe", params=[sources, password]) + + self.assertEqual( + redact_argv(["probe", "one", "secret"], compile_redaction_plan(command)), + ["probe", "one", REDACTED], + ) + + def test_duplicate_normalized_aliases_union_sensitive_markers(self) -> None: + sensitive = _sensitive(click.Option(["--credential", "secret_dest"])) + plain = click.Option(["--credential", "plain_dest"]) + command = click.Command("probe", params=[sensitive, plain]) + + self.assertEqual( + redact_argv( + ["probe", "--credential", "duplicate-secret"], + compile_redaction_plan(command), + ), + ["probe", "--credential", REDACTED], + ) + + +if __name__ == "__main__": + unittest.main()