diff --git a/design/file-helper.md b/design/file-helper.md index f5497ef256..914bcccfef 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,60 @@ 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 + +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. +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 +269,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/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/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..11c30f7b88 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() @@ -337,14 +371,34 @@ 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: 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 +535,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 +561,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 +590,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..cab67ee2e2 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,38 @@ 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.""" + """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(): 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()) + 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" - _ensure_safe_shared_directory(project_path, dest_templates) - for src in templates_src.iterdir(): - if not src.is_file() or src.name == "vscode-settings.json" or src.name.startswith("."): - continue - + files.mkdir(dest_templates, parents=True, exist_ok=True, observer=observe_directory_change) + for src in templates: 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 +470,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 +496,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,33 +504,39 @@ 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. 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. """ - 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]] = [] + 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() @@ -524,7 +550,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 +562,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 +652,40 @@ 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) + ) + + # 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 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) + planned_stale.append((rel, dst)) 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 +703,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,48 +721,22 @@ 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``. - if scanned_variant_dirs: + # preserved by the leaf-identity and managed-hash checks. + 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: 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. - 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: - 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). - 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/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/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..0bb2a6babf 100644 --- a/tests/specify_cli/integrations/test_installed_adapters.py +++ b/tests/specify_cli/integrations/test_installed_adapters.py @@ -188,6 +188,223 @@ 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( + 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..2957c49ca9 --- /dev/null +++ b/tests/specify_cli/test_shared_infra.py @@ -0,0 +1,847 @@ +"""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") + + +@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 + 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("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("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 + 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"),