From 79fe9e660148b28fea55959ff62d8b2820a77a2d Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Tue, 13 Oct 2026 12:58:17 -0500 Subject: [PATCH 1/5] fix: enforce FileHelper policy for shared infrastructure Route shared infrastructure and scoped manifest operations through explicit project helpers. Preserve allowed link identity and lifecycle rollback without recursively claiming newly created directories. Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 084bd222-8ae8-474a-9dcc-6e6c44e031f4 --- design/file-helper.md | 74 ++- src/specify_cli/__init__.py | 15 +- src/specify_cli/command_init.py | 2 + src/specify_cli/file_helper.py | 58 +- src/specify_cli/integrations/_file_changes.py | 16 + src/specify_cli/integrations/_helpers.py | 3 + src/specify_cli/integrations/_lifecycle.py | 113 +++- .../integrations/command_install.py | 3 + .../integrations/command_switch.py | 3 + .../integrations/command_upgrade.py | 4 + src/specify_cli/integrations/manifest.py | 73 ++- src/specify_cli/shared_infra.py | 338 ++++------ tests/integrations/test_cli.py | 75 +-- .../integrations/test_command_switch.py | 6 +- .../integrations/test_installed_adapters.py | 115 ++++ tests/specify_cli/test_shared_infra.py | 587 ++++++++++++++++++ tests/test_file_helper.py | 197 +++++- 17 files changed, 1384 insertions(+), 298 deletions(-) create mode 100644 tests/specify_cli/test_shared_infra.py diff --git a/design/file-helper.md b/design/file-helper.md index f5497ef256..879b7e57a3 100644 --- a/design/file-helper.md +++ b/design/file-helper.md @@ -2,11 +2,12 @@ `specify_cli.file_helper.FileHelper` owns filesystem operations under one explicitly established root. Project-command boundaries now validate persisted -policy, and init-options bootstrap I/O uses the helper. Other existing command -filesystem operations are **not yet migrated**: loading policy alone does not -enforce symlink behavior on those operations. Shared infrastructure, registration, -managed storage, catalogs, scripts and runtime operations retain their existing -behavior until their dependent migration stages. +policy, and init-options bootstrap I/O uses the helper. Shared infrastructure installation and template refresh now execute their +project-side operations through an explicitly handed-off helper, including their +shared manifest reads, hashes and saves. Other existing command filesystem +operations are **not yet migrated**: loading policy alone does not enforce +symlink behavior on registration, managed storage, catalogs, script execution or +runtime operations. Only `specify init --allow-symlinks` exposes the setting; init defaults to deny and persists the literal JSON boolean `allow_symlinks` in @@ -149,6 +150,13 @@ helpers for independent project/source/managed boundaries. - `mkdir(path, parents=False, exist_ok=False)` creates directories. - `read_bytes(path)` / `read_text(path, encoding="utf-8")` read regular files. +- `stat(path, follow_symlinks=True, missing_ok=False)` returns policy-checked + metadata. `missing_ok=True` returns `None` only for genuinely absent entries, + including beneath missing parents. Dangling link targets, invalid parents, + permissions, cycles and containment failures still raise. With + `follow_symlinks=False`, allow mode inspects a leaf link without following its + target; deny mode still rejects it, and linked parents and unsupported reparse + entries remain checked. - `create_bytes(path, content, mode=0o644)` / `create_text(...)` exclusively create files with existing parents. A no-follow `lstat` preflight rejects existing leaves, including dangling links, before opening the destination. @@ -195,6 +203,55 @@ delete includes the entire selected tree; there is no exclusion filter in this foundation. Future operations with exclusions must inspect only included entries, not reject a whole project for excluded/unrelated links. +## Shared infrastructure ownership + +`install_shared_infra` and `refresh_shared_templates` receive an explicit +`files: FileHelper` from init and integration lifecycle callers. Compatibility +callers that omit it resolve the selected project's current policy afresh. +Init supplies a helper using its current requested flag before final persistence, +not an earlier stored value. Config bootstrap remains deny-only. + +The helper executes project metadata probes, directory creation, reads/hashes, +atomic writes and stale managed-script deletion for `.specify/scripts//`, +`.specify/templates/`, `.specify/.gitignore`, and +`.specify/integrations/speckit.manifest.json`. Source/bundled asset operations, +archive verification, locks and other manifest consumers are unchanged. +`IntegrationManifest` has an optional helper mode for these scoped load/save, +record-existing, modification-check and lexical key-bookkeeping operations; +legacy consumers retain their existing behavior. + +Accessed denied links now abort instead of warning and succeeding with skipped +destinations, including with `--force`. Allow mode preserves contained links +while retaining missing-only, managed/unmodified refresh, recovered/customized +preservation and force-overwrite decisions. Policy violations never become a +recovered-file or replacement-manifest success. Malformed manifest content still +uses the existing warned recovery behavior. Manifest keys retain their lexical +project-relative identity even when file bytes are read through allowed aliases. +Stale leaf links remain customizations, including dangling/external-target links: +they are never hash-promoted into managed files or deleted by stale cleanup. +Eligible stale regular files reached through contained linked parents can be +removed through the helper without deleting their parent links. + +`mkdir`, `write_bytes`/`write_text`, and `delete` accept an optional explicit +`Callable[[Path, bool, bool], None]` mutation observer (path, before, removal). +The helper validates before callbacks, reports the actual contained target for +writes and directory creation, and reports the nonfollow entry for deletion. +After callbacks occur only following successful mutations; callback failures +propagate and atomic-write temporary cleanup remains active. `mkdir` reports +only genuinely new directories in parent order, never existing directory trees. +The integration lifecycle journals file/package mutations through its existing +observer and tracks directory creation separately without claiming contents. +Rollback restores observed target bytes/modes while preserving links, then +removes only identified empty new directories; unrelated contents survive. +Shared scope census retains lexical link entries and does not traverse linked +scope roots or descendants. Other adapter/package/registry/lock restrictions +are unchanged; operational permission does not waive those boundaries. +In particular, external integration transactions still reject a linked +`.specify/integrations` scope root because adapter-package and registry storage +are outside this migration. Shared manifest leaf links are supported; standalone +shared operations also support contained linked manifest parents. Integration +uninstall retains its existing scope validation. + ## Safety limits and migration Preflight detects static policy failures before recursive deletion or parent @@ -207,9 +264,10 @@ relative traversal or platform-specific mechanisms are future work, not a claim of this API. The helper does not constrain subprocesses or third-party code. Exclusive creation uses `O_EXCL` and `O_NOFOLLOW` where available. -Existing safe-write mechanisms informed this helper, but existing shared -infrastructure warning/skip behavior and development-mode link creation are -unchanged. Migrate one owned operation boundary at a time, carrying original +Existing safe-write mechanisms informed this helper; shared installation and +refresh now enforce its policy instead of warning and skipping denied links. +Development-mode registration and unmigrated private safety helpers retain their +existing behavior. Migrate one owned operation boundary at a time, carrying original paths and an explicit root/policy into the helper before any early resolution. Linked command directories and development-mode registration remain dependent stages. Non-atomic configuration writes preserve legacy metadata semantics; diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index 181b0feec1..727d61e149 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -30,6 +30,7 @@ import os import sys from pathlib import Path +from .file_helper import FileHelper import typer from rich.align import Align @@ -117,8 +118,11 @@ def _refresh_shared_templates( invoke_separator: str, invoke_prefix: str = "/", force: bool = False, + files: FileHelper | None = None, ) -> None: """Refresh default-sensitive shared templates without touching scripts.""" + from ._file_policy import project_files + _refresh_shared_templates_impl( project_path, version=get_speckit_version(), @@ -128,6 +132,7 @@ def _refresh_shared_templates( invoke_separator=invoke_separator, invoke_prefix=invoke_prefix, force=force, + files=files if files is not None else project_files(project_path), ) @@ -140,6 +145,7 @@ def _install_shared_infra( invoke_prefix: str = "/", refresh_managed: bool = False, refresh_hint: str | None = None, + files: FileHelper | None = None, ) -> bool: """Install shared infrastructure files into *project_path*. @@ -154,8 +160,8 @@ def _install_shared_infra( Overwrite policy: - * ``force=True`` — overwrite every existing file (still skips symlinks - to avoid following links outside the project root). + * ``force=True`` — overwrite every existing regular file permitted by + the operation's FileHelper policy, preserving allowed links. * ``refresh_managed=True`` — overwrite only files whose on-disk hash still matches the previously recorded manifest hash (i.e. unmodified files installed by spec-kit). Files with diverging hashes are @@ -172,6 +178,8 @@ def _install_shared_infra( Returns ``True`` on success. """ + from ._file_policy import project_files + return _install_shared_infra_impl( project_path, script_type, @@ -184,6 +192,7 @@ def _install_shared_infra( invoke_prefix=invoke_prefix, refresh_managed=refresh_managed, refresh_hint=refresh_hint, + files=files if files is not None else project_files(project_path), ) @@ -196,6 +205,7 @@ def _install_shared_infra_or_exit( invoke_prefix: str = "/", refresh_managed: bool = False, refresh_hint: str | None = None, + files: FileHelper | None = None, ) -> bool: try: return _install_shared_infra( @@ -207,6 +217,7 @@ def _install_shared_infra_or_exit( invoke_prefix=invoke_prefix, refresh_managed=refresh_managed, refresh_hint=refresh_hint, + files=files, ) except (ValueError, OSError) as exc: console.print(f"[red]Error:[/red] Failed to install shared infrastructure: {exc}") diff --git a/src/specify_cli/command_init.py b/src/specify_cli/command_init.py index cf6e9e1f39..e141ef9079 100644 --- a/src/specify_cli/command_init.py +++ b/src/specify_cli/command_init.py @@ -27,6 +27,7 @@ ) from ._console import StepTracker, console, select_with_arrows, show_banner from ._utils import check_tool +from .file_helper import FileHelper def _stdin_is_interactive() -> bool: @@ -790,6 +791,7 @@ def init( selected_script, tracker=tracker, force=force, + files=FileHelper(project_path, allow_symlinks=allow_symlinks), invoke_separator=resolved_integration.effective_invoke_separator( integration_parsed_options, project_root=project_path ), diff --git a/src/specify_cli/file_helper.py b/src/specify_cli/file_helper.py index 49441ce494..9d98170024 100644 --- a/src/specify_cli/file_helper.py +++ b/src/specify_cli/file_helper.py @@ -1,6 +1,7 @@ """Scoped filesystem operations; see design/file-helper.md for the contract.""" from dataclasses import dataclass, field +from collections.abc import Callable import codecs import errno import os @@ -11,6 +12,7 @@ _PathType = TypeVar("_PathType", bound=PurePath) +MutationObserver = Callable[[Path, bool, bool], None] def _strip_extended_length_prefix(path: _PathType) -> _PathType: @@ -153,11 +155,26 @@ def _require_file(path: Path) -> None: raise UnsupportedPathError(f"Expected a regular file: {path}") def mkdir( - self, path: Path | str, *, parents: bool = False, exist_ok: bool = False + self, path: Path | str, *, parents: bool = False, exist_ok: bool = False, + observer: MutationObserver | None = None, ) -> None: """Create a directory, preflighting the entire requested hierarchy.""" destination = self._path(path, allow_missing=True) - destination.mkdir(parents=parents, exist_ok=exist_ok) + if observer is None or destination.exists() or (not parents and not destination.parent.exists()): + destination.mkdir(parents=parents, exist_ok=exist_ok) + return + missing = [destination] + if parents: + current = destination.parent + while not current.exists(): + missing.append(current) + current = current.parent + for directory in reversed(missing): + if self._path(path, allow_missing=True) != destination: + raise FileHelperError(f"Destination changed during mkdir: {path}") + observer(directory, True, False) + directory.mkdir(exist_ok=exist_ok if directory == destination else True) + observer(directory, False, False) def read_bytes(self, path: Path | str) -> bytes: destination = self._path(path) @@ -167,6 +184,20 @@ def read_bytes(self, path: Path | str) -> bytes: def read_text(self, path: Path | str, *, encoding: str = "utf-8") -> str: return self.read_bytes(path).decode(encoding) + def stat( + self, path: Path | str, *, follow_symlinks: bool = True, missing_ok: bool = False + ) -> os.stat_result | None: + """Inspect an entry under policy; missing link targets always fail.""" + destination = self._path(path, allow_missing=True, follow_leaf=follow_symlinks) + try: + entry = destination.stat(follow_symlinks=follow_symlinks) + except FileNotFoundError: + if missing_ok: + return None + raise + self._entry_mode(destination) + return entry + def create_bytes(self, path: Path | str, content: bytes, *, mode: int = 0o644) -> None: """Exclusively create a file; never overwrite an existing entry.""" destination = self._path(path, allow_missing=True, follow_leaf=False) @@ -187,11 +218,16 @@ def create_text( ) -> None: self.create_bytes(path, content.encode(encoding), mode=mode) - def write_bytes(self, path: Path | str, content: bytes, *, mode: int = 0o644) -> None: + def write_bytes( + self, path: Path | str, content: bytes, *, mode: int = 0o644, + observer: MutationObserver | None = None, + ) -> None: """Atomically create/replace a file, preserving any allowed leaf link.""" destination = self._path(path, allow_missing=True) if destination.exists(): self._require_file(destination) + if observer is not None: + observer(destination, True, False) fd, temporary = tempfile.mkstemp(prefix=f".{destination.name}.", dir=destination.parent) temporary_path = Path(temporary) try: @@ -201,13 +237,16 @@ def write_bytes(self, path: Path | str, content: bytes, *, mode: int = 0o644) -> if self._path(path, allow_missing=True) != destination: raise FileHelperError(f"Destination changed during write: {path}") os.replace(temporary_path, destination) + if observer is not None: + observer(destination, False, False) finally: temporary_path.unlink(missing_ok=True) def write_text( - self, path: Path | str, content: str, *, encoding: str = "utf-8", mode: int = 0o644 + self, path: Path | str, content: str, *, encoding: str = "utf-8", mode: int = 0o644, + observer: MutationObserver | None = None, ) -> None: - self.write_bytes(path, content.encode(encoding), mode=mode) + self.write_bytes(path, content.encode(encoding), mode=mode, observer=observer) def overwrite_text( self, path: Path | str, content: str, *, encoding: str = "utf-8" @@ -257,7 +296,10 @@ def _deletion_plan(self, path: Path, *, recursive: bool) -> list[tuple[Path, boo plan.append((path, True)) return plan - def delete(self, path: Path | str, *, recursive: bool = False) -> None: + def delete( + self, path: Path | str, *, recursive: bool = False, + observer: MutationObserver | None = None, + ) -> None: """Delete an entry; recursive deletion never follows descendant links.""" destination = self._path(path, follow_leaf=False) if destination == self._canonical_root: @@ -268,7 +310,11 @@ def delete(self, path: Path | str, *, recursive: bool = False) -> None: checked = self._path(entry, follow_leaf=False) if checked != entry: raise FileHelperError(f"Destination changed during deletion: {entry}") + if observer is not None: + observer(entry, True, True) if directory: entry.rmdir() else: entry.unlink() + if observer is not None: + observer(entry, False, False) diff --git a/src/specify_cli/integrations/_file_changes.py b/src/specify_cli/integrations/_file_changes.py index 9ed0ebc62f..514c1e477a 100644 --- a/src/specify_cli/integrations/_file_changes.py +++ b/src/specify_cli/integrations/_file_changes.py @@ -10,6 +10,9 @@ file_change_observer: ContextVar[Callable[[Path, bool, bool], None] | None] = ContextVar( "integration_file_change_observer", default=None ) +directory_change_observer: ContextVar[Callable[[Path, bool, bool], None] | None] = ContextVar( + "integration_directory_change_observer", default=None +) def before_file_change(path: Path, *, removal: bool = False) -> None: @@ -23,6 +26,19 @@ def after_file_change(path: Path) -> None: if observer is not None: observer(path, False, False) +def observe_file_change(path: Path, before: bool, removal: bool = False) -> None: + """Forward a scoped helper's actual mutation entry to the lifecycle journal.""" + if before: + before_file_change(path, removal=removal) + else: + after_file_change(path) + +def observe_directory_change(path: Path, before: bool, removal: bool = False) -> None: + """Observe newly created directories without claiming their contents.""" + observer = directory_change_observer.get() + if observer is not None: + observer(path, before, removal) + @contextmanager def changing_file(path: Path, *, removal: bool = False): diff --git a/src/specify_cli/integrations/_helpers.py b/src/specify_cli/integrations/_helpers.py index 9bfec2e8c7..0414fbf3f5 100644 --- a/src/specify_cli/integrations/_helpers.py +++ b/src/specify_cli/integrations/_helpers.py @@ -329,10 +329,13 @@ def _set_default_integration( ) if refresh_templates: + from .._file_policy import project_files + files = project_files(project_root) try: _install_shared_infra( project_root, resolved_script, + files=files, invoke_separator=_invoke_separator_for_integration( integration, {"integration_settings": settings}, key, parsed_options, project_root=project_root, diff --git a/src/specify_cli/integrations/_lifecycle.py b/src/specify_cli/integrations/_lifecycle.py index 0b9bd7da01..3698444258 100644 --- a/src/specify_cli/integrations/_lifecycle.py +++ b/src/specify_cli/integrations/_lifecycle.py @@ -19,13 +19,15 @@ from rich.markup import escape from .._console import console +from .._file_policy import project_files +from ..file_helper import FileHelper from ..integration_state import ( default_integration_key, installed_integration_keys, try_read_integration_json, ) from . import BUILTIN_INTEGRATION_KEYS, INTEGRATION_REGISTRY, installer -from ._file_changes import file_change_observer +from ._file_changes import directory_change_observer, file_change_observer _init_directory: ContextVar[tuple[Path, bool, list[Path]] | None] = ContextVar( "external_integration_init_directory", default=None @@ -143,6 +145,8 @@ def __init__( self.backup_bytes = 0 self.backup_entries = 0 self.directory_states: dict[Path, bool] = {} + self.created_directories: dict[Path, tuple[int, int]] = {} + self.pending_directories: set[Path] = set() for path in paths: self._remember_parents(path) @@ -256,6 +260,25 @@ def observe(self, path: Path, before: bool, removal: bool = False) -> None: else: self.pending.add(path) + def observe_directory(self, path: Path, before: bool, removal: bool = False) -> None: + path = self.safe_path(path, allow_leaf_symlink=False) + if before: + try: + path.lstat() + except FileNotFoundError: + pass + else: + raise installer.IntegrationInstallError(f"Directory changed before creation: {path}") + self._remember_parents(path) + self.directory_states[path] = False + self.pending_directories.add(path) + else: + entry = path.lstat() + if path not in self.pending_directories or not stat.S_ISDIR(entry.st_mode): + raise installer.IntegrationInstallError(f"Unobserved directory creation: {path}") + self.created_directories[path] = (entry.st_dev, entry.st_ino) + self.pending_directories.remove(path) + def _restore_snapshots(root: Path, journal: _FileJournal) -> list[Path]: conflicts = [] @@ -286,11 +309,55 @@ def _restore_snapshots(root: Path, journal: _FileJournal) -> list[Path]: parent = path.parent boundary = root if path.is_relative_to(root) else Path.home().absolute() while parent != boundary and not journal.directory_states[parent]: + if parent in journal.created_directories: + break try: parent.rmdir() except OSError: break parent = parent.parent + for path, identity in sorted(journal.created_directories.items(), key=lambda item: len(item[0].parts), reverse=True): + journal.safe_path(path) + try: + entry = path.lstat() + except FileNotFoundError: + continue + if not stat.S_ISDIR(entry.st_mode) or (entry.st_dev, entry.st_ino) != identity: + conflicts.append(path) + continue + try: + path.rmdir() + except FileNotFoundError: + continue + except OSError as exc: + if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST): + raise + continue + # Legacy writers also create uncensused parents. Resume their existing + # empty-parent cleanup once deferred helper-created children are gone. + parent = path.parent + boundary = root if path.is_relative_to(root) else Path.home().absolute() + while parent != boundary and not journal.directory_states.get(parent, True): + if parent in journal.created_directories: + break + journal.safe_path(parent, allow_leaf_symlink=False) + try: + parent.rmdir() + except FileNotFoundError: + pass + except OSError as exc: + if exc.errno not in (errno.ENOTEMPTY, errno.EEXIST): + raise + break + parent = parent.parent + for path in journal.pending_directories: + journal.safe_path(path) + try: + path.lstat() + except FileNotFoundError: + continue + else: + conflicts.append(path) return conflicts @@ -298,12 +365,16 @@ def _restore_snapshots(root: Path, journal: _FileJournal) -> list[Path]: def _transaction( root: Path, target: str | None, expected_state: dict[str, Any] | None, expected_records: dict[str, dict[str, Any]], - *, creation_root: Path | None = None, + *, creation_root: Path | None = None, shared_files: FileHelper | None = None, ): """Journal operation-owned writes under an inter-process integration lock.""" + from ..file_helper import UnsupportedPathError from ..shared_infra import _exclusive_project_lock from .manifest import IntegrationManifest + if shared_files is not None and shared_files.root.resolve() != root.resolve(): + raise installer.IntegrationInstallError("Shared FileHelper must own the transaction project") + folders = { ".specify/integrations", ".specify/templates", ".specify/scripts", ".specify/.gitignore", ".specify/presets", ".specify/extensions", @@ -331,12 +402,20 @@ def _transaction( target_folder = (target_integration.config or {}).get("folder") if target_integration else None if target_folder: folders.add(target_folder.rstrip("/")) - paths = [ - installer.safe_project_path( - root, folder, allow_leaf_symlink=folder in manifest_leaves and folder not in folders - ) - for folder in sorted(folders | manifest_leaves) - ] + paths = [] + for folder in sorted(folders | manifest_leaves): + if shared_files is not None and folder in {".specify/scripts", ".specify/templates", ".specify/.gitignore"}: + entry = shared_files.stat(folder, missing_ok=True) + expected = stat.S_ISREG if folder == ".specify/.gitignore" else stat.S_ISDIR + if entry is not None and not expected(entry.st_mode): + raise UnsupportedPathError(f"Shared infrastructure scope has an invalid type: {folder}") + # Retain lexical scope identity. The lstat census does not descend + # into a linked scope; helper callbacks journal actual target entries. + paths.append(root / folder) + else: + paths.append(installer.safe_project_path( + root, folder, allow_leaf_symlink=folder in manifest_leaves and folder not in folders + )) home_scopes = [] for key in {*installed_integration_keys(state or {}), target} & BUILTIN_INTEGRATION_KEYS: integration = INTEGRATION_REGISTRY[key] @@ -367,6 +446,7 @@ def _transaction( ) journal = _FileJournal(root, paths, backup, _initial_entries(paths), tuple(home_scopes)) token = file_change_observer.set(journal.observe) + directory_token = directory_change_observer.set(journal.observe_directory) try: with installer.event_recovery_transaction(): yield @@ -375,6 +455,8 @@ def _transaction( # Restoration itself must not add writes to the journal. file_change_observer.reset(token) token = None + directory_change_observer.reset(directory_token) + directory_token = None conflicts = _restore_snapshots(root, journal) if conflicts: preserve_backup = True @@ -393,6 +475,8 @@ def _transaction( finally: if token is not None: file_change_observer.reset(token) + if directory_token is not None: + directory_change_observer.reset(directory_token) finally: if not preserve_backup: shutil.rmtree(backup) @@ -488,9 +572,18 @@ def invoke(*args, **kwargs): ) stack.callback(_init_directory.reset, token) transaction = ( - _transaction(root, key, state, records, creation_root=creation_root) + _transaction( + root, key, state, records, creation_root=creation_root, + shared_files=( + FileHelper(root, allow_symlinks=values.get("allow_symlinks", False)) + if root.exists() else None + ), + ) if operation == "init" - else _transaction(root, key, state, records) + else _transaction( + root, key, state, records, + shared_files=project_files(root) if operation != "uninstall" else None, + ) ) with transaction: if recovery_token is not None and recovery_binding is None: diff --git a/src/specify_cli/integrations/command_install.py b/src/specify_cli/integrations/command_install.py index c6b8e3e728..57cefa2bf2 100644 --- a/src/specify_cli/integrations/command_install.py +++ b/src/specify_cli/integrations/command_install.py @@ -38,6 +38,8 @@ def integration_install( from .. import _require_specify_project, _install_shared_infra_or_exit project_root = _require_specify_project() + from .._file_policy import project_files + files = project_files(project_root) integration = get_integration(key) if integration is None: console.print(f"[red]Error:[/red] Unknown integration '{key}'") @@ -116,6 +118,7 @@ def integration_install( _install_shared_infra_or_exit( project_root, selected_script, + files=files, invoke_separator=_invoke_separator_for_integration( infra_integration, current, infra_key, infra_parsed, project_root=project_root, diff --git a/src/specify_cli/integrations/command_switch.py b/src/specify_cli/integrations/command_switch.py index 39139fb393..0d08c68e23 100644 --- a/src/specify_cli/integrations/command_switch.py +++ b/src/specify_cli/integrations/command_switch.py @@ -35,6 +35,8 @@ def integration_switch( from .. import _print_cli_warning, _require_specify_project, _install_shared_infra_or_exit project_root = _require_specify_project() + from .._file_policy import project_files + files = project_files(project_root) target_integration = get_integration(target) if target_integration is None: console.print(f"[red]Error:[/red] Unknown integration '{target}'") @@ -239,6 +241,7 @@ def integration_switch( _install_shared_infra_or_exit( project_root, selected_script, + files=files, force=refresh_shared_infra, refresh_managed=True, invoke_separator=_invoke_separator_for_integration( diff --git a/src/specify_cli/integrations/command_upgrade.py b/src/specify_cli/integrations/command_upgrade.py index 004efa69af..c7e2ac10bd 100644 --- a/src/specify_cli/integrations/command_upgrade.py +++ b/src/specify_cli/integrations/command_upgrade.py @@ -46,6 +46,8 @@ def integration_upgrade( from .. import _require_specify_project, _install_shared_infra_or_exit, _install_shared_infra project_root = _require_specify_project() + from .._file_policy import project_files + files = project_files(project_root) current = _read_integration_json(project_root) installed_key = _default_integration_key(current) installed_keys = _installed_integration_keys(current) @@ -207,6 +209,7 @@ def integration_upgrade( _install_shared_infra_or_exit( project_root, selected_script, + files=files, force=force, invoke_separator=_invoke_separator_for_integration( infra_integration, current, infra_key, infra_parsed, @@ -254,6 +257,7 @@ def integration_upgrade( _install_shared_infra( project_root, selected_script, + files=files, invoke_separator=_invoke_separator_for_integration( integration, {"integration_settings": settings}, key, parsed_options, project_root=project_root, diff --git a/src/specify_cli/integrations/manifest.py b/src/specify_cli/integrations/manifest.py index cd1f430c43..9d3b93667b 100644 --- a/src/specify_cli/integrations/manifest.py +++ b/src/specify_cli/integrations/manifest.py @@ -11,12 +11,19 @@ import hashlib import json import os +import stat import tempfile from datetime import datetime, timezone from pathlib import Path from typing import Any -from ._file_changes import after_file_change, before_file_change +from ..file_helper import FileHelper, FileHelperError +from ._file_changes import ( + after_file_change, + before_file_change, + observe_directory_change, + observe_file_change, +) OWNERSHIP_MODES = frozenset({"whole", "partial", "shared"}) @@ -160,6 +167,7 @@ def __init__( version: str = "", *, resolve_project_root: bool = True, + files: FileHelper | None = None, ) -> None: self.key = key self.project_root = ( @@ -168,6 +176,9 @@ def __init__( else project_root.absolute() ) self.version = version + if files is not None and files.root.resolve() != self.project_root.resolve(): + raise ValueError("Manifest FileHelper must own the selected project root") + self._file_helper = files self._files: dict[str, str] = {} # rel_path → sha256 hex self._ownership_modes: dict[str, str] = {} self._recovered_files: set[str] = set() @@ -244,6 +255,18 @@ def record_existing( if ownership is not None: _validate_ownership_mode(ownership) rel = Path(rel_path) + if self._file_helper is not None: + normalized = self._scoped_key(rel) + self._files[normalized] = hashlib.sha256( + self._file_helper.read_bytes(rel) + ).hexdigest() + if ownership is not None: + self._ownership_modes[normalized] = ownership + if recovered: + self._recovered_files.add(normalized) + else: + self._recovered_files.discard(normalized) + return abs_path = _validate_record_path(rel, self.project_root) if not abs_path.is_file(): raise ValueError( @@ -280,6 +303,13 @@ def remove(self, rel_path: str | Path) -> bool: rel = Path(rel_path) if rel.is_absolute() or ".." in rel.parts: return False + if self._file_helper is not None: + if rel.anchor: + return False + normalized = self._scoped_key(rel) + self._recovered_files.discard(normalized) + self._ownership_modes.pop(normalized, None) + return self._files.pop(normalized, None) is not None try: abs_path = _validate_rel_path(rel, self.project_root) normalized = abs_path.relative_to(self.project_root).as_posix() @@ -330,6 +360,10 @@ def is_recovered(self, rel_path: str | Path) -> bool: rel = Path(rel_path) if rel.is_absolute() or ".." in rel.parts: return False + if self._file_helper is not None: + if rel.anchor: + return False + return self._scoped_key(rel) in self._recovered_files try: abs_path = _validate_rel_path(rel, self.project_root) normalized = abs_path.relative_to(self.project_root).as_posix() @@ -345,6 +379,24 @@ def check_modified(self) -> list[str]: # Skip paths that are absolute or attempt to escape the project root if rel_path.is_absolute() or ".." in rel_path.parts: continue + if self._file_helper is not None: + if rel_path.anchor: + continue + entry = self._file_helper.stat(rel_path, missing_ok=True) + if entry is None: + continue + if not stat.S_ISREG(entry.st_mode): + modified.append(rel) + continue + try: + changed = hashlib.sha256(self._file_helper.read_bytes(rel_path)).hexdigest() != expected_hash + except (FileHelperError, FileNotFoundError): + raise + except OSError: + changed = True + if changed: + modified.append(rel) + continue abs_path = self.project_root / rel_path if not abs_path.exists() and not abs_path.is_symlink(): continue @@ -481,6 +533,12 @@ def uninstall( # -- Persistence ------------------------------------------------------ + @staticmethod + def _scoped_key(rel: Path) -> str: + if rel.anchor or ".." in rel.parts: + raise ValueError(f"Manifest paths must be canonical and relative: {rel}") + return rel.as_posix() + def save(self) -> Path: """Write the manifest to disk. Returns the manifest path.""" self._installed_at = self._installed_at or datetime.now(timezone.utc).isoformat() @@ -501,6 +559,13 @@ def save(self) -> Path: } path = self.manifest_path content = json.dumps(data, indent=2) + "\n" + if self._file_helper is not None: + self._file_helper.stat(path, missing_ok=True) + self._file_helper.mkdir(path.parent, parents=True, exist_ok=True, observer=observe_directory_change) + self._file_helper.write_text( + path, content.replace("\n", os.linesep), observer=observe_file_change, + ) + return path before_file_change(path) _ensure_safe_manifest_destination(self.project_root, path) fd, temp_name = tempfile.mkstemp(prefix=f".{path.name}.", dir=path.parent) @@ -523,15 +588,17 @@ def load( project_root: Path, *, resolve_project_root: bool = True, + files: FileHelper | None = None, ) -> IntegrationManifest: """Load an existing manifest from disk. Raises ``FileNotFoundError`` if the manifest does not exist. """ - inst = cls(key, project_root, resolve_project_root=resolve_project_root) + inst = cls(key, project_root, resolve_project_root=resolve_project_root, files=files) path = inst.manifest_path try: - data = json.loads(path.read_text(encoding="utf-8")) + text = files.read_text(path) if files is not None else path.read_text(encoding="utf-8") + data = json.loads(text) except UnicodeDecodeError as exc: raise ValueError( f"Integration manifest at {path} is not valid UTF-8" diff --git a/src/specify_cli/shared_infra.py b/src/specify_cli/shared_infra.py index 134e329758..af79f43474 100644 --- a/src/specify_cli/shared_infra.py +++ b/src/specify_cli/shared_infra.py @@ -8,11 +8,15 @@ import logging import os import re +import stat import tempfile from pathlib import Path from typing import Any +from ._file_policy import project_files +from .file_helper import FileHelper, FileHelperError from .integrations.base import IntegrationBase +from .integrations._file_changes import observe_directory_change, observe_file_change from .integrations.manifest import IntegrationManifest logger = logging.getLogger(__name__) @@ -119,15 +123,20 @@ def load_speckit_manifest( *, version: str, console: Any | None = None, + files: FileHelper | None = None, ) -> IntegrationManifest: """Load the shared infrastructure manifest, preserving existing entries.""" + project_path = Path(project_path).absolute() manifest_path = project_path / ".specify" / "integrations" / "speckit.manifest.json" - if manifest_path.exists(): + files = files if files is not None else project_files(project_path) + if files.stat(manifest_path, missing_ok=True) is not None: try: - manifest = IntegrationManifest.load("speckit", project_path) + manifest = IntegrationManifest.load("speckit", project_path, files=files) manifest.version = version return manifest - except (ValueError, FileNotFoundError, OSError, UnicodeDecodeError) as exc: + except (FileHelperError, FileNotFoundError): + raise + except (ValueError, OSError, UnicodeDecodeError) as exc: if console is not None: console.print( f"[yellow]Warning:[/yellow] Could not read shared infrastructure " @@ -137,7 +146,7 @@ def load_speckit_manifest( "A new shared manifest will be created; previously tracked " "shared files may be treated as untracked." ) - return IntegrationManifest("speckit", project_path, version=version) + return IntegrationManifest("speckit", project_path, version=version, files=files) def shared_templates_source( @@ -415,28 +424,31 @@ def refresh_shared_templates( invoke_separator: str, invoke_prefix: str = "/", force: bool = False, + files: FileHelper | None = None, ) -> None: """Refresh default-sensitive shared templates without touching scripts.""" + project_path = Path(project_path).absolute() templates_src = shared_templates_source(core_pack=core_pack, repo_root=repo_root) if not templates_src.is_dir(): return - manifest = load_speckit_manifest(project_path, version=version, console=console) + files = files if files is not None else project_files(project_path) + manifest = load_speckit_manifest(project_path, version=version, console=console, files=files) tracked_files = manifest.files modified = set(manifest.check_modified()) skipped_files: list[str] = [] planned_updates: list[tuple[Path, str, str]] = [] dest_templates = project_path / ".specify" / "templates" - _ensure_safe_shared_directory(project_path, dest_templates) + files.mkdir(dest_templates, parents=True, exist_ok=True, observer=observe_directory_change) for src in templates_src.iterdir(): if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."): continue dst = dest_templates / src.name - _ensure_safe_shared_destination(project_path, dst) + entry = files.stat(dst, missing_ok=True) rel = dst.relative_to(project_path).as_posix() - if dst.exists() and not force: + if entry is not None and not force: if rel not in tracked_files or rel in modified or manifest.is_recovered(rel): # Never overwrite a recovered (pre-existing user) file without # --force, matching install_shared_infra's is_recovered gate @@ -451,7 +463,7 @@ def refresh_shared_templates( planned_updates.append((dst, rel, content)) for dst, rel, content in planned_updates: - _write_shared_text(project_path, dst, content) + files.write_text(dst, content, observer=observe_file_change) manifest.record_existing(rel) manifest.save() @@ -477,6 +489,7 @@ def install_shared_infra( invoke_prefix: str = "/", refresh_managed: bool = False, refresh_hint: str | None = None, + files: FileHelper | None = None, ) -> bool: """Install shared scripts and templates into *project_path*. @@ -484,31 +497,35 @@ def install_shared_infra( the previously recorded manifest hash are overwritten with the bundled version. Files whose hash diverges are treated as user customizations and preserved with a warning. ``force=True`` overwrites every regular file - (symlinks and symlinked-parent destinations are always preserved with a - warning — the safe-destination check refuses to follow them so writes - cannot escape the project root). ``refresh_hint`` is shown after the + permitted by the operation's FileHelper policy, preserving allowed links. + Denied or invalid accessed links fail explicitly. ``refresh_hint`` is shown after the customization warning to tell the user which flag would overwrite their customizations. """ - from .integrations.manifest import _sha256, _validate_rel_path - - manifest = load_speckit_manifest(project_path, version=version, console=console) + project_path = Path(project_path).absolute() + files = files if files is not None else project_files(project_path) + manifest = load_speckit_manifest(project_path, version=version, console=console, files=files) prior_hashes = dict(manifest.files) + def _is_file(dst: Path) -> bool: + entry = files.stat(dst, missing_ok=True) + return entry is not None and stat.S_ISREG(entry.st_mode) + def _is_managed(rel: str, dst: Path) -> bool: expected = prior_hashes.get(rel) - if not expected or not dst.is_file() or dst.is_symlink(): + if not expected or not _is_file(dst): return False if manifest.is_recovered(rel): return False try: - return _sha256(dst) == expected + return hashlib.sha256(files.read_bytes(dst)).hexdigest() == expected + except FileNotFoundError: + raise except OSError: return False skipped_files: list[str] = [] preserved_user_files: list[str] = [] - symlinked_files: list[str] = [] planned_copies: list[tuple[Path, str, bytes, int]] = [] planned_templates: list[tuple[Path, str, str]] = [] # Track every shared path the current bundle produces so we can detect @@ -524,7 +541,7 @@ def _is_managed(rel: str, dst: Path) -> bool: def _decide_overwrite(rel: str, dst: Path) -> tuple[bool, str | None]: """Return (write, bucket) where bucket is 'skip', 'preserved', or None.""" - if not dst.exists(): + if files.stat(dst, missing_ok=True) is None: return True, None if force: return True, None @@ -536,156 +553,87 @@ def _decide_overwrite(rel: str, dst: Path) -> tuple[bool, str | None]: return False, "skip" return False, "skip" - def _safe_dest_or_bucket(dst: Path, rel: str, *, parent_must_exist: bool = True) -> bool: - """Run the safe-destination check and bucket symlinked paths. - - Returns True when the destination is safe to consider (write or skip). - Returns False (and records *rel* under ``symlinked_files``) when the - destination or any of its ancestors is a symlink — those paths can't - be written to safely, but they shouldn't abort the whole switch - either. They're surfaced as a separate "symlinked" warning bucket. - - Other unsafe-path errors (e.g. path escape, parent-not-a-directory) - are NOT caught here: they re-raise so the operation aborts, since - treating them as "symlinked" would mask security-relevant failures. - """ - try: - _ensure_safe_shared_destination(project_path, dst, parent_must_exist=parent_must_exist) - except SymlinkedSharedPathError: - symlinked_files.append(rel) - return False - return True - - def _ensure_or_bucket_dir(directory: Path) -> bool: - """Create *directory* unless an ancestor is symlinked. + def _ensure_dir(directory: Path) -> None: + files.mkdir(directory, parents=True, exist_ok=True, observer=observe_directory_change) - Returns True when the directory is safe to use. Returns False (and - records the path under ``symlinked_files``) when a symlink ancestor - forces us to skip the whole subtree. Other unsafe-path errors - (escape, not-a-directory) re-raise so the operation aborts. - """ - try: - _ensure_safe_shared_directory(project_path, directory) - except SymlinkedSharedPathError: - symlinked_files.append(directory.relative_to(project_path).as_posix()) - return False - return True + def _consider_destination(dst: Path, rel: str) -> bool: + write, bucket = _decide_overwrite(rel, dst) + if write: + return True + if bucket == "preserved": + preserved_user_files.append(rel) + else: + skipped_files.append(rel) + # Observed files are recovered, not a managed baseline (#2107). + if rel not in prior_hashes and _is_file(dst): + try: + manifest.record_existing(rel, recovered=True) + except (FileHelperError, FileNotFoundError): + raise + except (OSError, ValueError) as exc: + console.print( + f"[yellow]⚠[/yellow] could not record {rel} in manifest: {exc}" + ) + return False scripts_src = shared_scripts_source(core_pack=core_pack, repo_root=repo_root) if scripts_src.is_dir(): dest_scripts = project_path / ".specify" / "scripts" - if _ensure_or_bucket_dir(dest_scripts): - for variant_dir in variant_dirs: - variant_src = scripts_src / variant_dir - if not variant_src.is_dir(): + _ensure_dir(dest_scripts) + for variant_dir in variant_dirs: + variant_src = scripts_src / variant_dir + if not variant_src.is_dir(): + continue + dest_variant = dest_scripts / variant_dir + _ensure_dir(dest_variant) + for src_path in variant_src.rglob("*"): + if not src_path.is_file(): continue - dest_variant = dest_scripts / variant_dir - if not _ensure_or_bucket_dir(dest_variant): + if "__pycache__" in src_path.parts: continue - for src_path in variant_src.rglob("*"): - if not src_path.is_file(): - continue - # Python bytecode caches are local artifacts, not - # workflow scripts — never install them. - if "__pycache__" in src_path.parts: - continue - # Mark scanned only once a real source file is seen. An - # empty (or symlink-skipped) variant stays untracked, so - # stale-cleanup cannot treat its managed scripts as obsolete. - scanned_variant_dirs.add(variant_dir) - - rel_path = src_path.relative_to(variant_src) - dst_path = dest_variant / rel_path - rel = dst_path.relative_to(project_path).as_posix() - seen_rels.add(rel) - if not _safe_dest_or_bucket(dst_path, rel, parent_must_exist=False): - continue - write, bucket = _decide_overwrite(rel, dst_path) - if not write: - if bucket == "preserved": - preserved_user_files.append(rel) - else: - skipped_files.append(rel) - # Record the existing-on-disk file in the manifest so a - # fresh manifest run against an already-populated - # ``.specify/`` tree does not silently drop it (#2107). - # ``prior_hashes`` is the function-scope snapshot taken - # at entry, so this membership check is O(1) and avoids - # the repeated ``dict(self._files)`` copy that - # ``manifest.files`` performs on every access. - if dst_path.is_file() and rel not in prior_hashes: - try: - manifest.record_existing(rel, recovered=True) - except (OSError, ValueError) as exc: - # Tolerate races / permission issues / non-file - # collisions so one weird path does not abort - # the whole install. - console.print( - f"[yellow]⚠[/yellow] could not record {rel} in manifest: {exc}" - ) - continue - - if not _ensure_or_bucket_dir(dst_path.parent): - continue - content = src_path.read_text(encoding="utf-8") - content = IntegrationBase.resolve_command_refs( - content, invoke_separator, invoke_prefix - ) - content = _resolve_dynamic_command_refs( - content, invoke_separator, invoke_prefix - ) - planned_copies.append( - ( - dst_path, - rel, - content.encode("utf-8"), - src_path.stat().st_mode & 0o777, - ) + # Empty/missing sources never trigger stale cleanup. + scanned_variant_dirs.add(variant_dir) + dst_path = dest_variant / src_path.relative_to(variant_src) + rel = dst_path.relative_to(project_path).as_posix() + seen_rels.add(rel) + if not _consider_destination(dst_path, rel): + continue + _ensure_dir(dst_path.parent) + content = src_path.read_text(encoding="utf-8") + content = IntegrationBase.resolve_command_refs( + content, invoke_separator, invoke_prefix + ) + content = _resolve_dynamic_command_refs( + content, invoke_separator, invoke_prefix + ) + planned_copies.append( + ( + dst_path, + rel, + content.encode("utf-8"), + src_path.stat().st_mode & 0o777, ) + ) templates_src = shared_templates_source(core_pack=core_pack, repo_root=repo_root) if templates_src.is_dir(): dest_templates = project_path / ".specify" / "templates" - if _ensure_or_bucket_dir(dest_templates): - for src in templates_src.iterdir(): - if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."): - continue + _ensure_dir(dest_templates) + for src in templates_src.iterdir(): + if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."): + continue - dst = dest_templates / src.name - rel = dst.relative_to(project_path).as_posix() - seen_rels.add(rel) - if not _safe_dest_or_bucket(dst, rel): - continue - write, bucket = _decide_overwrite(rel, dst) - if not write: - if bucket == "preserved": - preserved_user_files.append(rel) - else: - skipped_files.append(rel) - # Record the existing-on-disk template in the manifest so a - # fresh manifest run against an already-populated - # ``.specify/`` tree does not silently drop it (#2107). - # ``prior_hashes`` is the function-scope snapshot taken at - # entry, so this membership check is O(1) and avoids the - # repeated ``dict(self._files)`` copy that ``manifest.files`` - # performs on every access. - if dst.is_file() and rel not in prior_hashes: - try: - manifest.record_existing(rel, recovered=True) - except (OSError, ValueError) as exc: - # Tolerate races / permission issues / non-file - # collisions so one weird path does not abort - # the whole install. - console.print( - f"[yellow]⚠[/yellow] could not record {rel} in manifest: {exc}" - ) - continue + dst = dest_templates / src.name + rel = dst.relative_to(project_path).as_posix() + seen_rels.add(rel) + if not _consider_destination(dst, rel): + continue - content = src.read_text(encoding="utf-8") - content = IntegrationBase.resolve_command_refs( - content, invoke_separator, invoke_prefix - ) - planned_templates.append((dst, rel, content)) + content = src.read_text(encoding="utf-8") + content = IntegrationBase.resolve_command_refs( + content, invoke_separator, invoke_prefix + ) + planned_templates.append((dst, rel, content)) # Managed ``.specify/.gitignore`` — keeps machine-local state (the # ``feature.json`` pointer and per-machine ``local-config.yml`` overrides) @@ -695,36 +643,22 @@ def _ensure_or_bucket_dir(directory: Path) -> bool: # tracked in ``speckit.manifest.json`` (not the per-integration manifest) and # is therefore intentionally left in place by ``integration uninstall``. specify_dir = project_path / ".specify" - if _ensure_or_bucket_dir(specify_dir): - gitignore_dst = specify_dir / ".gitignore" - gitignore_rel = gitignore_dst.relative_to(project_path).as_posix() - seen_rels.add(gitignore_rel) - if _safe_dest_or_bucket(gitignore_dst, gitignore_rel): - write, bucket = _decide_overwrite(gitignore_rel, gitignore_dst) - if write: - planned_templates.append( - (gitignore_dst, gitignore_rel, SPECIFY_GITIGNORE_CONTENT) - ) - elif bucket == "preserved": - preserved_user_files.append(gitignore_rel) - else: - skipped_files.append(gitignore_rel) - if gitignore_dst.is_file() and gitignore_rel not in prior_hashes: - try: - manifest.record_existing(gitignore_rel, recovered=True) - except (OSError, ValueError) as exc: - console.print( - f"[yellow]⚠[/yellow] could not record {gitignore_rel} in manifest: {exc}" - ) + _ensure_dir(specify_dir) + gitignore_dst = specify_dir / ".gitignore" + gitignore_rel = gitignore_dst.relative_to(project_path).as_posix() + seen_rels.add(gitignore_rel) + if _consider_destination(gitignore_dst, gitignore_rel): + planned_templates.append( + (gitignore_dst, gitignore_rel, SPECIFY_GITIGNORE_CONTENT) + ) for dst_path, rel, content, mode in planned_copies: - if not _ensure_or_bucket_dir(dst_path.parent): - continue - _write_shared_bytes(project_path, dst_path, content, mode=mode) + _ensure_dir(dst_path.parent) + files.write_bytes(dst_path, content, mode=mode, observer=observe_file_change) manifest.record_existing(rel) for dst, rel, content in planned_templates: - _write_shared_text(project_path, dst, content) + files.write_text(dst, content, observer=observe_file_change) manifest.record_existing(rel) if skipped_files: @@ -742,19 +676,6 @@ def _ensure_or_bucket_dir(directory: Path) -> bool: "[cyan]specify integration upgrade --force[/cyan]." ) - if symlinked_files: - console.print( - f"[yellow]⚠[/yellow] Skipped {len(symlinked_files)} symlinked shared " - "infrastructure path(s) — symlinks are never overwritten because they " - "may resolve outside the project root:" - ) - for path in symlinked_files: - console.print(f" {path}") - console.print( - "To restore the bundled version, remove or replace the symlink manually, " - "then re-run the command." - ) - if preserved_user_files: console.print( f"[yellow]⚠[/yellow] Preserved {len(preserved_user_files)} customized shared " @@ -773,7 +694,7 @@ def _ensure_or_bucket_dir(directory: Path) -> bool: # was actually scanned (so a missing/empty source never triggers mass # deletion), scoped to the selected variants, and only for *managed* copies — # a user-customized file (hash diverges), a symlink, or a recovered entry is - # preserved by ``_is_managed``. + # preserved by the leaf-identity and managed-hash checks. if scanned_variant_dirs: stale_removed: list[str] = [] script_prefixes = tuple( @@ -783,38 +704,23 @@ def _ensure_or_bucket_dir(directory: Path) -> bool: if rel in seen_rels or not rel.startswith(script_prefixes): continue # Guard corrupted/hand-edited manifest keys BEFORE any filesystem - # access: absolute, ``..``, or (on Windows) drive-relative keys such - # as ``C:tmp`` are not ``is_absolute()`` yet discard the project root - # when joined. The lexical check is a fast reject; ``_validate_rel_path`` - # resolves the join and confirms containment, catching the rest. A key - # that still escapes is *skipped*, never turned into an install-time - # hard failure. Mirrors IntegrationManifest.is_recovered / remove. + # access. Invalid keys are ignored; accessed paths obey helper policy. rel_path = Path(rel) - if rel_path.is_absolute() or ".." in rel_path.parts: - continue - try: - _validate_rel_path(rel_path, project_path) - except ValueError: + if rel_path.anchor or ".." in rel_path.parts: continue dst = project_path / rel_path + entry = files.stat(dst, follow_symlinks=False, missing_ok=True) # Already gone from disk but still tracked: drop the orphaned manifest # entry so the manifest stays consistent (nothing to unlink). - if not dst.exists() and not dst.is_symlink(): + if entry is None: manifest.remove(rel) continue + if stat.S_ISLNK(entry.st_mode): + continue if not _is_managed(rel, dst): continue # user-modified / symlink / recovered → preserve - # Never unlink through a symlinked ancestor (writes/deletes could - # escape the project root). The safe-destination check buckets such - # paths under ``symlinked_files`` and we leave them in place. - if not _safe_dest_or_bucket(dst, rel): - continue try: - from .integrations._file_changes import after_file_change, before_file_change - - before_file_change(dst, removal=True) - dst.unlink() - after_file_change(dst) + files.delete(dst, observer=observe_file_change) except OSError as exc: console.print(f"[yellow]⚠[/yellow] could not remove stale {rel}: {exc}") continue diff --git a/tests/integrations/test_cli.py b/tests/integrations/test_cli.py index ed6864253b..7d160b1685 100644 --- a/tests/integrations/test_cli.py +++ b/tests/integrations/test_cli.py @@ -740,16 +740,10 @@ def test_shared_infra_stale_cleanup_ignores_unsafe_manifest_keys(self, tmp_path) assert victim.read_bytes() == victim_bytes def test_shared_infra_stale_cleanup_skips_escaping_key_without_failing( - self, tmp_path, monkeypatch + self, tmp_path ): - """A key that passes the lexical guard but escapes containment — e.g. a - Windows drive-relative ``C:tmp`` that is not ``is_absolute()`` yet discards - the project root when joined — is skipped via ``_validate_rel_path``, never - unlinked, and never turned into an install-time hard failure (#3076 review - round 4). Simulated portably by forcing ``_validate_rel_path`` to reject the - managed key, since real drive-relative paths only escape on Windows.""" + """Corrupt stale keys are rejected lexically before accessing files.""" from specify_cli import _install_shared_infra - from specify_cli.integrations import manifest as manifest_mod from specify_cli.integrations.manifest import IntegrationManifest project = tmp_path / "escaping-key" @@ -759,24 +753,13 @@ def test_shared_infra_stale_cleanup_skips_escaping_key_without_failing( scripts_dir.mkdir(parents=True) # A managed stale orphan that would normally be removed. - stale_rel = ".specify/scripts/bash/update-agent-context.sh" stale = scripts_dir / "update-agent-context.sh" stale.write_text("# legacy orphan\n", encoding="utf-8") manifest = IntegrationManifest("speckit", project, version="test") - manifest.record_existing(stale_rel) + import hashlib + manifest._files[".specify/scripts/bash/../bash/update-agent-context.sh"] = hashlib.sha256(stale.read_bytes()).hexdigest() manifest.save() - # Force the containment check to reject this key, as it would for a - # drive-relative escape on Windows. The cleanup must skip it gracefully. - real_validate = manifest_mod._validate_rel_path - - def fake_validate(rel, root): - if str(rel).endswith("update-agent-context.sh"): - raise ValueError("simulated drive-relative escape") - return real_validate(rel, root) - - monkeypatch.setattr(manifest_mod, "_validate_rel_path", fake_validate) - # Must not raise (no install-time hard failure from a corrupted key). _install_shared_infra(project, "sh", force=False) @@ -844,8 +827,8 @@ def test_shared_infra_warns_when_manifest_cannot_be_decoded(self, tmp_path, caps assert "A new shared manifest will be created" in captured.out @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") - def test_shared_infra_buckets_symlinked_script_destination(self, tmp_path, capsys): - """Symlinked script destinations are bucketed with a warning; the symlink target is preserved.""" + def test_shared_infra_rejects_symlinked_script_destination(self, tmp_path): + """Denied script links abort without changing their targets.""" from specify_cli import _install_shared_infra project = tmp_path / "symlink-script-test" @@ -858,15 +841,14 @@ def test_shared_infra_buckets_symlinked_script_destination(self, tmp_path, capsy scripts_dir.mkdir(parents=True) os.symlink(outside, scripts_dir / "common.sh") - _install_shared_infra(project, "sh", force=True) + with pytest.raises(ValueError, match="Refusing symlink"): + _install_shared_infra(project, "sh", force=True) - captured = capsys.readouterr() - assert "symlinked shared infrastructure" in captured.out assert outside.read_text(encoding="utf-8") == "# outside\n" @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") - def test_shared_infra_buckets_symlinked_template_destination(self, tmp_path, capsys): - """Symlinked template destinations are bucketed with a warning; the symlink target is preserved.""" + def test_shared_infra_rejects_symlinked_template_destination(self, tmp_path): + """Denied template links abort without changing their targets.""" from specify_cli import _install_shared_infra project = tmp_path / "symlink-template-test" @@ -879,10 +861,9 @@ def test_shared_infra_buckets_symlinked_template_destination(self, tmp_path, cap templates_dir.mkdir(parents=True) os.symlink(outside, templates_dir / "plan-template.md") - _install_shared_infra(project, "sh", force=True) + with pytest.raises(ValueError, match="Refusing symlink"): + _install_shared_infra(project, "sh", force=True) - captured = capsys.readouterr() - assert "symlinked shared infrastructure" in captured.out assert outside.read_text(encoding="utf-8") == "# outside\n" @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") @@ -900,7 +881,7 @@ def test_shared_template_refresh_refuses_symlinked_destination(self, tmp_path): templates_dir.mkdir(parents=True) os.symlink(outside, templates_dir / "plan-template.md") - with pytest.raises(ValueError, match="Refusing to overwrite symlinked"): + with pytest.raises(ValueError, match="Refusing symlink"): _refresh_shared_templates(project, invoke_separator=".", force=True) assert outside.read_text(encoding="utf-8") == "# outside\n" @@ -943,7 +924,7 @@ def test_shared_infra_refuses_symlinked_shared_manifest(self, tmp_path): templates_src.mkdir(parents=True) (templates_src / "plan-template.md").write_text("# plan\n", encoding="utf-8") - with pytest.raises(ValueError, match="symlinked integration manifest"): + with pytest.raises(ValueError, match="Refusing symlink"): install_shared_infra( project, "sh", @@ -978,7 +959,7 @@ def test_shared_template_refresh_preflights_before_writing(self, tmp_path): outside.write_text("# outside\n", encoding="utf-8") os.symlink(outside, templates_dir / "z-template.md") - with pytest.raises(ValueError, match="Refusing to overwrite symlinked"): + with pytest.raises(ValueError, match="Refusing symlink"): refresh_shared_templates( project, version="test", @@ -993,8 +974,8 @@ def test_shared_template_refresh_preflights_before_writing(self, tmp_path): assert outside.read_text(encoding="utf-8") == "# outside\n" @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") - def test_shared_infra_install_buckets_unsafe_destinations_and_continues(self, tmp_path): - """Symlinked destinations are bucketed with a warning; safe destinations in the same install still complete.""" + def test_shared_infra_install_rejects_unsafe_destinations_before_writes(self, tmp_path): + """Denied links abort planning before regular files are overwritten.""" from specify_cli.shared_infra import install_shared_infra project = tmp_path / "preflight-install-test" @@ -1014,19 +995,19 @@ def test_shared_infra_install_buckets_unsafe_destinations_and_continues(self, tm outside.write_text("# outside\n", encoding="utf-8") os.symlink(outside, scripts_dir / "z.sh") - install_shared_infra( - project, - "sh", - version="test", - core_pack=core_pack, - repo_root=tmp_path / "unused", - console=_NoopConsole(), - force=True, - ) + with pytest.raises(ValueError, match="Refusing symlink"): + install_shared_infra( + project, + "sh", + version="test", + core_pack=core_pack, + repo_root=tmp_path / "unused", + console=_NoopConsole(), + force=True, + ) - # Symlinked z.sh is preserved (bucketed); regular a.sh is overwritten. assert outside.read_text(encoding="utf-8") == "# outside\n" - assert existing.read_text(encoding="utf-8") == "# new a\n" + assert existing.read_text(encoding="utf-8") == "# old a\n" def test_shared_infra_install_supports_nested_script_sources(self, tmp_path): """Nested script source files create safe destination parents at write time.""" diff --git a/tests/specify_cli/integrations/test_command_switch.py b/tests/specify_cli/integrations/test_command_switch.py index b1824102aa..86561bf632 100644 --- a/tests/specify_cli/integrations/test_command_switch.py +++ b/tests/specify_cli/integrations/test_command_switch.py @@ -638,7 +638,7 @@ def test_switch_preserves_recovered_files(self, tmp_path): # Recovered file must NOT be overwritten — team content preserved. assert shared_script.read_bytes() == custom_bytes - def test_switch_skips_symlinked_parent_directory(self, tmp_path): + def test_switch_rejects_symlinked_parent_directory(self, tmp_path): """Regression: if .specify/scripts/bash is a symlink, switch must not write through it. Copilot follow-up on #2375: leaf-only symlink check let writes escape @@ -668,8 +668,8 @@ def test_switch_skips_symlinked_parent_directory(self, tmp_path): ], catch_exceptions=False) finally: os.chdir(old_cwd) - assert result.exit_code == 0 - # Symlinked tree reported, not written through. + assert result.exit_code == 1 + # Denied parents abort instead of warning and succeeding with skips. assert "symlink" in result.output.lower() # Outside dir contents unchanged. assert (outside / "common.sh").read_bytes() == sentinel diff --git a/tests/specify_cli/integrations/test_installed_adapters.py b/tests/specify_cli/integrations/test_installed_adapters.py index 43ff50634b..85ab6a9200 100644 --- a/tests/specify_cli/integrations/test_installed_adapters.py +++ b/tests/specify_cli/integrations/test_installed_adapters.py @@ -188,6 +188,121 @@ def snapshot(project): } +@pytest.mark.parametrize("operation", ["init", "upgrade"]) +@pytest.mark.parametrize("kind", ["leaf", "gitignore", "scripts-root", "templates-root", "nested-parent"]) +def test_shared_helper_allowed_links_restore_in_public_external_lifecycle( + tmp_path, server, monkeypatch, operation, kind, +): + import specify_cli + from specify_cli.integrations import installer + + publish(server) + project = catalog_project(tmp_path, server) + result = run(project, ["integration", "install", KEY, "--trust-integration", "--script", "sh"]) + assert result.exit_code == 0, result.output + options_path = project / ".specify/init-options.json" + options = json.loads(options_path.read_text()) if options_path.exists() else {} + options["allow_symlinks"] = operation != "init" + options_path.write_text(json.dumps(options)) + target = project / "user-target" + target.mkdir() + (target / "unrelated").write_text("user content\n") + if kind in {"leaf", "gitignore"}: + link = project / (".specify/.gitignore" if kind == "gitignore" else ".specify/scripts/bash/common.sh") + target_file = target / link.name + link.rename(target_file) + link.symlink_to(target_file) + else: + relative = { + "scripts-root": ".specify/scripts", "templates-root": ".specify/templates", + "nested-parent": ".specify/scripts/bash", + }[kind] + link = project / relative + moved = target / "managed" + link.rename(moved) + link.symlink_to(moved, target_is_directory=True) + target_file = moved / ( + "bash/common.sh" if kind == "scripts-root" else + "plan-template.md" if kind == "templates-root" else "common.sh" + ) + target_file.chmod(0o600) + core = tmp_path / "replacement-core" + nested = core / "scripts/bash/new/nested" + nested.mkdir(parents=True) + (nested / "added.sh").write_text("# new nested script\n") + (core / "scripts/bash/common.sh").write_text("# new common\n") + templates = core / "templates" + templates.mkdir() + (templates / "plan-template.md").write_text("# new plan\n") + monkeypatch.setattr(specify_cli, "_locate_core_pack", lambda: core) + before = snapshot(project) + directories = {path.relative_to(project) for path in project.rglob("*") if path.is_dir()} + link_identity = os.readlink(link) + + def late_failure(*args, **kwargs): + if kind == "gitignore": + from specify_cli.shared_infra import SPECIFY_GITIGNORE_CONTENT + assert target_file.read_text() == SPECIFY_GITIGNORE_CONTENT + else: + assert target_file.read_text() == ("# new plan\n" if kind == "templates-root" else "# new common\n") + raise OSError("sample late shared failure") + + if operation == "init": + monkeypatch.setattr(specify_cli, "save_init_options", late_failure) + arguments = [ + "init", "--here", "--force", "--integration", KEY, + "--ignore-agent-tools", "--script", "sh", "--allow-symlinks", + ] + else: + publish(server, version="2.0.0") + monkeypatch.setattr(installer, "persist_package", late_failure) + arguments = ["integration", "upgrade", KEY, "--trust-integration", "--force", "--script", "sh"] + result = run(project, arguments) + + assert result.exit_code == 1, result.output + assert "sample late shared failure" in " ".join(result.output.split()) + assert "Recovery snapshots retained" not in result.output + assert snapshot(project) == before + assert {path.relative_to(project) for path in project.rglob("*") if path.is_dir()} == directories + assert link.is_symlink() and os.readlink(link) == link_identity + if os.name != "nt": + assert target_file.stat().st_mode & 0o777 == 0o600 + + +@pytest.mark.parametrize("kind", ["leaf", "gitignore", "scope", "parent"]) +def test_external_reinit_current_deny_flag_overrides_stored_allow(tmp_path, server, kind): + publish(server) + project = catalog_project(tmp_path, server) + result = run(project, ["integration", "install", KEY, "--trust-integration", "--script", "sh"]) + assert result.exit_code == 0, result.output + options_path = project / ".specify/init-options.json" + options = json.loads(options_path.read_text()) if options_path.exists() else {} + options["allow_symlinks"] = True + options_path.write_text(json.dumps(options)) + relative = { + "leaf": ".specify/scripts/bash/common.sh", + "gitignore": ".specify/.gitignore", + "scope": ".specify/scripts", + "parent": ".specify/scripts/bash", + }[kind] + link = project / relative + target = project / "target" + link.rename(target) + link.symlink_to(target, target_is_directory=kind not in {"leaf", "gitignore"}) + before = snapshot(project) + + result = run(project, [ + "init", "--here", "--force", "--integration", KEY, + "--ignore-agent-tools", "--script", "sh", + ]) + + assert result.exit_code == 1, result.output + assert "Refusing symlink" in result.output + assert snapshot(project) == before + assert link.is_symlink() + assert json.loads(options_path.read_text())["allow_symlinks"] is True + + @pytest.mark.parametrize("existing", [False, True]) @pytest.mark.parametrize("change", ["unchanged", "partial", "concurrent", "deleted"]) def test_round9_pending_write_preserves_unattributed_changes(tmp_path, server, monkeypatch, existing, change): diff --git a/tests/specify_cli/test_shared_infra.py b/tests/specify_cli/test_shared_infra.py new file mode 100644 index 0000000000..4d4466b1a4 --- /dev/null +++ b/tests/specify_cli/test_shared_infra.py @@ -0,0 +1,587 @@ +"""Scoped project operations owned by shared infrastructure.""" + +import errno +import hashlib +import json +import os +import shutil + +import pytest + +from specify_cli.file_helper import ( + FileHelper, + PathEscapeError, + SymlinkDeniedError, + UnsupportedPathError, +) +from specify_cli.integrations.manifest import IntegrationManifest +from specify_cli.shared_infra import install_shared_infra, refresh_shared_templates + + +class Console: + def print(self, *args, **kwargs): + pass + + +@pytest.fixture +def shared_project(tmp_path): + project = tmp_path / "project" + project.mkdir() + core = tmp_path / "core" + scripts = core / "scripts" / "bash" + scripts.mkdir(parents=True) + (scripts / "common.sh").write_text("# bundled\n") + (scripts / "common.sh").chmod(0o755) + templates = core / "templates" + templates.mkdir() + (templates / "plan-template.md").write_text("# bundled plan\n") + return project, core + + +def install(project, core, **kwargs): + return install_shared_infra( + project, "sh", version="test", core_pack=core, repo_root=core, + console=Console(), **kwargs, + ) + + +@pytest.mark.parametrize("force", [False, True]) +@pytest.mark.parametrize("relative", [ + ".specify/scripts/bash/common.sh", + ".specify/templates/plan-template.md", + ".specify/.gitignore", +]) +def test_deny_linked_destination_errors_instead_of_skipping(shared_project, relative, force): + project, core = shared_project + destination = project / relative + destination.parent.mkdir(parents=True) + target = project / "custom" + target.write_text("custom\n") + destination.symlink_to(target) + + with pytest.raises(SymlinkDeniedError): + install(project, core, force=force) + + assert destination.is_symlink() + assert target.read_text() == "custom\n" + assert not (project / ".specify/integrations/speckit.manifest.json").exists() + + +def allow(project): + (project / ".specify").mkdir(exist_ok=True) + (project / ".specify/init-options.json").write_text('{"allow_symlinks": true}\n') + return FileHelper(project, allow_symlinks=True) + + +@pytest.mark.parametrize("relative,kind", [ + (".specify/scripts/bash/common.sh", "leaf"), + (".specify/scripts/bash/common.sh", "parent"), + (".specify/templates/plan-template.md", "leaf"), + (".specify/templates/plan-template.md", "parent"), + (".specify/.gitignore", "leaf"), +]) +@pytest.mark.parametrize("mode", ["missing-only", "refresh", "customized", "recovered", "force"]) +def test_allowed_links_preserve_overwrite_and_manifest_decisions(shared_project, relative, kind, mode): + project, core = shared_project + files = allow(project) + install(project, core, files=files) + destination = project / relative + if kind == "parent": + target_dir = project / "user-target" + destination.parent.rename(target_dir) + destination.parent.symlink_to(target_dir, target_is_directory=True) + target = target_dir / destination.name + link = destination.parent + else: + target = project / "user-target" + destination.rename(target) + destination.symlink_to(target) + link = destination + original = target.read_text() + if mode in {"customized", "recovered", "force"}: + target.write_text("user customization\n") + original = target.read_text() + if mode == "recovered": + (project / ".specify/integrations/speckit.manifest.json").unlink() + install(project, core, files=files) + if relative.endswith("common.sh"): + (core / "scripts/bash/common.sh").write_text("# replacement\n") + replacement = "# replacement\n" + elif relative.endswith("plan-template.md"): + (core / "templates/plan-template.md").write_text("# replacement plan\n") + replacement = "# replacement plan\n" + else: + from specify_cli.shared_infra import SPECIFY_GITIGNORE_CONTENT + replacement = SPECIFY_GITIGNORE_CONTENT + + install( + project, core, files=files, force=mode == "force", + refresh_managed=mode in {"refresh", "customized", "recovered"}, + ) + assert link.is_symlink() + expected = replacement if mode in {"refresh", "force"} else original + assert target.read_text() == expected + manifest = IntegrationManifest.load("speckit", project, files=files) + assert relative in manifest.files + assert all(not key.startswith("user-target") for key in manifest.files) + if mode != "customized": + assert manifest.files[relative] == hashlib.sha256(target.read_bytes()).hexdigest() + assert manifest.is_recovered(relative) == (mode == "recovered") + if relative.endswith("common.sh") and mode in {"refresh", "force"} and os.name != "nt": + assert target.stat().st_mode & 0o777 == 0o755 + + +@pytest.mark.parametrize("kind", ["leaf", "parent"]) +@pytest.mark.parametrize("mode", ["managed", "customized", "recovered", "force"]) +def test_template_only_refresh_preserves_allowed_link_and_ownership(shared_project, kind, mode): + project, core = shared_project + files = allow(project) + install(project, core, files=files) + relative = ".specify/templates/plan-template.md" + destination = project / relative + target = project / "target" + if kind == "leaf": + destination.rename(target) + destination.symlink_to(target) + link = destination + else: + destination.parent.rename(target) + destination.parent.symlink_to(target, target_is_directory=True) + target /= destination.name + link = destination.parent + if mode in {"customized", "recovered", "force"}: + target.write_text("user plan\n") + if mode == "recovered": + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative, recovered=True) + manifest.save() + original = target.read_text() + (core / "templates/plan-template.md").write_text("replacement plan\n") + refresh_shared_templates( + project, version="test", core_pack=core, repo_root=core, console=Console(), + invoke_separator=".", files=files, force=mode == "force", + ) + assert link.is_symlink() + assert target.read_text() == ("replacement plan\n" if mode in {"managed", "force"} else original) + manifest = IntegrationManifest.load("speckit", project, files=files) + assert manifest.is_recovered(relative) == (mode == "recovered") + + +def test_wrapper_reads_policy_afresh_between_invocations(shared_project, monkeypatch): + import specify_cli + project, core = shared_project + monkeypatch.setattr(specify_cli, "_locate_core_pack", lambda: core) + files = allow(project) + install(project, core, files=files) + destination = project / ".specify/templates/plan-template.md" + target = project / "target" + destination.rename(target) + destination.symlink_to(target) + options = project / ".specify/init-options.json" + options.write_text('{"allow_symlinks": false}') + with pytest.raises(SymlinkDeniedError): + specify_cli._install_shared_infra(project, "sh", force=True) + options.write_text('{"allow_symlinks": true}') + specify_cli._install_shared_infra(project, "sh", force=True) + options.write_text('{"allow_symlinks": false}') + with pytest.raises(SymlinkDeniedError): + specify_cli._refresh_shared_templates(project, invoke_separator=".", force=True) + assert destination.is_symlink() + + +def test_init_current_allow_flag_updates_contained_link_and_persists_bool(shared_project, monkeypatch): + from typer.testing import CliRunner + + import specify_cli + project, core = shared_project + monkeypatch.setattr(specify_cli, "_locate_core_pack", lambda: core) + (project / ".specify/scripts/bash").mkdir(parents=True) + (project / ".specify/init-options.json").write_text('{"allow_symlinks": false, "unrelated": "keep"}') + target = project / "user-script" + target.write_text("# user\n") + destination = project / ".specify/scripts/bash/common.sh" + destination.symlink_to(target) + result = CliRunner().invoke(specify_cli.app, [ + "init", str(project), "--force", "--integration", "copilot", + "--script", "sh", "--ignore-agent-tools", "--allow-symlinks", + ], catch_exceptions=False) + assert result.exit_code == 0, result.output + assert destination.is_symlink() + assert target.read_text() == "# bundled\n" + options = json.loads((project / ".specify/init-options.json").read_text()) + assert options["allow_symlinks"] is True + assert options["unrelated"] == "keep" + + +@pytest.mark.parametrize("allow_links", [False, True]) +def test_trusted_root_alias_ignores_unrelated_links(shared_project, tmp_path, allow_links): + project, core = shared_project + root_alias = tmp_path / "root-alias" + root_alias.symlink_to(project, target_is_directory=True) + (project / "unrelated").symlink_to(tmp_path / "outside-missing") + install(root_alias, core, files=FileHelper(root_alias, allow_symlinks=allow_links)) + assert (project / ".specify/scripts/bash/common.sh").read_text() == "# bundled\n" + assert (project / "unrelated").is_symlink() + + +@pytest.mark.parametrize("operation", ["install", "refresh"]) +@pytest.mark.parametrize("kind", ["leaf", "parent"]) +@pytest.mark.parametrize("target_kind", ["external", "dangling", "cycle"]) +def test_allow_invalid_destinations_fail_before_writes(shared_project, tmp_path, operation, kind, target_kind): + project, core = shared_project + files = allow(project) + destination = project / ".specify/templates" + if kind == "leaf": + destination.mkdir() + destination /= "plan-template.md" + if target_kind == "external": + target = tmp_path / "outside" + if kind == "parent": + target.mkdir() + (target / "keep").write_text("outside\n") + else: + target.write_text("outside\n") + error = PathEscapeError + elif target_kind == "dangling": + target = project / "missing" + error = FileNotFoundError + else: + target = destination + error = UnsupportedPathError + destination.symlink_to(target, target_is_directory=kind == "parent") + + with pytest.raises(error): + if operation == "install": + install(project, core, files=files, force=True) + else: + refresh_shared_templates( + project, version="test", core_pack=core, repo_root=core, + console=Console(), invoke_separator=".", files=files, force=True, + ) + assert destination.is_symlink() + assert not (project / ".specify/integrations/speckit.manifest.json").exists() + if target_kind == "external": + assert (target / "keep" if kind == "parent" else target).read_text() == "outside\n" + + +@pytest.mark.parametrize("kind", ["leaf", "parent"]) +@pytest.mark.parametrize("allow_links", [False, True]) +def test_manifest_load_hash_save_obey_explicit_helper(shared_project, kind, allow_links): + project, core = shared_project + files = allow(project) if allow_links else FileHelper(project) + install(project, core, files=files) + manifest_path = project / ".specify/integrations/speckit.manifest.json" + if kind == "leaf": + target = project / "saved-manifest.json" + manifest_path.rename(target) + manifest_path.symlink_to(target) + link = manifest_path + else: + target_dir = project / "saved-manifests" + manifest_path.parent.rename(target_dir) + manifest_path.parent.symlink_to(target_dir, target_is_directory=True) + target = target_dir / manifest_path.name + link = manifest_path.parent + if not allow_links: + before = target.read_bytes() + with pytest.raises(SymlinkDeniedError): + install(project, core, files=files, force=True) + assert target.read_bytes() == before + else: + install(project, core, files=files, force=True) + data = json.loads(target.read_text()) + for relative, digest in data["files"].items(): + assert digest == hashlib.sha256(files.read_bytes(relative)).hexdigest() + assert link.is_symlink() + + +@pytest.mark.parametrize("kind", ["external", "dangling", "cycle"]) +@pytest.mark.parametrize("allow_links", [False, True]) +def test_manifest_policy_failures_never_recover_as_new(shared_project, tmp_path, kind, allow_links): + project, core = shared_project + files = allow(project) if allow_links else FileHelper(project) + directory = project / ".specify/integrations" + directory.mkdir(parents=True, exist_ok=True) + destination = directory / "speckit.manifest.json" + target = project / "missing" + if kind == "external": + target = tmp_path / "outside.json" + target.write_text("{}") + elif kind == "cycle": + target = destination + destination.symlink_to(target) + error = SymlinkDeniedError if not allow_links else { + "external": PathEscapeError, "dangling": FileNotFoundError, "cycle": UnsupportedPathError, + }[kind] + with pytest.raises(error): + install(project, core, files=files, force=True) + assert destination.is_symlink() + assert not (project / ".specify/scripts").exists() + + +@pytest.mark.parametrize("newline", ["\n", "\r\n"]) +def test_scoped_manifest_preserves_native_json_text_format(shared_project, monkeypatch, newline): + project, core = shared_project + monkeypatch.setattr("specify_cli.integrations.manifest.os.linesep", newline) + install(project, core, files=FileHelper(project)) + raw = (project / ".specify/integrations/speckit.manifest.json").read_bytes() + data = json.loads(raw) + assert raw == (json.dumps(data, indent=2) + "\n").replace("\n", newline).encode("utf-8") + + +@pytest.mark.parametrize("phase", ["manifest-load", "managed-hash", "recovered-record"]) +def test_link_becoming_dangling_during_read_is_not_swallowed(shared_project, monkeypatch, phase): + from pathlib import Path + project, core = shared_project + files = allow(project) + install(project, core, files=files) + manifest_path = project / ".specify/integrations/speckit.manifest.json" + destination = manifest_path if phase == "manifest-load" else project / ".specify/scripts/bash/common.sh" + target = project / "target" + destination.rename(target) + destination.symlink_to(target) + if phase == "recovered-record": + manifest_path.unlink() + original = FileHelper.read_bytes + + def read_bytes(helper, path): + absolute = Path(path) if Path(path).is_absolute() else helper.root / path + if absolute == destination: + target.unlink() + return original(helper, path) + + monkeypatch.setattr(FileHelper, "read_bytes", read_bytes) + with pytest.raises(FileNotFoundError): + install(project, core, files=files, refresh_managed=phase == "managed-hash") + assert destination.is_symlink() + if phase == "recovered-record": + assert not manifest_path.exists() + + +@pytest.mark.parametrize("allow_links", [False, True]) +@pytest.mark.parametrize("kind", ["contained", "dangling", "external", "cycle"]) +def test_stale_leaf_links_are_never_hash_promoted_or_deleted(shared_project, tmp_path, allow_links, kind): + project, core = shared_project + files = allow(project) if allow_links else FileHelper(project) + install(project, core, files=files) + relative = ".specify/scripts/bash/stale.sh" + stale = project / relative + stale.write_text("managed stale\n") + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative) + manifest.save() + stale.unlink() + target = project / "stale-target" + if kind == "contained": + target.write_text("managed stale\n") + elif kind == "external": + target = tmp_path / "outside" + target.write_text("managed stale\n") + elif kind == "cycle": + target = stale + stale.symlink_to(target) + if not allow_links: + with pytest.raises(SymlinkDeniedError): + install(project, core, files=files, force=True) + else: + install(project, core, files=files, force=True) + assert relative in IntegrationManifest.load("speckit", project, files=files).files + assert stale.is_symlink() + if kind in {"contained", "external"}: + assert target.read_text() == "managed stale\n" + + +@pytest.mark.parametrize("customized,recovered", [(False, False), (True, False), (False, True)]) +def test_stale_cleanup_through_allowed_parent_obeys_ownership(shared_project, customized, recovered): + project, core = shared_project + files = allow(project) + install(project, core, files=files) + directory = project / ".specify/scripts/bash" + target = project / "script-target" + directory.rename(target) + directory.symlink_to(target, target_is_directory=True) + relative = ".specify/scripts/bash/stale.sh" + stale = target / "stale.sh" + stale.write_text("managed stale\n") + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative, recovered=recovered) + manifest.save() + if customized: + stale.write_text("customized\n") + install(project, core, files=files, force=True) + assert directory.is_symlink() + assert stale.exists() == (customized or recovered) + assert (relative in IntegrationManifest.load("speckit", project, files=files).files) == (customized or recovered) + + +@pytest.mark.parametrize("kind", ["leaf", "templates-root", "scripts-root", "nested-parent"]) +@pytest.mark.parametrize("new_parents", [False, True]) +def test_real_lifecycle_rolls_back_allowed_targets(shared_project, tmp_path, monkeypatch, kind, new_parents): + from specify_cli.integrations._lifecycle import _transaction + + project, core = shared_project + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + files = allow(project) + target = project / "user-target" + target.mkdir() + (target / "unrelated").write_text("keep\n") + if kind == "leaf": + destination = project / ".specify/scripts/bash/common.sh" + target_file = target / "common.sh" + elif kind == "templates-root": + destination = project / ".specify/templates" + target_file = target / "plan-template.md" + elif kind == "scripts-root": + destination = project / ".specify/scripts" + (target / "bash").mkdir() + target_file = target / "bash/common.sh" + else: + destination = project / ".specify/scripts/bash" + target_file = target / "common.sh" + destination.parent.mkdir(parents=True, exist_ok=True) + target_file.write_text("original\n") + target_file.chmod(0o600) + destination.symlink_to(target_file if kind == "leaf" else target, target_is_directory=kind != "leaf") + if new_parents: + nested = core / ("templates" if kind == "templates-root" else "scripts/bash") / "new/nested" + nested.mkdir(parents=True) + (nested / "added.sh").write_text("# new\n") + before = {path.relative_to(target): path.read_bytes() for path in target.rglob("*") if path.is_file()} + directories = {path.relative_to(target) for path in target.rglob("*") if path.is_dir()} + link_identity = os.readlink(destination) + + with ( + pytest.raises(RuntimeError, match="late failure"), + _transaction(project, None, None, {}, shared_files=files), + ): + install(project, core, files=files, force=True) + raise RuntimeError("late failure") + + assert destination.is_symlink() + assert os.readlink(destination) == link_identity + assert {path.relative_to(target): path.read_bytes() for path in target.rglob("*") if path.is_file()} == before + assert {path.relative_to(target) for path in target.rglob("*") if path.is_dir()} == directories + if os.name != "nt": + assert target_file.stat().st_mode & 0o777 == 0o600 + + +def test_directory_rollback_preserves_unrelated_new_contents(shared_project, tmp_path, monkeypatch): + from specify_cli.integrations._lifecycle import _transaction + project, core = shared_project + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + files = allow(project) + target = project / "target" + target.mkdir() + alias = project / ".specify/scripts" + alias.symlink_to(target, target_is_directory=True) + nested = core / "scripts/bash/new/nested" + nested.mkdir(parents=True) + (nested / "owned.sh").write_text("owned\n") + + with ( + pytest.raises(RuntimeError, match="late failure"), + _transaction(project, None, None, {}, shared_files=files), + ): + install(project, core, files=files) + (target / "bash/new/nested/unrelated").write_text("unrelated\n") + raise RuntimeError("late failure") + + assert (target / "bash/new/nested/unrelated").read_text() == "unrelated\n" + assert not (target / "bash/new/nested/owned.sh").exists() + assert not (target / "bash/common.sh").exists() + assert alias.is_symlink() + + +@pytest.mark.parametrize("fail", [False, True]) +def test_directory_observer_context_restores_nested_transactions(shared_project, tmp_path, monkeypatch, fail): + from specify_cli.integrations._file_changes import ( + directory_change_observer, + file_change_observer, + ) + from specify_cli.integrations._lifecycle import _transaction + project, _ = shared_project + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + other = tmp_path / "other-project" + other.mkdir() + def outer_file(*args): + pass + + def outer_directory(*args): + pass + file_token = file_change_observer.set(outer_file) + directory_token = directory_change_observer.set(outer_directory) + try: + with _transaction(project, None, None, {}): + current_file = file_change_observer.get() + current_directory = directory_change_observer.get() + try: + with _transaction(other, None, None, {}): + assert file_change_observer.get() != current_file + assert directory_change_observer.get() != current_directory + if fail: + raise RuntimeError("nested failure") + except RuntimeError as exc: + assert str(exc) == "nested failure" + assert file_change_observer.get() == current_file + assert directory_change_observer.get() == current_directory + assert file_change_observer.get() is outer_file + assert directory_change_observer.get() is outer_directory + finally: + directory_change_observer.reset(directory_token) + file_change_observer.reset(file_token) + + +def test_directory_cleanup_io_failure_reports_rollback_error(shared_project, tmp_path, monkeypatch): + from pathlib import Path + + from specify_cli.integrations import _lifecycle, installer + project, core = shared_project + home = tmp_path / "home" + home.mkdir() + monkeypatch.setenv("HOME", str(home)) + monkeypatch.setenv("USERPROFILE", str(home)) + files = allow(project) + target = project / "target" + target.mkdir() + (project / ".specify/scripts").symlink_to(target, target_is_directory=True) + failed = target / "bash" + original_rmdir = Path.rmdir + original_mkdtemp = _lifecycle.tempfile.mkdtemp + backups = [] + + def record_backup(*args, **kwargs): + directory = original_mkdtemp(*args, **kwargs) + if kwargs.get("prefix") == "speckit-integration-rollback-": + backups.append(Path(directory)) + return directory + + def rmdir(path, *args, **kwargs): + if path == failed: + raise PermissionError(errno.EACCES, "directory cleanup denied", str(path)) + return original_rmdir(path, *args, **kwargs) + + monkeypatch.setattr(_lifecycle.tempfile, "mkdtemp", record_backup) + monkeypatch.setattr(Path, "rmdir", rmdir) + try: + with ( + pytest.raises(installer.IntegrationInstallError, match="rollback failed.*directory cleanup denied") as error, + _lifecycle._transaction(project, None, None, {}, shared_files=files), + ): + install(project, core, files=files) + raise RuntimeError("original failure") + assert "original failure" in str(error.value) + assert isinstance(error.value.__cause__, RuntimeError) + assert failed.is_dir() + assert not (failed / "common.sh").exists() + finally: + for backup in backups: + shutil.rmtree(backup) diff --git a/tests/test_file_helper.py b/tests/test_file_helper.py index a04b9e0a34..7b648a392f 100644 --- a/tests/test_file_helper.py +++ b/tests/test_file_helper.py @@ -3,6 +3,7 @@ from dataclasses import FrozenInstanceError import errno import os +import stat from pathlib import Path, PureWindowsPath import subprocess from types import SimpleNamespace @@ -42,6 +43,194 @@ def create(path, target): return create +@pytest.mark.parametrize("follow", [False, True]) +def test_stat_missing_and_nondirectory_parents(root, follow): + files = FileHelper(root) + assert files.stat("absent/child", follow_symlinks=follow, missing_ok=True) is None + with pytest.raises(FileNotFoundError): + files.stat("absent/child", follow_symlinks=follow) + (root / "file").write_text("existing") + with pytest.raises(NotADirectoryError): + files.stat("file/child", follow_symlinks=follow, missing_ok=True) + assert stat.S_ISREG(files.stat("file", follow_symlinks=follow).st_mode) + + +@pytest.mark.parametrize("kind", ["contained", "dangling", "external", "cycle"]) +@pytest.mark.parametrize("follow", [False, True]) +@pytest.mark.parametrize("allow", [False, True]) +def test_stat_link_targets_never_become_missing_defaults(root, link, tmp_path, kind, follow, allow): + target = root / "target" + if kind == "contained": + target.write_text("content") + elif kind == "external": + target = tmp_path / "outside" + target.write_text("outside") + elif kind == "cycle": + target = root / "alias" + alias = link(root / "alias", target) + files = FileHelper(root, allow_symlinks=allow) + if not allow: + with pytest.raises(SymlinkDeniedError): + files.stat(alias, follow_symlinks=follow, missing_ok=True) + elif not follow: + assert stat.S_ISLNK(files.stat(alias, follow_symlinks=False, missing_ok=True).st_mode) + elif kind == "contained": + assert stat.S_ISREG(files.stat(alias, missing_ok=True).st_mode) + else: + error = {"dangling": FileNotFoundError, "external": PathEscapeError, "cycle": UnsupportedPathError}[kind] + with pytest.raises(error): + files.stat(alias, missing_ok=True) + + +@pytest.mark.parametrize("allow", [False, True]) +@pytest.mark.parametrize("follow", [False, True]) +def test_stat_rejects_unsupported_reparse_entries(root, monkeypatch, allow, follow): + path = root / "junction" + path.mkdir() + original = Path.lstat + + def lstat(entry, *args, **kwargs): + result = original(entry, *args, **kwargs) + if entry == path: + return SimpleNamespace(st_mode=result.st_mode, st_file_attributes=stat.FILE_ATTRIBUTE_REPARSE_POINT) + return result + + monkeypatch.setattr(Path, "lstat", lstat) + with pytest.raises(UnsupportedPathError): + FileHelper(root, allow_symlinks=allow).stat(path, follow_symlinks=follow) + + +def test_stat_permission_errors_propagate(root, monkeypatch): + path = root / "file" + path.write_text("existing") + original = Path.stat + + def denied(entry, *args, **kwargs): + if entry == path: + raise PermissionError("metadata denied") + return original(entry, *args, **kwargs) + + monkeypatch.setattr(Path, "stat", denied) + with pytest.raises(PermissionError, match="metadata denied"): + FileHelper(root).stat(path, missing_ok=True) + + +@pytest.mark.parametrize("failure", ["none", "before", "replace", "after"]) +def test_atomic_write_observes_actual_target_and_cleans_temporary(root, link, monkeypatch, failure): + target = root / "target" + target.write_text("old") + alias = link(root / "alias", target) + events = [] + + def observer(path, before, removal): + events.append((path, before, removal)) + assert path == target + assert path.read_text() == ("old" if before else "new") + if failure == ("before" if before else "after"): + raise RuntimeError("observer failure") + + if failure == "replace": + def fail_replace(*args): + raise RuntimeError("replace failure") + monkeypatch.setattr(os, "replace", fail_replace) + files = FileHelper(root, allow_symlinks=True) + if failure == "none": + files.write_text(alias, "new", observer=observer) + else: + with pytest.raises(RuntimeError): + files.write_text(alias, "new", observer=observer) + assert alias.is_symlink() + assert target.read_text() == ("old" if failure in {"before", "replace"} else "new") + assert events == [(target, True, False)] + ( + [(target, False, False)] if failure in {"none", "after"} else [] + ) + assert sorted(path.name for path in root.iterdir()) == ["alias", "target"] + + +def test_mkdir_observes_only_new_actual_directories_in_order(root, link): + target = root / "target" + target.mkdir() + alias = link(root / "alias", target) + events = [] + files = FileHelper(root, allow_symlinks=True) + def observer(*event): + events.append(event) + files.mkdir(alias / "new/nested", parents=True, observer=observer) + files.mkdir(alias, exist_ok=True, observer=observer) + assert events == [ + (target / "new", True, False), (target / "new", False, False), + (target / "new/nested", True, False), (target / "new/nested", False, False), + ] + with pytest.raises(FileExistsError): + files.mkdir(alias, observer=observer) + with pytest.raises(FileNotFoundError): + files.mkdir("missing/child", observer=observer) + assert len(events) == 4 + + +def test_delete_observes_leaf_link_not_external_target(root, link, tmp_path): + target = tmp_path / "outside" + target.write_text("keep") + alias = link(root / "alias", target) + events = [] + FileHelper(root, allow_symlinks=True).delete(alias, observer=lambda *event: events.append(event)) + assert events == [(alias, True, True), (alias, False, False)] + assert target.read_text() == "keep" + assert not alias.is_symlink() + + +@pytest.mark.parametrize("operation", ["mkdir", "delete"]) +@pytest.mark.parametrize("failure", ["before", "after"]) +def test_directory_and_delete_callback_failures_propagate(root, link, operation, failure): + target = root / "target" + target.mkdir() + alias = link(root / "alias", target) + path = alias / "entry" + if operation == "delete": + (target / "entry").write_text("old") + events = [] + + def observer(entry, before, removal): + assert entry == target / "entry" + assert removal == (before and operation == "delete") + events.append(before) + if failure == ("before" if before else "after"): + raise RuntimeError("observer failure") + + with pytest.raises(RuntimeError, match="observer failure"): + getattr(FileHelper(root, allow_symlinks=True), operation)(path, observer=observer) + assert events == ([True] if failure == "before" else [True, False]) + assert (target / "entry").exists() == ( + (operation == "delete" and failure == "before") or (operation == "mkdir" and failure == "after") + ) + assert alias.is_symlink() + + +def test_recursive_delete_observation_follows_preflight_and_mutation_order(root): + tree = root / "tree" + (tree / "nested").mkdir(parents=True) + (tree / "nested/file").write_text("old") + events = [] + FileHelper(root).delete(tree, recursive=True, observer=lambda *event: events.append(event)) + assert events == [ + (tree / "nested/file", True, True), (tree / "nested/file", False, False), + (tree / "nested", True, True), (tree / "nested", False, False), + (tree, True, True), (tree, False, False), + ] + + +@pytest.mark.parametrize("operation", ["write_text", "mkdir", "delete"]) +def test_denied_mutations_never_call_observer(root, link, operation): + target = root / "target" + target.write_text("keep") + alias = link(root / "alias", target) + events = [] + args = (alias, "new") if operation == "write_text" else (alias,) + with pytest.raises(SymlinkDeniedError): + getattr(FileHelper(root), operation)(*args, observer=lambda *event: events.append(event)) + assert events == [] + + @pytest.mark.parametrize("allow", [False, True]) def test_plain_operations(root, allow): files = FileHelper(root, allow_symlinks=allow) @@ -164,7 +353,7 @@ def test_overwrite_text_rejects_invalid_destination(root, path, allow): assert (root / "directory").is_dir() -@pytest.mark.parametrize("operation", ["read_bytes", "write_bytes", "create_bytes", "mkdir", "delete"]) +@pytest.mark.parametrize("operation", ["read_bytes", "write_bytes", "create_bytes", "mkdir", "delete", "stat"]) @pytest.mark.parametrize("kind", ["leaf", "parent", "dangling"]) def test_deny_rejects_accessed_links_without_mutation(root, link, operation, kind): target = root / "target" @@ -269,7 +458,7 @@ def test_multi_link_cycle_and_excessive_chain_fail_explicitly(root, link): files.read_bytes("chain-0") -@pytest.mark.parametrize("operation", ["read_bytes", "write_bytes", "mkdir", "create_bytes", "delete"]) +@pytest.mark.parametrize("operation", ["read_bytes", "write_bytes", "mkdir", "create_bytes", "delete", "stat"]) def test_external_parent_is_never_traversed(root, link, operation): outside = root.parent / "outside" outside.mkdir() @@ -898,7 +1087,7 @@ def extended(path): @pytest.mark.skipif(os.name != "nt", reason="Native Windows path semantics required") @pytest.mark.parametrize("spelling", ["/outside/file", r"\outside\file", "C:outside/file", "C:"]) -@pytest.mark.parametrize("operation", ["read_bytes", "create_bytes", "write_bytes", "mkdir", "delete"]) +@pytest.mark.parametrize("operation", ["read_bytes", "create_bytes", "write_bytes", "mkdir", "delete", "stat"]) @pytest.mark.parametrize("allow", [False, True]) def test_native_windows_anchored_relative_operations_rejected(root, spelling, operation, allow): args = (spelling, b"bad") if operation in ("create_bytes", "write_bytes") else (spelling,) @@ -1037,6 +1226,8 @@ def test_native_windows_junctions_never_traversed_or_deleted(root, junction, all files = FileHelper(root, allow_symlinks=allow) for operation in ( lambda: files.read_text(alias / "keep"), + lambda: files.stat(alias), + lambda: files.stat(alias, follow_symlinks=False), lambda: files.write_text(alias / "keep", "bad"), lambda: files.create_text(alias / "new", "bad"), lambda: files.mkdir(alias / "new-directory"), From bfe131196ac75a429a9a6c2fc9f8c18ec9d52b68 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Tue, 13 Oct 2026 13:31:35 -0500 Subject: [PATCH 2/5] fix: limit shared template checks to selected destinations Restrict manifest modification queries to destinations selected by the refresh caller. Preserve default queries and unrelated tracked entries without expanding FileHelper or lifecycle responsibilities. Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 084bd222-8ae8-474a-9dcc-6e6c44e031f4 --- design/file-helper.md | 5 ++ src/specify_cli/integrations/manifest.py | 6 +- src/specify_cli/shared_infra.py | 19 +++-- tests/integrations/test_manifest.py | 17 ++++ tests/specify_cli/test_shared_infra.py | 100 +++++++++++++++++++++++ 5 files changed, 139 insertions(+), 8 deletions(-) diff --git a/design/file-helper.md b/design/file-helper.md index 879b7e57a3..914bcccfef 100644 --- a/design/file-helper.md +++ b/design/file-helper.md @@ -205,6 +205,11 @@ entries, not reject a whole project for excluded/unrelated links. ## Shared infrastructure ownership +FileHelper owns symlink-aware filesystem CRUD and metadata access, not project +configuration, file ownership, overwrite decisions or transaction/rollback +policy. Callers select the paths they actually need to access; the helper +enforces symlink policy on those operations, not on unrelated project entries. + `install_shared_infra` and `refresh_shared_templates` receive an explicit `files: FileHelper` from init and integration lifecycle callers. Compatibility callers that omit it resolve the selected project's current policy afresh. diff --git a/src/specify_cli/integrations/manifest.py b/src/specify_cli/integrations/manifest.py index 9d3b93667b..11c30f7b88 100644 --- a/src/specify_cli/integrations/manifest.py +++ b/src/specify_cli/integrations/manifest.py @@ -371,10 +371,12 @@ def is_recovered(self, rel_path: str | Path) -> bool: return False return normalized in self._recovered_files - def check_modified(self) -> list[str]: - """Return relative paths of tracked files whose content changed on disk.""" + def check_modified(self, *, paths: set[str] | None = None) -> list[str]: + """Check tracked content, optionally restricted to selected lexical keys.""" modified: list[str] = [] for rel, expected_hash in self._files.items(): + if paths is not None and rel not in paths: + continue rel_path = Path(rel) # Skip paths that are absolute or attempt to escape the project root if rel_path.is_absolute() or ".." in rel_path.parts: diff --git a/src/specify_cli/shared_infra.py b/src/specify_cli/shared_infra.py index af79f43474..aea837d33a 100644 --- a/src/specify_cli/shared_infra.py +++ b/src/specify_cli/shared_infra.py @@ -426,7 +426,11 @@ def refresh_shared_templates( force: bool = False, files: FileHelper | None = None, ) -> None: - """Refresh default-sensitive shared templates without touching scripts.""" + """Refresh selected bundled templates without inspecting other tracked files. + + Modification checks cover only the destinations this refresh produces; + unrelated manifest entries retain their hashes and ownership records. + """ project_path = Path(project_path).absolute() templates_src = shared_templates_source(core_pack=core_pack, repo_root=repo_root) if not templates_src.is_dir(): @@ -435,16 +439,19 @@ def refresh_shared_templates( files = files if files is not None else project_files(project_path) manifest = load_speckit_manifest(project_path, version=version, console=console, files=files) tracked_files = manifest.files - modified = set(manifest.check_modified()) + templates = [ + src for src in templates_src.iterdir() + if src.is_file() and src.name != "vscode-settings.json" and not src.name.startswith(".") + ] + modified = set(manifest.check_modified( + paths={f".specify/templates/{src.name}" for src in templates}, + )) skipped_files: list[str] = [] planned_updates: list[tuple[Path, str, str]] = [] dest_templates = project_path / ".specify" / "templates" files.mkdir(dest_templates, parents=True, exist_ok=True, observer=observe_directory_change) - for src in templates_src.iterdir(): - if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."): - continue - + for src in templates: dst = dest_templates / src.name entry = files.stat(dst, missing_ok=True) rel = dst.relative_to(project_path).as_posix() diff --git a/tests/integrations/test_manifest.py b/tests/integrations/test_manifest.py index 19f6c08ac6..260c48d7b7 100644 --- a/tests/integrations/test_manifest.py +++ b/tests/integrations/test_manifest.py @@ -223,6 +223,23 @@ def test_remove_rejects_parent_traversal(self, tmp_path): class TestManifestCheckModified: + @pytest.mark.parametrize("use_helper", [False, True]) + def test_selected_paths_leave_default_query_unchanged(self, tmp_path, use_helper): + from specify_cli.file_helper import FileHelper + + m = IntegrationManifest( + "test", tmp_path, files=FileHelper(tmp_path) if use_helper else None, + ) + for name in ("selected.txt", "unrelated.txt"): + path = tmp_path / name + path.write_text("original", encoding="utf-8") + m.record_existing(name) + path.write_text("changed", encoding="utf-8") + + assert m.check_modified(paths=set()) == [] + assert m.check_modified(paths={"selected.txt", "untracked.txt"}) == ["selected.txt"] + assert m.check_modified() == ["selected.txt", "unrelated.txt"] + def test_unmodified_file(self, tmp_path): m = IntegrationManifest("test", tmp_path) m.record_file("f.txt", "original") diff --git a/tests/specify_cli/test_shared_infra.py b/tests/specify_cli/test_shared_infra.py index 4d4466b1a4..25fed612bb 100644 --- a/tests/specify_cli/test_shared_infra.py +++ b/tests/specify_cli/test_shared_infra.py @@ -167,6 +167,106 @@ def test_template_only_refresh_preserves_allowed_link_and_ownership(shared_proje assert manifest.is_recovered(relative) == (mode == "recovered") +@pytest.mark.parametrize("allow_links", [False, True]) +@pytest.mark.parametrize("force", [False, True]) +@pytest.mark.parametrize("relative", [ + ".specify/scripts/bash/common.sh", + ".specify/scripts/bash/stale.sh", + ".specify/scripts/powershell/common.ps1", + ".specify/templates/obsolete-template.md", + ".specify/templates/vscode-settings.json", + ".specify/templates/.hidden-template.md", + ".specify/.gitignore", +]) +@pytest.mark.parametrize("kind", ["contained", "dangling", "external", "cycle"]) +def test_template_refresh_does_not_access_unselected_manifest_entries( + shared_project, tmp_path, monkeypatch, allow_links, force, relative, kind, +): + from pathlib import Path + + project, core = shared_project + files = FileHelper(project, allow_symlinks=allow_links) + install(project, core, files=files) + destination = project / relative + destination.parent.mkdir(parents=True, exist_ok=True) + destination.write_text("original unrelated\n") + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative, recovered=True) + manifest.save() + before_hashes = manifest.files + before_recovered = manifest.recovered_files + destination.unlink() + target = project / "unrelated-target" + if kind == "contained": + target.write_text("original unrelated\n") + elif kind == "external": + target = tmp_path / "outside" + target.write_text("original unrelated\n") + elif kind == "cycle": + target = destination + destination.symlink_to(target) + link_identity = destination.lstat().st_ino + link_target = os.readlink(destination) + (core / "templates/plan-template.md").write_text("replacement plan\n") + (core / "templates/vscode-settings.json").write_text("{}") + (core / "templates/.hidden-template.md").write_text("hidden\n") + original_stat = FileHelper.stat + original_read = FileHelper.read_bytes + + def check_selected(path): + absolute = Path(path) if Path(path).is_absolute() else project / path + assert absolute != destination, "Template refresh accessed an unrelated entry" + + def checked_stat(helper, path, **kwargs): + check_selected(path) + return original_stat(helper, path, **kwargs) + + def checked_read(helper, path): + check_selected(path) + return original_read(helper, path) + + monkeypatch.setattr(FileHelper, "stat", checked_stat) + monkeypatch.setattr(FileHelper, "read_bytes", checked_read) + refresh_shared_templates( + project, version="test", core_pack=core, repo_root=core, console=Console(), + invoke_separator=".", files=files, force=force, + ) + + assert (project / ".specify/templates/plan-template.md").read_text() == "replacement plan\n" + assert destination.is_symlink() + assert destination.lstat().st_ino == link_identity + assert os.readlink(destination) == link_target + if kind in {"contained", "external"}: + assert target.read_text() == "original unrelated\n" + after = IntegrationManifest.load("speckit", project, files=files) + assert after.files[relative] == before_hashes[relative] + assert after.recovered_files == before_recovered + + +@pytest.mark.parametrize("force", [False, True]) +def test_template_refresh_still_rejects_selected_denied_link(shared_project, force): + project, core = shared_project + files = FileHelper(project) + install(project, core, files=files) + manifest = project / ".specify/integrations/speckit.manifest.json" + before_manifest = manifest.read_bytes() + destination = project / ".specify/templates/plan-template.md" + target = project / "selected-target" + destination.rename(target) + destination.symlink_to(target) + before_content = target.read_bytes() + + with pytest.raises(SymlinkDeniedError): + refresh_shared_templates( + project, version="test", core_pack=core, repo_root=core, console=Console(), + invoke_separator=".", files=files, force=force, + ) + + assert target.read_bytes() == before_content + assert destination.is_symlink() + assert manifest.read_bytes() == before_manifest + + def test_wrapper_reads_policy_afresh_between_invocations(shared_project, monkeypatch): import specify_cli project, core = shared_project From 09ddb195266f9f237f11e577f7ab357cab9adf51 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Tue, 13 Oct 2026 13:50:38 -0500 Subject: [PATCH 3/5] test: cover shared symlink policy on install and switch Add public install/switch allow and deny cases while preserving existing overwrite behavior. Qualify integration lifecycle documentation for the already-migrated shared infrastructure without changing FileHelper or runtime code. Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 084bd222-8ae8-474a-9dcc-6e6c44e031f4 --- design/integration.md | 16 ++- .../integrations/test_installed_adapters.py | 102 ++++++++++++++++++ 2 files changed, 117 insertions(+), 1 deletion(-) diff --git a/design/integration.md b/design/integration.md index b110bbafca..414da72f84 100644 --- a/design/integration.md +++ b/design/integration.md @@ -295,7 +295,21 @@ Host settings merges, native event updates/removal, legacy migrations, and extension/preset rendering caches participate in the journal without becoming uninstall-owned files. Existing built-in home-scoped destinations are journaled only for participating built-ins; external adapters remain project-local. -Host write helpers reject symlinked destinations and ancestors before creating +FileHelper-owned shared infrastructure follows the selected symlink setting +for `.specify/scripts/`, `.specify/templates/`, `.specify/.gitignore`, and the +shared `speckit.manifest.json` I/O. Deny rejects accessed links; allow uses +supported contained targets while preserving link identity. Init supplies its +current requested setting; install/upgrade/switch read persisted project policy. +FileHelper performs filesystem operations, not ownership or overwrite decisions. +Shared callers retain their existing customization/recovery rules, and allow +mode preserves stale leaf links without following their targets. Helper-mode +manifest policy failures propagate rather than becoming tolerant recovery or preserved-file +success. This does not relax adapter-package/registry scope validation, config +bootstrap, legacy manifest consumers, or uninstall. +External adapter transactions still reject linked `.specify/integrations` scope +roots; shared manifest leaf links are supported. + +Other host write helpers reject symlinked destinations and ancestors before creating directories or writing. Manifest recording validates lexical paths before resolution, including project-local and dangling symlinks, so resolution cannot hide a link to an unrelated project file. Removing an owned leaf symlink unlinks diff --git a/tests/specify_cli/integrations/test_installed_adapters.py b/tests/specify_cli/integrations/test_installed_adapters.py index 85ab6a9200..0bb2a6babf 100644 --- a/tests/specify_cli/integrations/test_installed_adapters.py +++ b/tests/specify_cli/integrations/test_installed_adapters.py @@ -188,6 +188,108 @@ def snapshot(project): } +@pytest.mark.parametrize("operation", ["install", "switch"]) +@pytest.mark.parametrize("kind", ["leaf", "scripts-root", "templates-root", "nested-parent"]) +@pytest.mark.parametrize("allow_links", [False, True]) +def test_public_install_switch_honor_persisted_shared_symlink_policy( + tmp_path, server, monkeypatch, operation, kind, allow_links, +): + import specify_cli + + publish(server) + project = catalog_project(tmp_path, server) + core = tmp_path / "shared-core" + scripts = core / "scripts/bash" + scripts.mkdir(parents=True) + (scripts / "common.sh").write_text("# original common\n") + templates = core / "templates" + templates.mkdir() + (templates / "plan-template.md").write_text("# original plan\n") + monkeypatch.setattr(specify_cli, "_locate_core_pack", lambda: core) + if operation == "switch": + installed = run(project, ["integration", "install", "claude", "--script", "sh"]) + assert installed.exit_code == 0, installed.output + + options_path = project / ".specify/init-options.json" + options = json.loads(options_path.read_text()) if options_path.exists() else {} + options["allow_symlinks"] = allow_links + options_path.write_text(json.dumps(options)) + target = project / "shared-target" + target.mkdir() + (target / "unrelated.txt").write_text("user content\n") + if kind == "leaf": + relative = ".specify/scripts/bash/common.sh" + link = project / relative + target_file = target / "common.sh" + if operation == "switch": + link.rename(target_file) + else: + link.parent.mkdir(parents=True) + target_file.write_text("# user common\n") + link.symlink_to(target_file) + else: + relative = { + "scripts-root": ".specify/scripts", + "templates-root": ".specify/templates", + "nested-parent": ".specify/scripts/bash", + }[kind] + link = project / relative + moved = target / "managed" + if operation == "switch": + link.rename(moved) + else: + moved.mkdir() + link.parent.mkdir(parents=True, exist_ok=True) + link.symlink_to(moved, target_is_directory=True) + target_file = moved / ( + "bash/common.sh" if kind == "scripts-root" else + "plan-template.md" if kind == "templates-root" else "common.sh" + ) + relative = ( + ".specify/templates/plan-template.md" + if kind == "templates-root" else ".specify/scripts/bash/common.sh" + ) + identity = (link.lstat().st_dev, link.lstat().st_ino, os.readlink(link)) + if operation == "switch": + (scripts / "common.sh").write_text("# replacement common\n") + (templates / "plan-template.md").write_text("# replacement plan\n") + nested = scripts / "new/nested" + nested.mkdir(parents=True) + (nested / "added.sh").write_text("# newly bundled\n") + before = snapshot(project) + + result = run(project, [ + "integration", operation, KEY, "--trust-integration", "--script", "sh", + ]) + + assert link.is_symlink() + assert (link.lstat().st_dev, link.lstat().st_ino, os.readlink(link)) == identity + assert (target / "unrelated.txt").read_text() == "user content\n" + assert json.loads(options_path.read_text())["allow_symlinks"] is allow_links + if not allow_links: + assert result.exit_code == 1, result.output + assert "Refusing symlink" in result.output + assert snapshot(project) == before + assert KEY not in read_records(project) + return + + assert result.exit_code == 0, result.output + if operation == "switch": + expected = "# replacement plan\n" if kind == "templates-root" else "# replacement common\n" + else: + expected = ( + "# user common\n" if kind == "leaf" else + "# original plan\n" if kind == "templates-root" else "# original common\n" + ) + assert target_file.read_text() == expected + assert (project / ".specify/scripts/bash/new/nested/added.sh").read_text() == "# newly bundled\n" + manifest = json.loads((project / ".specify/integrations/speckit.manifest.json").read_text()) + assert manifest["files"][relative] == hashlib.sha256(target_file.read_bytes()).hexdigest() + assert (relative in manifest.get("recovered_files", [])) == (operation == "install" and kind == "leaf") + assert KEY in read_records(project) + assert json.loads((project / ".specify/integration.json").read_text())["integration"] == KEY + + @pytest.mark.parametrize("operation", ["init", "upgrade"]) @pytest.mark.parametrize("kind", ["leaf", "gitignore", "scripts-root", "templates-root", "nested-parent"]) def test_shared_helper_allowed_links_restore_in_public_external_lifecycle( From 00c717519ab7d485574f5f91d7fecad9cf7b0c95 Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Tue, 13 Oct 2026 14:11:00 -0500 Subject: [PATCH 4/5] fix: preflight stale shared paths before file updates Check stale-path policy during shared infrastructure planning so static policy failures precede file overwrites. Retain post-write leaf and ownership checks, without changing FileHelper or lifecycle behavior. Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 084bd222-8ae8-474a-9dcc-6e6c44e031f4 --- src/specify_cli/shared_infra.py | 35 +++++---- tests/specify_cli/test_shared_infra.py | 99 ++++++++++++++++++++++++++ 2 files changed, 120 insertions(+), 14 deletions(-) diff --git a/src/specify_cli/shared_infra.py b/src/specify_cli/shared_infra.py index aea837d33a..4889101a04 100644 --- a/src/specify_cli/shared_infra.py +++ b/src/specify_cli/shared_infra.py @@ -505,7 +505,8 @@ def install_shared_infra( version. Files whose hash diverges are treated as user customizations and preserved with a warning. ``force=True`` overwrites every regular file permitted by the operation's FileHelper policy, preserving allowed links. - Denied or invalid accessed links fail explicitly. ``refresh_hint`` is shown after the + Denied or invalid accessed links fail explicitly. Stale cleanup paths are + policy-checked before planned file writes. ``refresh_hint`` is shown after the customization warning to tell the user which flag would overwrite their customizations. """ @@ -535,6 +536,7 @@ def _is_managed(rel: str, dst: Path) -> bool: preserved_user_files: list[str] = [] planned_copies: list[tuple[Path, str, bytes, int]] = [] planned_templates: list[tuple[Path, str, str]] = [] + planned_stale: list[tuple[str, Path]] = [] # Track every shared path the current bundle produces so we can detect # manifest entries the core no longer ships (stale-script cleanup, #3076). seen_rels: set[str] = set() @@ -659,6 +661,22 @@ def _consider_destination(dst: Path, rel: str) -> bool: (gitignore_dst, gitignore_rel, SPECIFY_GITIGNORE_CONTENT) ) + # Validate stale paths before overwriting shared files. Ownership is checked + # again during cleanup because writes through aliases can change the bytes. + if scanned_variant_dirs: + script_prefixes = tuple( + f".specify/scripts/{variant_dir}/" for variant_dir in scanned_variant_dirs + ) + for rel in prior_hashes: + if rel in seen_rels or not rel.startswith(script_prefixes): + continue + rel_path = Path(rel) + if rel_path.anchor or ".." in rel_path.parts: + continue + dst = project_path / rel_path + files.stat(dst, follow_symlinks=False, missing_ok=True) + planned_stale.append((rel, dst)) + for dst_path, rel, content, mode in planned_copies: _ensure_dir(dst_path.parent) files.write_bytes(dst_path, content, mode=mode, observer=observe_file_change) @@ -702,20 +720,9 @@ def _consider_destination(dst: Path, rel: str) -> bool: # deletion), scoped to the selected variants, and only for *managed* copies — # a user-customized file (hash diverges), a symlink, or a recovered entry is # preserved by the leaf-identity and managed-hash checks. - if scanned_variant_dirs: + if planned_stale: stale_removed: list[str] = [] - script_prefixes = tuple( - f".specify/scripts/{variant_dir}/" for variant_dir in scanned_variant_dirs - ) - for rel in list(prior_hashes): - if rel in seen_rels or not rel.startswith(script_prefixes): - continue - # Guard corrupted/hand-edited manifest keys BEFORE any filesystem - # access. Invalid keys are ignored; accessed paths obey helper policy. - rel_path = Path(rel) - if rel_path.anchor or ".." in rel_path.parts: - continue - dst = project_path / rel_path + for rel, dst in planned_stale: entry = files.stat(dst, follow_symlinks=False, missing_ok=True) # Already gone from disk but still tracked: drop the orphaned manifest # entry so the manifest stays consistent (nothing to unlink). diff --git a/tests/specify_cli/test_shared_infra.py b/tests/specify_cli/test_shared_infra.py index 25fed612bb..fd4ba2da1b 100644 --- a/tests/specify_cli/test_shared_infra.py +++ b/tests/specify_cli/test_shared_infra.py @@ -491,6 +491,105 @@ def test_stale_leaf_links_are_never_hash_promoted_or_deleted(shared_project, tmp assert target.read_text() == "managed stale\n" +@pytest.mark.parametrize("operation", ["force", "managed", "builtin-init"]) +@pytest.mark.parametrize("kind", ["denied-leaf", "denied-parent", "external-parent", "dangling-parent", "cyclic-parent"]) +def test_stale_policy_failure_precedes_shared_file_updates( + shared_project, tmp_path, monkeypatch, operation, kind, +): + from typer.testing import CliRunner + + import specify_cli + + project, core = shared_project + allow_links = kind not in {"denied-leaf", "denied-parent"} + files = allow(project) if allow_links else FileHelper(project) + install(project, core, files=files) + relative = ".specify/scripts/bash/obsolete/stale.sh" + stale = project / relative + stale.parent.mkdir() + stale.write_text("managed stale\n") + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative) + manifest.save() + if kind == "denied-leaf": + target = project / "stale-target" + stale.rename(target) + stale.symlink_to(target) + error = SymlinkDeniedError + message = "Refusing symlink" + else: + parent = stale.parent + if kind == "denied-parent": + target = project / "stale-target" + parent.rename(target) + error, message = SymlinkDeniedError, "Refusing symlink" + elif kind == "external-parent": + target = tmp_path / "outside" + parent.rename(target) + error, message = PathEscapeError, "outside root" + else: + stale.unlink() + parent.rmdir() + if kind == "dangling-parent": + target = project / "missing" + error, message = FileNotFoundError, "No such file" + else: + target = parent + error, message = UnsupportedPathError, "Symlink cycle" + parent.symlink_to(target, target_is_directory=True) + protected = [ + project / ".specify/scripts/bash/common.sh", + project / ".specify/templates/plan-template.md", + project / ".specify/.gitignore", + project / ".specify/integrations/speckit.manifest.json", + ] + for path in protected: + path.chmod(0o600) + before = [(path.read_bytes(), path.stat().st_mode) for path in protected] + (core / "scripts/bash/common.sh").write_text("# replacement common\n") + (core / "templates/plan-template.md").write_text("# replacement plan\n") + + if operation == "builtin-init": + monkeypatch.setattr(specify_cli, "_locate_core_pack", lambda: core) + arguments = [ + "init", str(project), "--force", "--integration", "copilot", + "--ignore-agent-tools", "--script", "sh", + ] + if allow_links: + arguments.append("--allow-symlinks") + result = CliRunner().invoke(specify_cli.app, arguments, catch_exceptions=False) + assert result.exit_code == 1, result.output + assert message in result.output + else: + with pytest.raises(error): + install( + project, core, files=files, force=operation == "force", + refresh_managed=operation == "managed", + ) + + assert [(path.read_bytes(), path.stat().st_mode) for path in protected] == before + + +def test_stale_plan_rechecks_managed_hash_after_shared_writes(shared_project): + project, core = shared_project + files = allow(project) + install(project, core, files=files) + directory = project / ".specify/scripts/bash" + alias = directory / "obsolete" + alias.symlink_to(directory, target_is_directory=True) + relative = ".specify/scripts/bash/obsolete/common.sh" + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative) + manifest.save() + (core / "scripts/bash/common.sh").write_text("# replacement common\n") + + install(project, core, files=files, refresh_managed=True) + + assert (directory / "common.sh").read_text() == "# replacement common\n" + assert alias.is_symlink() + assert relative in IntegrationManifest.load("speckit", project, files=files).files + + @pytest.mark.parametrize("customized,recovered", [(False, False), (True, False), (False, True)]) def test_stale_cleanup_through_allowed_parent_obeys_ownership(shared_project, customized, recovered): project, core = shared_project From 7846fc0982275e57ebcaae04daf2e24fcea010ff Mon Sep 17 00:00:00 2001 From: Manfred Riem <15701806+mnriem@users.noreply.github.com> Date: Tue, 13 Oct 2026 14:34:46 -0500 Subject: [PATCH 5/5] fix: reject embedded drive components in stale manifest keys Reject anchored native path components before constructing stale shared destinations. Add portable Windows parsing regressions and preserve valid POSIX colon filenames without changing FileHelper or lifecycle code. Assisted-by: GitHub Copilot (model: GPT-6.1 Sol, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 084bd222-8ae8-474a-9dcc-6e6c44e031f4 --- src/specify_cli/shared_infra.py | 4 +- tests/specify_cli/test_shared_infra.py | 61 ++++++++++++++++++++++++++ 2 files changed, 64 insertions(+), 1 deletion(-) diff --git a/src/specify_cli/shared_infra.py b/src/specify_cli/shared_infra.py index 4889101a04..cab67ee2e2 100644 --- a/src/specify_cli/shared_infra.py +++ b/src/specify_cli/shared_infra.py @@ -671,7 +671,9 @@ def _consider_destination(dst: Path, rel: str) -> bool: if rel in seen_rels or not rel.startswith(script_prefixes): continue rel_path = Path(rel) - if rel_path.anchor or ".." in rel_path.parts: + if rel_path.anchor or any( + part == ".." or Path(part).anchor for part in rel_path.parts + ): continue dst = project_path / rel_path files.stat(dst, follow_symlinks=False, missing_ok=True) diff --git a/tests/specify_cli/test_shared_infra.py b/tests/specify_cli/test_shared_infra.py index fd4ba2da1b..2957c49ca9 100644 --- a/tests/specify_cli/test_shared_infra.py +++ b/tests/specify_cli/test_shared_infra.py @@ -590,6 +590,67 @@ def test_stale_plan_rechecks_managed_hash_after_shared_writes(shared_project): assert relative in IntegrationManifest.load("speckit", project, files=files).files +@pytest.mark.parametrize("suffix", ["D:outside", "D:", "nested/D:outside", "D:/outside", r"D:\outside"]) +def test_stale_windows_drive_components_are_skipped_before_access( + shared_project, tmp_path, monkeypatch, suffix, +): + from pathlib import Path, PureWindowsPath + + from specify_cli import shared_infra + + project, core = shared_project + files = FileHelper(project) + install(project, core, files=files) + victim = tmp_path / "outside" + victim.write_bytes(b"outside user content\n") + relative = f".specify/scripts/bash/{suffix}" + manifest_path = project / ".specify/integrations/speckit.manifest.json" + data = json.loads(manifest_path.read_text()) + data["files"][relative] = hashlib.sha256(victim.read_bytes()).hexdigest() + manifest_path.write_text(json.dumps(data)) + parsed = PureWindowsPath(relative) + assert not parsed.anchor and ".." not in parsed.parts + original_stat = FileHelper.stat + + def windows_manifest_path(value): + if isinstance(value, str) and not Path(value).is_absolute(): + return PureWindowsPath(value) + return Path(value) + + def checked_stat(helper, path, **kwargs): + checked = Path(path) + if checked.is_absolute(): + checked = checked.relative_to(project) + assert "D:" not in str(checked), "Corrupt drive component reached filesystem access" + return original_stat(helper, path, **kwargs) + + monkeypatch.setattr(shared_infra, "Path", windows_manifest_path) + monkeypatch.setattr(FileHelper, "stat", checked_stat) + install(project, core, files=files, refresh_managed=True) + + assert victim.read_bytes() == b"outside user content\n" + assert relative in json.loads(manifest_path.read_text())["files"] + assert (project / ".specify/scripts/bash/common.sh").read_text() == "# bundled\n" + + +@pytest.mark.skipif(os.name == "nt", reason="Colons are native POSIX filename characters") +def test_posix_stale_drive_looking_filename_remains_a_native_filename(shared_project): + project, core = shared_project + files = FileHelper(project) + install(project, core, files=files) + relative = ".specify/scripts/bash/D:obsolete" + stale = project / relative + stale.write_text("managed stale\n") + manifest = IntegrationManifest.load("speckit", project, files=files) + manifest.record_existing(relative) + manifest.save() + + install(project, core, files=files) + + assert not stale.exists() + assert relative not in IntegrationManifest.load("speckit", project, files=files).files + + @pytest.mark.parametrize("customized,recovered", [(False, False), (True, False), (False, True)]) def test_stale_cleanup_through_allowed_parent_obeys_ownership(shared_project, customized, recovered): project, core = shared_project