diff --git a/design/file-helper.md b/design/file-helper.md index f5497ef256..591cedf487 100644 --- a/design/file-helper.md +++ b/design/file-helper.md @@ -88,6 +88,29 @@ file-change observer snapshots it, then retain the before/after observer hooks around the write. Downstream init failures therefore remain covered by the existing lifecycle rollback, including prior policy and unrelated options. +Extension registry I/O also uses a deny-mode helper rooted at the project, so +`.specify`, `extensions`, and `.registry` are not trusted aliases. Registry +reads and writes reject linked storage regardless of operational symlink +policy. For a not-yet-created project, the nearest existing ancestor supplies +the root without creating storage during reads. Writes validate the destination +before file-change observers run and preserve the existing non-atomic text-write +and metadata semantics. Missing or malformed registries retain their recovery +behavior, but parent policy is validated even for a missing or directory-valued +registry. Artifact inventory, template resolution, and event command resolution +fail closed on forbidden storage. CLI commands preserve their diagnostics and +JSON error envelopes; warning-only dependency and configuration probes preserve +their no-results fallback without reading through forbidden paths. +Generated event dispatchers validate the registry hierarchy using only the +standard library before delegating to an installed CLI or using their inline +fallback, so unavailable or stale packages cannot bypass link and reparse-point +rejection. A rejected hierarchy produces a diagnostic and exit status 2. +Bundled Bash, Python, and Windows PowerShell 5.1 template helpers reject linked +registry storage before reading or scanning the extension layer, including +dangling parent links. Higher-priority replacement layers retain their existing +short-circuit behavior. +This boundary has the portable helper's +concurrent-mutation limits described below. + ```python from specify_cli.file_helper import FileHelper @@ -136,6 +159,7 @@ helpers for independent project/source/managed boundaries. | Operation | Default deny | Explicit allow | |---|---|---| +| Read metadata | Reject accessed parent or leaf links | Follow only existing contained regular-file or directory targets | | Read file | Reject accessed parent or leaf links | Follow only existing contained regular-file targets | | Create directory | Reject accessed links before creating parents | Follow contained directory links; reject dangling targets | | Exclusively create file | Reject accessed links | Follow contained parents; reject an existing leaf without opening its target | @@ -148,6 +172,8 @@ helpers for independent project/source/managed boundaries. ## API and failures - `mkdir(path, parents=False, exist_ok=False)` creates directories. +- `stat(path)` reads file or directory metadata after applying the accessed-path + policy to the full hierarchy, including when a later component is missing. - `read_bytes(path)` / `read_text(path, encoding="utf-8")` read regular files. - `create_bytes(path, content, mode=0o644)` / `create_text(...)` exclusively create files with existing parents. A no-follow `lstat` preflight rejects diff --git a/design/integration.md b/design/integration.md index 65c62c3162..976752dab0 100644 --- a/design/integration.md +++ b/design/integration.md @@ -261,6 +261,11 @@ metadata; they do not copy agent trees, user data, or the installed package store. Content snapshots are created lazily before the first observed mutation, with an aggregate limit of 128 MiB of file content and 4,096 entries per transaction. Exceeding either limit fails before the affected write or removal. +Switch resolves the target's event configuration before removing the current +integration or its metadata. When events are enabled, this validates extension +registry storage even if a linked parent appears absent. A rejected hierarchy +leaves the current installation unchanged, including built-in-only switches +that do not use the external-adapter lifecycle journal. Preset and extension package installation/replacement, configuration rescue directories, and project extension configuration participate in the journal. Package replacements keep one bounded directory snapshot rather than duplicate diff --git a/scripts/bash/common.sh b/scripts/bash/common.sh index ff18cba4d9..ef7bcf606e 100644 --- a/scripts/bash/common.sh +++ b/scripts/bash/common.sh @@ -427,8 +427,21 @@ _python3_command() { fi } +_validate_extension_registry_hierarchy() { + local ext_dir="$1" + local specify_dir="${ext_dir%/*}" + local path + for path in "$specify_dir" "$ext_dir" "$ext_dir/.registry"; do + if [ -L "$path" ]; then + echo "Error: refusing symlink in extension registry hierarchy: $path" >&2 + return 1 + fi + done +} + _sorted_extension_ids() { local ext_dir="$1" + _validate_extension_registry_hierarchy "$ext_dir" || return 1 local -a python_cmd=() local _python_cmd_line while IFS= read -r _python_cmd_line; do @@ -438,12 +451,23 @@ _sorted_extension_ids() { local py_stderr sorted_ids py_stderr=$(mktemp) if sorted_ids=$(SPECKIT_EXTENSIONS="$ext_dir" "${python_cmd[@]}" -c " -import json, os, re, sys +import json, os, re, stat, sys from pathlib import Path root = Path(os.environ['SPECKIT_EXTENSIONS']) registered = {} registry = root / '.registry' +for path in (root.parent, root, registry): + try: + entry = path.lstat() + except FileNotFoundError: + break + except OSError as exc: + print('registry_invalid: ' + str(exc), file=sys.stderr) + sys.exit(1) + if stat.S_ISLNK(entry.st_mode) or getattr(entry, 'st_file_attributes', 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT: + print('registry_invalid: refusing symlink or reparse point', file=sys.stderr) + sys.exit(1) if os.path.lexists(registry): if not registry.is_file(): print('registry_invalid: not a regular file', file=sys.stderr) @@ -475,9 +499,10 @@ ranked = [] for ext_id, meta in registered.items(): if isinstance(ext_id, str) and re.fullmatch(r'[a-z0-9-]+', ext_id) and isinstance(meta, dict) and bool(meta.get('enabled', True)): ranked.append((priority(meta.get('priority')), ext_id)) -for path in root.iterdir(): - if path.is_dir() and re.fullmatch(r'[a-z0-9-]+', path.name) and path.name not in registered: - ranked.append((10, path.name)) +if root.is_dir(): + for path in root.iterdir(): + if path.is_dir() and re.fullmatch(r'[a-z0-9-]+', path.name) and path.name not in registered: + ranked.append((10, path.name)) for _, ext_id in sorted(ranked): print(ext_id) " 2>"$py_stderr"); then @@ -593,19 +618,17 @@ except Exception: # Priority 3: Extension-provided templates local ext_dir="$repo_root/.specify/extensions" - if [ -d "$ext_dir" ]; then - local sorted_extensions="" - if ! sorted_extensions=$(_sorted_extension_ids "$ext_dir"); then - return 2 - fi - while IFS= read -r extension_id; do - [ -n "$extension_id" ] || continue - local ext="$ext_dir/$extension_id" - local candidate="$ext/templates/${template_name}.md" - [ -f "$candidate" ] || candidate="$ext/${template_name}.md" - [ -f "$candidate" ] && echo "$candidate" && return 0 - done <<< "$sorted_extensions" + local sorted_extensions="" + if ! sorted_extensions=$(_sorted_extension_ids "$ext_dir"); then + return 2 fi + while IFS= read -r extension_id; do + [ -n "$extension_id" ] || continue + local ext="$ext_dir/$extension_id" + local candidate="$ext/templates/${template_name}.md" + [ -f "$candidate" ] || candidate="$ext/${template_name}.md" + [ -f "$candidate" ] && echo "$candidate" && return 0 + done <<< "$sorted_extensions" # Priority 4: Core templates local core="$base/${template_name}.md" @@ -815,7 +838,7 @@ except Exception as exc: # Priority 3: Extension-provided templates (always "replace") local ext_dir="$repo_root/.specify/extensions" - if [ "$effective_base_found" = false ] && [ -d "$ext_dir" ]; then + if [ "$effective_base_found" = false ]; then local sorted_extensions="" if ! sorted_extensions=$(_sorted_extension_ids "$ext_dir"); then return 2 diff --git a/scripts/powershell/common.ps1 b/scripts/powershell/common.ps1 index c0aa9154b5..01543de509 100644 --- a/scripts/powershell/common.ps1 +++ b/scripts/powershell/common.ps1 @@ -366,9 +366,26 @@ function Get-NormalizedPriority { return $(if ($parsedPriority -ge 1) { $parsedPriority } else { 10 }) } +function Assert-ExtensionRegistryHierarchy { + param([Parameter(Mandatory=$true)][string]$ExtensionsDir) + + $current = Split-Path -Parent (Split-Path -Parent $ExtensionsDir) + foreach ($component in @('.specify', 'extensions', '.registry')) { + $entry = Get-ChildItem -LiteralPath $current -Force -ErrorAction Stop | + Where-Object { $_.Name -eq $component } | + Select-Object -First 1 + if ($null -eq $entry) { return } + if ($entry.Attributes -band [System.IO.FileAttributes]::ReparsePoint) { + throw "Refusing symlink or unsupported reparse point: $(Join-Path $current $component)" + } + $current = Join-Path $current $component + } +} + function Get-SortedExtensionIds { param([Parameter(Mandatory=$true)][string]$ExtensionsDir) + Assert-ExtensionRegistryHierarchy -ExtensionsDir $ExtensionsDir $registeredNames = @() $ranked = @() $registryFile = Join-Path $ExtensionsDir '.registry' @@ -509,14 +526,12 @@ function Resolve-Template { # Priority 3: Extension-provided templates $extDir = Join-Path $RepoRoot '.specify/extensions' - if (Test-Path $extDir) { - foreach ($extensionId in Get-SortedExtensionIds -ExtensionsDir $extDir) { - $candidate = Join-Path $extDir "$extensionId/templates/$TemplateName.md" - if (-not (Test-Path $candidate)) { - $candidate = Join-Path $extDir "$extensionId/$TemplateName.md" - } - if (Test-Path $candidate) { return $candidate } + foreach ($extensionId in Get-SortedExtensionIds -ExtensionsDir $extDir) { + $candidate = Join-Path $extDir "$extensionId/templates/$TemplateName.md" + if (-not (Test-Path $candidate)) { + $candidate = Join-Path $extDir "$extensionId/$TemplateName.md" } + if (Test-Path $candidate) { return $candidate } } # Priority 4: Core templates @@ -729,7 +744,7 @@ except Exception as exc: # Priority 3: Extension-provided templates (always "replace") $extDir = Join-Path $RepoRoot '.specify/extensions' - if (-not $effectiveBaseFound -and (Test-Path $extDir)) { + if (-not $effectiveBaseFound) { foreach ($extensionId in Get-SortedExtensionIds -ExtensionsDir $extDir) { $candidate = Join-Path $extDir "$extensionId/templates/$TemplateName.md" if (-not (Test-Path $candidate)) { diff --git a/scripts/python/common.py b/scripts/python/common.py index 0cd17e013d..60f5a182fd 100644 --- a/scripts/python/common.py +++ b/scripts/python/common.py @@ -5,6 +5,7 @@ import json import os import re +import stat import subprocess import sys from dataclasses import dataclass @@ -246,7 +247,23 @@ def _sorted_preset_ids(presets_dir: Path) -> list[str]: return [] +def _validate_extension_registry_hierarchy(extensions_dir: Path) -> None: + for path in (extensions_dir.parent, extensions_dir, extensions_dir / ".registry"): + try: + entry = path.lstat() + except FileNotFoundError: + return + except OSError as exc: + raise TemplateResolutionError(f"Cannot inspect extension registry {path}: {exc}") from exc + if ( + stat.S_ISLNK(entry.st_mode) + or getattr(entry, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT + ): + raise TemplateResolutionError(f"Refusing symlink or unsupported reparse point: {path}") + + def _sorted_extension_ids(extensions_dir: Path) -> list[str]: + _validate_extension_registry_hierarchy(extensions_dir) registry = extensions_dir / ".registry" registered_ids: set[str] = set() extensions: dict[object, object] = {} @@ -340,12 +357,11 @@ def resolve_template(template_name: str, repo_root: Path) -> Path | None: return candidate ext_dir = repo_root / ".specify" / "extensions" - if ext_dir.is_dir(): - for extension_id in _sorted_extension_ids(ext_dir): - ext = ext_dir / extension_id - candidate = _conventional_template(ext, template_name) - if candidate is not None: - return candidate + for extension_id in _sorted_extension_ids(ext_dir): + ext = ext_dir / extension_id + candidate = _conventional_template(ext, template_name) + if candidate is not None: + return candidate core = base / f"{template_name}.md" if core.is_file(): diff --git a/src/specify_cli/artifacts/catalog.py b/src/specify_cli/artifacts/catalog.py index 2780201cbf..1e35b8bd29 100644 --- a/src/specify_cli/artifacts/catalog.py +++ b/src/specify_cli/artifacts/catalog.py @@ -210,12 +210,14 @@ def _validate_project(project_root: Path) -> None: def _validate_extension_registry(project_root: Path) -> None: extensions_dir = project_root / ".specify" / "extensions" - if not extensions_dir.exists(): - return - from ..extensions import ExtensionRegistry + from ..file_helper import FileHelperError - if ExtensionRegistry(extensions_dir).is_corrupt(): + try: + registry = ExtensionRegistry(extensions_dir) + except FileHelperError as exc: + raise ArtifactResolutionError() from exc + if registry.is_corrupt(): raise ArtifactResolutionError() diff --git a/src/specify_cli/events/__init__.py b/src/specify_cli/events/__init__.py index 47f86a5cb7..1a9c2db0f7 100644 --- a/src/specify_cli/events/__init__.py +++ b/src/specify_cli/events/__init__.py @@ -30,6 +30,7 @@ import yaml import typer +from ..file_helper import FileHelperError from ..integrations._file_changes import unlink as _unlink_file, write_text as _write_text if TYPE_CHECKING: @@ -104,11 +105,28 @@ import re import shlex import shutil +import stat import subprocess import sys from pathlib import Path, PurePosixPath, PureWindowsPath +def _validate_registry_hierarchy(project_root): + """Reject registry links without depending on an installed CLI package.""" + current = project_root + for component in (".specify", "extensions", ".registry"): + current = current / component + try: + entry = current.lstat() + except FileNotFoundError: + return + if ( + stat.S_ISLNK(entry.st_mode) + or getattr(entry, "st_file_attributes", 0) & stat.FILE_ATTRIBUTE_REPARSE_POINT + ): + raise ValueError(f"Refusing symlink or unsupported reparse point: {current}") + + def _script_under_base(base, token, project_root): """Return token resolved under base, or None if it leaves the project.""" posix_path = PurePosixPath(token) @@ -416,6 +434,11 @@ def main(): else: payload = "{}" project_root = Path(__file__).parent.parent.resolve() + try: + _validate_registry_hierarchy(project_root) + except (OSError, ValueError) as exc: + sys.stderr.write(f"Event command {command_name} error: {exc}\\n") + sys.exit(2) # Preferred path: specify_cli is importable (durable install) — delegate to # the full resolver, which also handles extension manifests whose file stem @@ -559,6 +582,8 @@ def _find_command_template(command_name: str, project_root: Path) -> tuple[Path candidate = exts_dir / ext_id / cmd["file"] if candidate.exists(): return candidate, ext_id + except FileHelperError: + raise except Exception: # Fall through to the on-disk scan if the registry/manifests can't be # read; event dispatch should degrade gracefully, not crash. @@ -789,7 +814,11 @@ def resolve_and_run_event_command( by the Claude Code hooks spec Gemini/Tabnine/Devin derive from). Only used when *envelope* is ``hookSpecificOutput``. """ - template_path, ext_id = _find_command_template(command_name, project_root) + try: + template_path, ext_id = _find_command_template(command_name, project_root) + except FileHelperError as exc: + sys.stderr.write(f"Event command {command_name} error: {exc}\n") + return 2 if not template_path: logger.warning("Event command '%s' not found", command_name) return 0 @@ -1060,13 +1089,13 @@ def _disabled_extension_ids(project_root: Path) -> set[str]: exts_dir = project_root / ".specify" / "extensions" disabled_ids: set[str] = set() - if not exts_dir.is_dir(): - return disabled_ids try: registry = ExtensionRegistry(exts_dir) for ext_id, meta in registry.list_by_priority(include_disabled=True): if not isinstance(meta, dict) or not meta.get("enabled", True): disabled_ids.add(ext_id) + except FileHelperError: + raise except Exception: pass return disabled_ids @@ -1096,11 +1125,10 @@ def collect_extension_events(project_root: Path) -> ResolvedEvents: events: ResolvedEvents = {} exts_dir = project_root / ".specify" / "extensions" + manager = ExtensionManager(project_root) if not exts_dir.is_dir(): return events - manager = ExtensionManager(project_root) - # Build the set of explicitly-disabled extension IDs. Extensions not # tracked in the registry are treated as enabled (backward compat). disabled_ids = _disabled_extension_ids(project_root) diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index f6005f6012..8bbf66f123 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -47,6 +47,7 @@ from .._utils import dump_frontmatter, relative_extension_path_violation, version_satisfies from ..catalogs import CatalogEntry as BaseCatalogEntry from ..catalogs import CatalogStackBase +from ..file_helper import FileHelper, FileHelperError from ..integration_state import ( INTEGRATION_STATE_SCHEMA, default_integration_key, @@ -761,25 +762,21 @@ def __init__(self, extensions_dir: Path): Args: extensions_dir: Path to .specify/extensions/ directory """ - self.extensions_dir = extensions_dir - self.registry_path = extensions_dir / self.REGISTRY_FILE + self.extensions_dir = extensions_dir.absolute() + self.registry_path = self.extensions_dir / self.REGISTRY_FILE + # Trust the project root, not the repository-controlled storage hierarchy. + root = self.extensions_dir.parent.parent + while not root.exists() and not root.is_symlink(): + root = root.parent + self._files = FileHelper(root) self.data = self._load() def _load(self) -> dict: """Load registry from disk.""" - if not self.registry_path.exists(): - return {"schema_version": self.SCHEMA_VERSION, "extensions": {}} - - # A non-regular file (e.g. a directory at the registry path) is not a - # readable registry. Recover to empty so construction — used by the - # install/enable/disable flows — does not crash. Resolution paths that - # must fail closed consult is_corrupt() instead of relying on this. - if not self.registry_path.is_file(): - return {"schema_version": self.SCHEMA_VERSION, "extensions": {}} - try: - with open(self.registry_path, "r", encoding="utf-8") as f: - data = json.load(f) + if not stat.S_ISREG(self._files.stat(self.registry_path).st_mode): + return {"schema_version": self.SCHEMA_VERSION, "extensions": {}} + data = json.loads(self._files.read_text(self.registry_path)) # Validate loaded data is a dict (handles corrupted registry files) if not isinstance(data, dict): return {"schema_version": self.SCHEMA_VERSION, "extensions": {}} @@ -810,17 +807,11 @@ def is_corrupt(self) -> bool: or dangling symlink, non-regular file, unreadable file, non-mapping root, or non-mapping ``extensions`` value returns ``True``. """ - # os.path.lexists (not Path.exists) so a dangling symlink is detected - # rather than followed to a non-existent target and mistaken for an - # absent registry — which would reopen the fail-open directory scan. - if not os.path.lexists(self.registry_path): - return False - if not self.registry_path.is_file(): - return True try: - with open(self.registry_path, "r", encoding="utf-8") as f: - data = json.load(f) - except (json.JSONDecodeError, OSError, UnicodeDecodeError): + data = json.loads(self._files.read_text(self.registry_path)) + except FileNotFoundError: + return False + except (json.JSONDecodeError, OSError, UnicodeDecodeError, FileHelperError): return True if not isinstance(data, dict): return True @@ -832,10 +823,14 @@ def _save(self): """Save registry to disk.""" from ..integrations._file_changes import after_file_change, before_file_change + # Reject linked destinations before lifecycle observers can read them. + try: + self._files.read_bytes(self.registry_path) + except FileNotFoundError: + pass before_file_change(self.registry_path) - self.extensions_dir.mkdir(parents=True, exist_ok=True) - with open(self.registry_path, "w", encoding="utf-8") as f: - json.dump(self.data, f, indent=2) + self._files.mkdir(self.extensions_dir, parents=True, exist_ok=True) + self._files.overwrite_text(self.registry_path, json.dumps(self.data, indent=2)) after_file_change(self.registry_path) def add(self, extension_id: str, metadata: dict): @@ -5375,7 +5370,7 @@ def _sibling_extension_ids(self) -> list[str]: absorbing its ``SPECKIT__*`` env vars into no one. The registry is the source of truth for "installed". - Returns an empty list if the registry is missing or corrupted + Returns an empty list if the registry is missing, corrupted, or forbidden (fresh project, ad-hoc test harness) so ``_get_env_config`` degrades to its pre-fix behaviour rather than crashing. ``UnicodeError`` is kept alongside ``OSError`` as belt-and-braces: ``_load()`` now starts @@ -5390,7 +5385,7 @@ def _sibling_extension_ids(self) -> list[str]: extensions_dir = self.project_root / ".specify" / "extensions" try: return list(ExtensionRegistry(extensions_dir).keys()) - except (OSError, UnicodeError): + except (OSError, UnicodeError, FileHelperError): return [] def _get_env_config(self) -> Dict[str, Any]: diff --git a/src/specify_cli/extensions/_command_update_transaction.py b/src/specify_cli/extensions/_command_update_transaction.py index 585b3cc2bf..c589496ce5 100644 --- a/src/specify_cli/extensions/_command_update_transaction.py +++ b/src/specify_cli/extensions/_command_update_transaction.py @@ -24,7 +24,6 @@ def run_update_command(extension: str | None) -> None: """Run discovery, confirmation, and transactional extension updates.""" from . import ( - ExtensionManager, ExtensionCatalog, ExtensionError, ValidationError, @@ -34,7 +33,7 @@ def run_update_command(extension: str | None) -> None: ) project_root = _commands._require_specify_project() - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) catalog = ExtensionCatalog(project_root) speckit_version = _commands.get_speckit_version() diff --git a/src/specify_cli/extensions/_commands.py b/src/specify_cli/extensions/_commands.py index 08fe6747fc..2046ba67f3 100644 --- a/src/specify_cli/extensions/_commands.py +++ b/src/specify_cli/extensions/_commands.py @@ -40,6 +40,17 @@ def _require_specify_project(*args, **kwargs): from .. import _require_specify_project as _f return _f(*args, **kwargs) +def _extension_manager(project_root: Path): + """Translate registry policy failures for human-readable CLI commands.""" + from ..file_helper import FileHelperError + from . import ExtensionManager + + try: + return ExtensionManager(project_root) + except FileHelperError as exc: + console.print(f"[red]Error:[/red] {_escape_markup(str(exc))}") + raise typer.Exit(1) from exc + def _locate_bundled_extension(*args, **kwargs): from .. import _locate_bundled_extension as _f diff --git a/src/specify_cli/extensions/command_add.py b/src/specify_cli/extensions/command_add.py index 1917663406..89f5ca51d5 100644 --- a/src/specify_cli/extensions/command_add.py +++ b/src/specify_cli/extensions/command_add.py @@ -14,6 +14,7 @@ from rich.panel import Panel from .._console import console +from ..file_helper import FileHelperError from . import _commands @@ -27,7 +28,7 @@ def extension_add( version: Optional[str] = typer.Option(None, "--version", help="Install an exact version from a catalog"), ): """Install an extension.""" - from . import ExtensionManager, ExtensionCatalog, ExtensionError, ValidationError, CompatibilityError, REINSTALL_COMMAND + from . import ExtensionCatalog, ExtensionError, ValidationError, CompatibilityError, REINSTALL_COMMAND # Compatibility callers invoke this function directly, in which case # Typer supplies its OptionInfo object instead of a parsed option value. @@ -43,7 +44,7 @@ def extension_add( console.print("[red]Error:[/red] --version requires a catalog install (without --dev or --from).") raise typer.Exit(1) - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) speckit_version = _commands.get_speckit_version() if force: @@ -341,6 +342,6 @@ def extension_add( except CompatibilityError as e: console.print(f"\n[red]Compatibility Error:[/red] {_escape_markup(str(e))}") raise typer.Exit(1) - except ExtensionError as e: + except (ExtensionError, FileHelperError) as e: console.print(f"\n[red]Error:[/red] {_escape_markup(str(e))}") raise typer.Exit(1) diff --git a/src/specify_cli/extensions/command_disable.py b/src/specify_cli/extensions/command_disable.py index 16ef6683e1..c0db1e3700 100644 --- a/src/specify_cli/extensions/command_disable.py +++ b/src/specify_cli/extensions/command_disable.py @@ -17,10 +17,10 @@ def extension_disable( extension: str = typer.Argument(help="Extension ID or name to disable"), ): """Disable an extension without removing it.""" - from . import ExtensionManager, HookExecutor + from . import HookExecutor project_root = _commands._require_specify_project() - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) hook_executor = HookExecutor(project_root) # Resolve extension ID from argument (handles ambiguous names) diff --git a/src/specify_cli/extensions/command_enable.py b/src/specify_cli/extensions/command_enable.py index 34bab2f279..3ca750611f 100644 --- a/src/specify_cli/extensions/command_enable.py +++ b/src/specify_cli/extensions/command_enable.py @@ -17,10 +17,10 @@ def extension_enable( extension: str = typer.Argument(help="Extension ID or name to enable"), ): """Enable a disabled extension.""" - from . import ExtensionError, ExtensionManager, HookExecutor + from . import ExtensionError, HookExecutor project_root = _commands._require_specify_project() - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) hook_executor = HookExecutor(project_root) # Resolve extension ID from argument (handles ambiguous names) diff --git a/src/specify_cli/extensions/command_info.py b/src/specify_cli/extensions/command_info.py index dfc685caa2..4348a23f60 100644 --- a/src/specify_cli/extensions/command_info.py +++ b/src/specify_cli/extensions/command_info.py @@ -39,7 +39,7 @@ def extension_info( project_root = _commands._require_specify_project() catalog = ExtensionCatalog(project_root) - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) installed = manager.list_installed() # Try to resolve from installed extensions first (by ID or name) diff --git a/src/specify_cli/extensions/command_list.py b/src/specify_cli/extensions/command_list.py index 90a536f9e4..3cd0a3da69 100644 --- a/src/specify_cli/extensions/command_list.py +++ b/src/specify_cli/extensions/command_list.py @@ -48,7 +48,7 @@ def extension_list( emit_json_error(error) project_root = _commands._require_specify_project() - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) installed = manager.list_installed() if not installed and not (available or all_extensions): diff --git a/src/specify_cli/extensions/command_remove.py b/src/specify_cli/extensions/command_remove.py index c210dae96b..1f2ff50552 100644 --- a/src/specify_cli/extensions/command_remove.py +++ b/src/specify_cli/extensions/command_remove.py @@ -19,10 +19,8 @@ def extension_remove( force: bool = typer.Option(False, "--force", help="Skip confirmation"), ): """Uninstall an extension.""" - from . import ExtensionManager - project_root = _commands._require_specify_project() - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) # Resolve extension ID from argument (handles ambiguous names) installed = manager.list_installed() diff --git a/src/specify_cli/extensions/command_set_priority.py b/src/specify_cli/extensions/command_set_priority.py index 3adbe73c9b..c91fb46ccd 100644 --- a/src/specify_cli/extensions/command_set_priority.py +++ b/src/specify_cli/extensions/command_set_priority.py @@ -18,7 +18,7 @@ def extension_set_priority( priority: int = typer.Argument(help="New priority (lower = higher precedence)"), ): """Set the resolution priority of an installed extension.""" - from . import ExtensionManager, normalize_priority + from . import normalize_priority project_root = _commands._require_specify_project() # Validate priority @@ -26,7 +26,7 @@ def extension_set_priority( console.print("[red]Error:[/red] Priority must be a positive integer (1 or higher)") raise typer.Exit(1) - manager = ExtensionManager(project_root) + manager = _commands._extension_manager(project_root) # Resolve extension ID from argument (handles ambiguous names) installed = manager.list_installed() diff --git a/src/specify_cli/file_helper.py b/src/specify_cli/file_helper.py index 49441ce494..3486136683 100644 --- a/src/specify_cli/file_helper.py +++ b/src/specify_cli/file_helper.py @@ -159,6 +159,10 @@ def mkdir( destination = self._path(path, allow_missing=True) destination.mkdir(parents=parents, exist_ok=exist_ok) + def stat(self, path: Path | str) -> os.stat_result: + """Read metadata under the same traversal policy as file contents.""" + return self._path(path).stat() + def read_bytes(self, path: Path | str) -> bytes: destination = self._path(path) self._require_file(destination) diff --git a/src/specify_cli/integrations/command_switch.py b/src/specify_cli/integrations/command_switch.py index f9a42de3ea..41cfb4a9ac 100644 --- a/src/specify_cli/integrations/command_switch.py +++ b/src/specify_cli/integrations/command_switch.py @@ -7,6 +7,7 @@ from rich.markup import escape from .._console import console +from ..file_helper import FileHelperError from ..integration_runtime import invoke_prefix_for_integration as _invoke_prefix_for_integration, invoke_separator_for_integration as _invoke_separator_for_integration from ..integration_state import ( dedupe_integration_keys as _dedupe_integration_keys, @@ -161,6 +162,23 @@ def integration_switch( console.print("No files were changed.") raise typer.Exit(1) + from ..events import resolve_events + + try: + events_map = resolve_events( + target_integration.key, + target_integration.config, + project_root, + target_parsed_options, + ) + except (OSError, FileHelperError) as exc: + console.print( + f"[red]Error:[/red] Could not resolve events for integration " + f"'{escape(target)}': {_cli_error_detail(exc)}" + ) + console.print("No files were changed.") + raise typer.Exit(1) from exc + # Phase 1: Uninstall current integration (if any) if installed_key: current_integration = get_integration(installed_key) @@ -292,13 +310,6 @@ def integration_switch( target_integration.key, project_root, version=_get_speckit_version() ) - from ..events import resolve_events - events_map = resolve_events( - target_integration.key, - target_integration.config, - project_root, - target_parsed_options, - ) try: target_integration.setup( project_root, manifest, diff --git a/src/specify_cli/presets/_manager.py b/src/specify_cli/presets/_manager.py index cf6c61f434..57da07b6ee 100644 --- a/src/specify_cli/presets/_manager.py +++ b/src/specify_cli/presets/_manager.py @@ -15,6 +15,7 @@ from .._install_ignore import load_install_ignore from .._utils import version_satisfies from ..extensions import REINSTALL_COMMAND, normalize_priority +from ..file_helper import FileHelperError from ..shared_infra import ( _ensure_safe_shared_destination, _ensure_safe_shared_directory, @@ -234,8 +235,8 @@ def find_unmet_extension_dependencies( ``"disabled"``, or ``"version"``. Optional dependencies (``required: false``) are never reported. - An unreadable registry yields no results rather than raising, since - this runs after the install has already succeeded. + An unreadable or forbidden registry yields no results rather than + raising, since this runs after the install has already succeeded. A registry version that cannot be parsed is treated as uncomparable, not as a mismatch: the extension is installed and @@ -273,13 +274,13 @@ def find_unmet_extension_dependencies( registry = ExtensionRegistry(extensions_dir) registered_ids = registry.keys() registry_corrupt = registry.is_corrupt() - except OSError: + except (OSError, FileHelperError): # Both reads can raise: _load() recovers from malformed content but # deliberately lets OSError through, and is_corrupt() re-reads the # file. This check runs *after* the install has completed, and # preset_add only handles preset-domain errors, so letting that # escape would turn a finished install into a traceback over a - # warning. An unreadable registry simply cannot be inspected. + # warning. An unreadable or forbidden registry cannot be inspected. return [] unmet: List[Dict[str, Any]] = [] diff --git a/src/specify_cli/presets/_resolver.py b/src/specify_cli/presets/_resolver.py index 5700fa8378..ec09de1bcb 100644 --- a/src/specify_cli/presets/_resolver.py +++ b/src/specify_cli/presets/_resolver.py @@ -8,6 +8,7 @@ from .._utils import dump_frontmatter from ..extensions import ExtensionRegistry, normalize_priority +from ..file_helper import FileHelperError from ._manifest import VALID_PRESET_STRATEGIES, PresetManifest, PresetValidationError from ._registry import PresetRegistry @@ -163,10 +164,15 @@ def _get_all_extensions_by_priority(self) -> list[tuple[int, str, dict | None]]: Returns: List of (priority, ext_id, metadata_or_none) tuples sorted by priority. """ + try: + registry = ExtensionRegistry(self.extensions_dir) + except FileHelperError as exc: + raise PresetValidationError( + f"Invalid extension registry {self.extensions_dir / ExtensionRegistry.REGISTRY_FILE}: " + f"{exc}" + ) from exc if not self.extensions_dir.exists(): return [] - - registry = ExtensionRegistry(self.extensions_dir) # Fail closed on a corrupt registry. ExtensionRegistry._load() recovers # by normalizing an unreadable registry to an empty mapping, which would # otherwise cause the directory scan below to admit every on-disk diff --git a/tests/specify_cli/artifacts/test_command_list.py b/tests/specify_cli/artifacts/test_command_list.py index bcefd1d768..1c5bbe9c4e 100644 --- a/tests/specify_cli/artifacts/test_command_list.py +++ b/tests/specify_cli/artifacts/test_command_list.py @@ -19,6 +19,21 @@ class TestCommandList: + def test_linked_registry_emits_resolution_error(self, spec_kit_project, tmp_path, monkeypatch): + registry = spec_kit_project / ".specify" / "extensions" / ".registry" + try: + registry.symlink_to(tmp_path / "missing-registry") + except OSError: + pytest.skip("Symlink creation unavailable") + monkeypatch.chdir(spec_kit_project) + + result = CliRunner().invoke(app, ["artifact", "list", "--json"]) + + assert result.exit_code == 1 + assert isinstance(result.exception, SystemExit) + assert result.stdout == "" + assert json.loads(result.stderr) == {"error": "artifact resolution failed"} + def test_list_requires_json_flag( self, spec_kit_project: Path, monkeypatch: pytest.MonkeyPatch ): diff --git a/tests/specify_cli/artifacts/test_mcp_list.py b/tests/specify_cli/artifacts/test_mcp_list.py index 7c57883838..9300b72fea 100644 --- a/tests/specify_cli/artifacts/test_mcp_list.py +++ b/tests/specify_cli/artifacts/test_mcp_list.py @@ -704,6 +704,24 @@ def test_artifact_list_maps_corrupt_registry_error(spec_kit_project: Path): details={"project_directory": str(spec_kit_project)}, ) +def test_artifact_list_maps_linked_registry_error(spec_kit_project: Path, tmp_path): + registry = spec_kit_project / ".specify" / "extensions" / ".registry" + try: + registry.symlink_to(tmp_path / "missing-registry") + except OSError: + pytest.skip("Symlink creation unavailable") + + result = _run( + create_server(launch_directory=spec_kit_project).call_tool(TOOL_NAME, {}) + ) + + assert result.is_error is True + assert result.structured_content == _expected_error( + "artifact_resolution_failed", + "artifact resolution failed", + details={"project_directory": str(spec_kit_project)}, + ) + @pytest.mark.parametrize( "operation_failure", diff --git a/tests/specify_cli/events/test_events.py b/tests/specify_cli/events/test_events.py index 9c2d07aa6b..7361072fcf 100644 --- a/tests/specify_cli/events/test_events.py +++ b/tests/specify_cli/events/test_events.py @@ -1506,6 +1506,45 @@ def test_opencode_non_numeric_timeout_uses_default(self, tmp_path): class TestCommandRunner: """Test the core command/script resolution and runner.""" + @pytest.mark.parametrize("probe", ["disabled_ids", "manifest"]) + def test_linked_registry_cannot_execute_disabled_command( + self, tmp_path, monkeypatch, capsys, probe + ): + import specify_cli.events as events + + extensions_dir = tmp_path / ".specify" / "extensions" + commands = extensions_dir / "disabled" / "commands" + commands.mkdir(parents=True) + (commands / "speckit.disabled.boot.md").write_text( + "---\nscripts:\n sh: scripts/boot.sh\n---\n", encoding="utf-8" + ) + external = tmp_path / "external-registry" + external.write_text( + '{"extensions": {"disabled": {"enabled": false}}}', encoding="utf-8" + ) + try: + (extensions_dir / ".registry").symlink_to(external) + except OSError: + pytest.skip("Symlink creation unavailable") + execute = MagicMock() + execute.return_value.stdout = "" + execute.return_value.stderr = "" + execute.return_value.returncode = 0 + monkeypatch.setattr(events.subprocess, "run", execute) + monkeypatch.setattr(events, "_resolve_event_command_argv", lambda *args: ["unused"]) + if probe == "manifest": + monkeypatch.setattr(events, "_disabled_extension_ids", lambda _: set()) + + result = resolve_and_run_event_command( + "speckit.disabled.boot", "session_start", "{}", tmp_path + ) + + assert result == 2 + execute.assert_not_called() + output = capsys.readouterr() + assert output.out == "" + assert "symlink" in output.err + def test_run_command_not_found(self, tmp_path): code = resolve_and_run_event_command("nonexistent.command", "session_start", "{}", tmp_path) assert code == 0 # no-ops gracefully @@ -1897,6 +1936,124 @@ def test_dispatcher_inline_fallback_runs_script(self, tmp_path): assert out_file.exists(), f"inline fallback did not run script; stderr={result.stderr!r} rc={result.returncode}" assert out_file.read_text() == '{"tool_name":"x"}' + @pytest.mark.parametrize("runtime", ["missing-package", "stale-package"]) + @pytest.mark.parametrize("storage", [ + "plain", "missing", "registry-live", "registry-dangling", + "extensions-live", "extensions-dangling", "specify-live", + ]) + def test_dispatcher_checks_registry_before_delegation_or_fallback( + self, tmp_path, runtime, storage + ): + import subprocess + import sys + from specify_cli.events import _EVENTS_DISPATCHER_TEMPLATE + + project = tmp_path / "project" + specify_dir = project / ".specify" + extensions_dir = specify_dir / "extensions" + commands = extensions_dir / "example" / "commands" + commands.mkdir(parents=True) + marker = tmp_path / "executed" + imported = tmp_path / "imported" + (specify_dir / "init-options.json").write_text('{"script": "py"}', encoding="utf-8") + (commands / "speckit.example.boot.md").write_text( + "---\nscripts:\n py: boot.py\n---\n", encoding="utf-8" + ) + (extensions_dir / "example" / "boot.py").write_text( + f"from pathlib import Path\nPath({str(marker)!r}).write_text('inline')\n", + encoding="utf-8", + ) + dispatcher = project / EVENTS_DISPATCHER_REL + dispatcher.write_text(_EVENTS_DISPATCHER_TEMPLATE, encoding="utf-8") + registry = extensions_dir / ".registry" + if storage != "missing": + registry.write_text('{"extensions": {}}', encoding="utf-8") + external = tmp_path / "external" + try: + if storage.startswith("registry-"): + registry.unlink() + if storage == "registry-live": + external.write_text('{"extensions": {}}', encoding="utf-8") + registry.symlink_to(external) + elif storage.startswith("extensions-"): + if storage == "extensions-live": + extensions_dir.rename(external) + else: + import shutil + shutil.rmtree(extensions_dir) + extensions_dir.symlink_to(external, target_is_directory=True) + elif storage == "specify-live": + specify_dir.rename(external) + specify_dir.symlink_to(external, target_is_directory=True) + except OSError: + pytest.skip("Symlink creation unavailable") + + shadow = tmp_path / "shadow" + package = shadow / "specify_cli" + package.mkdir(parents=True) + (package / "__init__.py").write_text("", encoding="utf-8") + if runtime == "stale-package": + (package / "events.py").write_text( + "from pathlib import Path\n" + f"Path({str(imported)!r}).write_text('imported')\n" + "EVENT_SCRIPT_PATH_CONFINEMENT = True\n" + "def resolve_and_run_event_command(*args, **kwargs):\n" + f" Path({str(marker)!r}).write_text('delegated')\n" + " return 0\n", + encoding="utf-8", + ) + env = dict(os.environ, PYTHONPATH=str(shadow)) + result = subprocess.run( + [sys.executable, str(dispatcher), "speckit.example.boot", "session_start"], + input="{}", capture_output=True, text=True, encoding="utf-8", + env=env, cwd=project, timeout=30, + ) + + allowed = storage in ("plain", "missing") + assert result.returncode == (0 if allowed else 2), result.stderr + assert marker.exists() is allowed + assert imported.exists() is (allowed and runtime == "stale-package") + assert result.stdout == "" + if allowed: + assert result.stderr == "" + expected = "inline" if runtime == "missing-package" else "delegated" + assert marker.read_text(encoding="utf-8") == expected + else: + assert "symlink" in result.stderr + assert "Traceback" not in result.stderr + if storage == "registry-live": + assert external.read_text(encoding="utf-8") == '{"extensions": {}}' + if storage.endswith("dangling"): + assert not external.exists() + + @pytest.mark.parametrize("component", [".specify", "extensions", ".registry"]) + def test_dispatcher_rejects_registry_reparse_points(self, tmp_path, monkeypatch, component): + from types import SimpleNamespace + from specify_cli.events import _EVENTS_DISPATCHER_TEMPLATE + + extensions_dir = tmp_path / ".specify" / "extensions" + extensions_dir.mkdir(parents=True) + registry = extensions_dir / ".registry" + registry.write_text('{"extensions": {}}', encoding="utf-8") + reparse = { + ".specify": tmp_path / ".specify", + "extensions": extensions_dir, + ".registry": registry, + }[component] + lstat = Path.lstat + + def reparse_lstat(path, **kwargs): + entry = lstat(path, **kwargs) + if path == reparse: + return SimpleNamespace(st_mode=entry.st_mode, st_file_attributes=0x400) + return entry + + namespace = {"__name__": "dispatcher_test"} + exec(compile(_EVENTS_DISPATCHER_TEMPLATE, "", "exec"), namespace) + monkeypatch.setattr(Path, "lstat", reparse_lstat) + with pytest.raises(ValueError, match="reparse"): + namespace["_validate_registry_hierarchy"](tmp_path) + def test_dispatcher_ignores_stale_specify_cli_without_confinement(self, tmp_path): """A generated dispatcher must not delegate to an older specify_cli that lacks EVENT_SCRIPT_PATH_CONFINEMENT (uvx-init plus stale diff --git a/tests/specify_cli/extensions/test_commands.py b/tests/specify_cli/extensions/test_commands.py new file mode 100644 index 0000000000..aa4c7b57ca --- /dev/null +++ b/tests/specify_cli/extensions/test_commands.py @@ -0,0 +1,46 @@ +"""Shared extension CLI error reporting.""" + +import json + +import pytest +from typer.testing import CliRunner + +from specify_cli import app + + +@pytest.mark.parametrize("arguments", [ + ["add", "example"], + ["list"], + ["list", "--json"], + ["info", "example"], + ["info", "example", "--json"], + ["enable", "example"], + ["disable", "example"], + ["remove", "example", "--force"], + ["set-priority", "example", "10"], + ["update", "example"], + ["update"], +]) +def test_linked_registry_reports_cli_error(project_dir, tmp_path, monkeypatch, arguments): + registry = project_dir / ".specify" / "extensions" / ".registry" + registry.parent.mkdir(parents=True, exist_ok=True) + external = tmp_path / "external-registry" + external.write_text('{"extensions": {}}', encoding="utf-8") + try: + registry.symlink_to(external) + except OSError: + pytest.skip("Symlink creation unavailable") + with monkeypatch.context() as context: + context.chdir(project_dir) + result = CliRunner().invoke(app, ["extension", *arguments]) + + assert result.exit_code == 1, result.output + assert isinstance(result.exception, SystemExit), repr(result.exception) + if "--json" in arguments: + assert result.stdout == "" + assert "symlink" in json.loads(result.stderr)["error"] + else: + assert "Error:" in result.output + assert "symlink" in result.output + assert "Traceback" not in result.output + assert external.read_text(encoding="utf-8") == '{"extensions": {}}' diff --git a/tests/specify_cli/integrations/test_command_switch.py b/tests/specify_cli/integrations/test_command_switch.py index 8341c9adee..47cf9e3bd7 100644 --- a/tests/specify_cli/integrations/test_command_switch.py +++ b/tests/specify_cli/integrations/test_command_switch.py @@ -27,6 +27,98 @@ ) class TestIntegrationSwitch: + def test_switch_event_read_error_preserves_current(self, tmp_path, monkeypatch): + project = _init_project(tmp_path, "claude", "--events true") + before = { + path: path.read_bytes() + for path in ( + project / ".claude/skills/speckit-plan/SKILL.md", + project / ".specify/integration.json", + project / ".specify/init-options.json", + project / ".specify/integrations/claude.manifest.json", + ) + } + + def denied(*args, **kwargs): + raise PermissionError("event registry read denied") + + monkeypatch.setattr("specify_cli.events.resolve_events", denied) + result = _run_in_project(project, ["integration", "switch", "copilot"]) + + assert result.exit_code == 1, result.output + assert {path: path.read_bytes() for path in before} == before + assert not (project / ".github").exists() + output = " ".join(strip_ansi(result.output).split()) + assert "event registry read denied" in output + assert "No files were changed." in output + assert "Uninstalling current integration" not in output + + @pytest.mark.parametrize("linked_entry", ["extensions", ".registry"]) + @pytest.mark.parametrize("target_exists", [False, True]) + @pytest.mark.parametrize("force", [False, True]) + def test_switch_rejects_linked_registry_before_removing_current( + self, tmp_path, monkeypatch, linked_entry, target_exists, force + ): + project = _init_project(tmp_path, "claude", "--events true") + extensions_dir = project / ".specify" / "extensions" + extensions_dir.mkdir(parents=True, exist_ok=True) + external = tmp_path / "external" + link = extensions_dir if linked_entry == "extensions" else extensions_dir / ".registry" + if linked_entry == "extensions": + shutil.rmtree(extensions_dir) + if target_exists: + external.mkdir() + (external / "sentinel").write_bytes(b"unchanged") + else: + link.unlink(missing_ok=True) + if target_exists: + external.write_text('{"extensions": {}}', encoding="utf-8") + try: + link.symlink_to(external, target_is_directory=linked_entry == "extensions") + except OSError: + pytest.skip("Symlink creation unavailable") + + def snapshot(root): + entries = {} + for path in root.rglob("*"): + relative = path.relative_to(root).as_posix() + if path.is_symlink(): + entries[relative] = ("link", os.readlink(path)) + elif path.is_dir(): + entries[relative] = ("directory", None) + else: + entries[relative] = ("file", path.read_bytes()) + return entries + + before = snapshot(project) + external_before = ( + snapshot(external) if external.is_dir() + else external.read_bytes() if external.exists() else None + ) + monkeypatch.chdir(project) + arguments = [ + "integration", "switch", "copilot", "--script", "sh", + "--integration-options", "--events true", + ] + if force: + arguments.append("--force") + + result = runner.invoke(app, arguments, catch_exceptions=True) + + assert snapshot(project) == before, "Rejected switch changed files or metadata" + assert result.exit_code == 1, result.output + assert isinstance(result.exception, SystemExit), repr(result.exception) + output = strip_ansi(result.output) + assert "Error:" in output and "symlink" in output + assert "No files were changed." in output + assert "Uninstalling current integration" not in output + assert "Installing integration:" not in output + external_after = ( + snapshot(external) if external.is_dir() + else external.read_bytes() if external.exists() else None + ) + assert external_after == external_before + @pytest.mark.parametrize("installed", [False, True]) def test_switch_syncs_manifest_hash_for_preset_override( self, tmp_path, tasks_preset_src, installed diff --git a/tests/specify_cli/presets/test_manager.py b/tests/specify_cli/presets/test_manager.py index ff1738ab4c..266e1e27b6 100644 --- a/tests/specify_cli/presets/test_manager.py +++ b/tests/specify_cli/presets/test_manager.py @@ -769,6 +769,22 @@ def _boom(*args, **kwargs): manifest ) == [] + def test_linked_registry_does_not_abort_dependency_probe( + self, project_dir, temp_dir, valid_pack_data, tmp_path + ): + manifest = self._manifest(temp_dir, valid_pack_data, ["speckit-inventory"]) + extensions_dir = project_dir / ".specify" / "extensions" + extensions_dir.mkdir(parents=True, exist_ok=True) + external = tmp_path / "external-registry" + external.write_text('{"extensions": {}}', encoding="utf-8") + try: + (extensions_dir / ".registry").symlink_to(external) + except OSError: + pytest.skip("Symlink creation unavailable") + + assert PresetManager(project_dir).find_unmet_extension_dependencies(manifest) == [] + assert external.read_text(encoding="utf-8") == '{"extensions": {}}' + def test_exact_duplicate_declarations_warn_once( diff --git a/tests/specify_cli/presets/test_resolver.py b/tests/specify_cli/presets/test_resolver.py index 5c62cb9c9f..6785cb4421 100644 --- a/tests/specify_cli/presets/test_resolver.py +++ b/tests/specify_cli/presets/test_resolver.py @@ -7,6 +7,7 @@ import yaml from specify_cli.extensions import ExtensionRegistry +from specify_cli.file_helper import SymlinkDeniedError from specify_cli.presets import ( PresetManager, PresetRegistry, @@ -435,12 +436,34 @@ def test_resolve_fails_closed_when_registry_is_broken_symlink(self, project_dir) extensions_dir / "does-not-exist" ) - registry = ExtensionRegistry(extensions_dir) - assert registry.is_corrupt() + with pytest.raises(SymlinkDeniedError): + ExtensionRegistry(extensions_dir) resolver = PresetResolver(project_dir) with pytest.raises(PresetValidationError, match="Invalid extension registry"): resolver.resolve("custom-template") + @pytest.mark.parametrize("parent", [".specify", "extensions"]) + @pytest.mark.parametrize("target_exists", [False, True]) + def test_linked_registry_parent_does_not_fall_back_to_core( + self, project_dir, tmp_path, parent, target_exists + ): + target = tmp_path / "external" + if target_exists: + target.mkdir() + link = project_dir / ".specify" + if parent == "extensions": + link /= "extensions" + else: + import shutil + shutil.rmtree(link) + try: + link.symlink_to(target, target_is_directory=True) + except OSError: + pytest.skip("Symlink creation unavailable") + + with pytest.raises(PresetValidationError, match="Invalid extension registry"): + PresetResolver(project_dir)._get_all_extensions_by_priority() + def test_resolve_pack_over_extension(self, project_dir, pack_dir, temp_dir, valid_pack_data): """Test that pack templates take priority over extension templates.""" # Create extension with templates diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 13350e2acb..1a2ae83c68 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -22,6 +22,7 @@ from pathlib import Path from datetime import datetime, timezone from unittest.mock import MagicMock +from types import SimpleNamespace from tests.http_helpers import route_opener_open_through_urlopen # noqa: F401 from specify_cli import extensions as _ext_module @@ -43,6 +44,7 @@ normalize_priority, ) from specify_cli._utils import version_satisfies +from specify_cli.file_helper import SymlinkDeniedError, UnsupportedPathError # Minimal valid ZIP (empty end-of-central-directory record). Passes # zipfile.is_zipfile() so --from download tests exercise the content guard. @@ -1323,6 +1325,163 @@ def test_provides_entry_description_must_be_a_string(self, temp_dir, valid_manif class TestExtensionRegistry: """Test ExtensionRegistry operations.""" + @pytest.mark.parametrize("parent", [".specify", "extensions"]) + @pytest.mark.parametrize("leaf", ["missing", "directory"]) + def test_load_rejects_parent_reparse_before_leaf_recovery( + self, tmp_path, monkeypatch, parent, leaf + ): + extensions_dir = tmp_path / ".specify" / "extensions" + extensions_dir.mkdir(parents=True) + if leaf == "directory": + (extensions_dir / ".registry").mkdir() + reparse = tmp_path / ".specify" if parent == ".specify" else extensions_dir + lstat = Path.lstat + + def reparse_lstat(path, **kwargs): + entry = lstat(path, **kwargs) + if path == reparse: + return SimpleNamespace(st_mode=entry.st_mode, st_file_attributes=0x400) + return entry + + monkeypatch.setattr(Path, "lstat", reparse_lstat) + with pytest.raises(UnsupportedPathError, match="reparse"): + ExtensionRegistry(extensions_dir) + + def test_load_recovers_ordinary_registry_directory(self, tmp_path): + extensions_dir = tmp_path / ".specify" / "extensions" + (extensions_dir / ".registry").mkdir(parents=True) + + registry = ExtensionRegistry(extensions_dir) + + assert registry.keys() == set() + assert registry.is_corrupt() + + @pytest.mark.parametrize("linked_entry", [".specify", "extensions", ".registry"]) + @pytest.mark.parametrize("target_exists", [False, True]) + def test_load_rejects_symlink_hierarchy(self, tmp_path, linked_entry, target_exists): + project = tmp_path / "project" + extensions_dir = project / ".specify" / "extensions" + extensions_dir.mkdir(parents=True) + outside = tmp_path / "outside" + is_directory = linked_entry != ".registry" + if target_exists: + if is_directory: + outside.mkdir() + else: + outside.write_text('{"extensions": {"external": {}}}', encoding="utf-8") + link = { + ".specify": project / ".specify", + "extensions": extensions_dir, + ".registry": extensions_dir / ".registry", + }[linked_entry] + if link.is_dir(): + shutil.rmtree(link) + try: + link.symlink_to(outside, target_is_directory=is_directory) + except OSError: + pytest.skip("Symlink creation unavailable") + + with pytest.raises(SymlinkDeniedError): + ExtensionRegistry(extensions_dir) + + assert outside.exists() is target_exists + if target_exists and not is_directory: + assert outside.read_text(encoding="utf-8") == '{"extensions": {"external": {}}}' + + @pytest.mark.parametrize("linked_entry", [".specify", "extensions", ".registry"]) + @pytest.mark.parametrize("target_exists", [False, True]) + def test_save_rejects_symlink_hierarchy(self, tmp_path, monkeypatch, linked_entry, target_exists): + from specify_cli.integrations import _file_changes + + project = tmp_path / "project" + extensions_dir = project / ".specify" / "extensions" + extensions_dir.mkdir(parents=True) + registry = ExtensionRegistry(extensions_dir) + outside = tmp_path / "outside" + is_directory = linked_entry != ".registry" + if target_exists: + if is_directory: + outside.mkdir() + else: + outside.write_text("external content", encoding="utf-8") + link = { + ".specify": project / ".specify", + "extensions": extensions_dir, + ".registry": extensions_dir / ".registry", + }[linked_entry] + if link.is_dir(): + shutil.rmtree(link) + try: + link.symlink_to(outside, target_is_directory=is_directory) + except OSError: + pytest.skip("Symlink creation unavailable") + before = MagicMock() + after = MagicMock() + monkeypatch.setattr(_file_changes, "before_file_change", before) + monkeypatch.setattr(_file_changes, "after_file_change", after) + + with pytest.raises(SymlinkDeniedError): + registry.add("test-ext", {"version": "1.0.0"}) + + before.assert_not_called() + after.assert_not_called() + assert outside.exists() is target_exists + if target_exists: + if is_directory: + assert list(outside.iterdir()) == [] + else: + assert outside.read_text(encoding="utf-8") == "external content" + + @pytest.mark.parametrize("project_exists", [False, True]) + def test_registry_creates_missing_storage(self, tmp_path, project_exists): + project = tmp_path / "project" + if project_exists: + project.mkdir() + extensions_dir = project / ".specify" / "extensions" + registry = ExtensionRegistry(extensions_dir) + assert not extensions_dir.exists() + assert registry.data["extensions"] == {} + assert not registry.is_corrupt() + + registry.add("test-ext", {"version": "1.0.0"}) + + assert ExtensionRegistry(extensions_dir).get("test-ext")["version"] == "1.0.0" + + def test_registry_observers_wrap_successful_write(self, tmp_path, monkeypatch): + from specify_cli.integrations import _file_changes + + extensions_dir = tmp_path / ".specify" / "extensions" + registry = ExtensionRegistry(extensions_dir) + observed = [] + monkeypatch.setattr( + _file_changes, "before_file_change", + lambda path: observed.append(("before", path.exists())), + ) + monkeypatch.setattr( + _file_changes, "after_file_change", + lambda path: observed.append(("after", json.loads(path.read_text())["extensions"])), + ) + + registry.add("test-ext", {"version": "1.0.0"}) + + assert observed[0] == ("before", False) + assert observed[1][0] == "after" + assert observed[1][1]["test-ext"]["version"] == "1.0.0" + + def test_corruption_probe_rejects_replaced_registry(self, tmp_path): + extensions_dir = tmp_path / ".specify" / "extensions" + extensions_dir.mkdir(parents=True) + registry = ExtensionRegistry(extensions_dir) + outside = tmp_path / "outside" + outside.write_text('{"extensions": {}}', encoding="utf-8") + try: + registry.registry_path.symlink_to(outside) + except OSError: + pytest.skip("Symlink creation unavailable") + + assert registry.is_corrupt() + assert outside.read_text(encoding="utf-8") == '{"extensions": {}}' + def test_empty_registry(self, temp_dir): """Test creating a new empty registry.""" extensions_dir = temp_dir / "extensions" @@ -8922,6 +9081,22 @@ def _install(self, project_root, ext_id): # from ``ExtensionManager.remove(..., keep_config=True)``). ExtensionRegistry(extensions_dir).add(ext_id, {}) + def test_linked_registry_does_not_crash_config_read(self, tmp_path, monkeypatch): + extensions_dir = tmp_path / ".specify" / "extensions" + extensions_dir.mkdir(parents=True) + external = tmp_path / "external-registry" + external.write_text('{"extensions": {"git-hooks": {}}}', encoding="utf-8") + try: + (extensions_dir / ".registry").symlink_to(external) + except OSError: + pytest.skip("Symlink creation unavailable") + monkeypatch.setenv("SPECKIT_GIT_HOOKS_URL", "value") + + assert ConfigManager(tmp_path, "git")._get_env_config() == { + "hooks": {"url": "value"}, + } + assert external.read_text(encoding="utf-8") == '{"extensions": {"git-hooks": {}}}' + def test_sibling_owns_longer_prefix_env(self, tmp_path, monkeypatch): """SPECKIT_GIT_HOOKS_URL belongs to git-hooks when co-installed with git.""" self._install(tmp_path, "git") @@ -9274,6 +9449,7 @@ def test_scaffold_config_rejects_symlinked_config_root(self, tmp_path): from specify_cli.extensions import ExtensionManager project = tmp_path / "project" project.mkdir() + manager = ExtensionManager(project) outside = tmp_path / "outside" outside.mkdir() (project / ".specify").symlink_to(outside, target_is_directory=True) @@ -9287,7 +9463,6 @@ def test_scaffold_config_rejects_symlinked_config_root(self, tmp_path): }]) (ext_dir / "config-template.yml").write_text("setting: default") - manager = ExtensionManager(project) deployed, skipped, failed = manager.scaffold_config("test-ext") assert deployed == [] diff --git a/tests/test_file_helper.py b/tests/test_file_helper.py index a04b9e0a34..e2fafe0675 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 @@ -59,6 +60,33 @@ def test_plain_operations(root, allow): files.delete("nested", recursive=True) assert list(root.iterdir()) == [] +@pytest.mark.parametrize("allow", [False, True]) +def test_stat_reads_metadata_without_creating_missing_paths(root, allow): + files = FileHelper(root, allow_symlinks=allow) + (root / "file").write_bytes(b"content") + + assert files.stat("file").st_size == 7 + assert stat.S_ISREG(files.stat("file").st_mode) + assert stat.S_ISDIR(files.stat(root).st_mode) + with pytest.raises(FileNotFoundError): + files.stat("missing/child") + assert not (root / "missing").exists() + with pytest.raises(PathEscapeError): + files.stat("../outside") + + +def test_stat_follows_only_allowed_contained_links(root, link): + target = root / "file" + target.write_bytes(b"content") + alias = link(root / "alias", target) + + assert FileHelper(root, allow_symlinks=True).stat(alias) == target.stat() + external = root.parent / "outside" + external.write_bytes(b"external") + outside_alias = link(root / "outside-alias", external) + with pytest.raises(PathEscapeError): + FileHelper(root, allow_symlinks=True).stat(outside_alias) + @pytest.mark.parametrize("allow", [False, True]) def test_overwrite_text_preserves_inode_and_native_text_behavior(root, allow): @@ -164,7 +192,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", ["stat", "read_bytes", "write_bytes", "create_bytes", "mkdir", "delete"]) @pytest.mark.parametrize("kind", ["leaf", "parent", "dangling"]) def test_deny_rejects_accessed_links_without_mutation(root, link, operation, kind): target = root / "target" @@ -960,6 +988,8 @@ def directory_reparse(path, **kwargs): monkeypatch.setattr(Path, "lstat", directory_reparse) files = FileHelper(root, allow_symlinks=allow) for operation in ( + lambda: files.stat(junction / "keep"), + lambda: files.stat(junction / "missing"), lambda: files.read_bytes(junction / "keep"), lambda: files.write_bytes(junction / "keep", b"bad"), lambda: files.overwrite_text(junction / "keep", "bad"), @@ -1036,6 +1066,8 @@ def test_native_windows_junctions_never_traversed_or_deleted(root, junction, all alias = junction(tree / "z-junction", target) files = FileHelper(root, allow_symlinks=allow) for operation in ( + lambda: files.stat(alias / "keep"), + lambda: files.stat(alias / "missing"), lambda: files.read_text(alias / "keep"), lambda: files.write_text(alias / "keep", "bad"), lambda: files.create_text(alias / "new", "bad"), diff --git a/tests/test_resolve_template_python_parity.py b/tests/test_resolve_template_python_parity.py index 9696358459..912edb6330 100644 --- a/tests/test_resolve_template_python_parity.py +++ b/tests/test_resolve_template_python_parity.py @@ -4,6 +4,7 @@ import json import os +import shutil import subprocess import sys import time @@ -15,6 +16,8 @@ from tests.conftest import requires_bash from tests.parity_helpers import ( HAS_POWERSHELL, + WINDOWS_POWERSHELL, + _bash_posix_path, bash_cmd, clean_env, install_composition_stack, @@ -31,6 +34,147 @@ SCRIPT = "resolve-template" TEMPLATE = "constitution-template" +@pytest.mark.parametrize("runtime", [ + "python", + pytest.param("bash", marks=requires_bash), + pytest.param("powershell", marks=pytest.mark.skipif( + not HAS_POWERSHELL, reason="PowerShell unavailable", + )), + pytest.param("windows-powershell51", marks=pytest.mark.skipif( + not WINDOWS_POWERSHELL, reason="Windows PowerShell 5.1 required", + )), +]) +@pytest.mark.parametrize("mode", ["path", "content"]) +@pytest.mark.parametrize("storage", [ + "ordinary", "missing", "registry-live", "registry-dangling", + "extensions-live", "extensions-dangling", "specify-live", "extensions-junction", +]) +def test_template_helpers_reject_linked_registry_storage(tmp_path, runtime, mode, storage): + repo = make_repo(tmp_path) + install_scripts(repo, SCRIPT) + core = repo / ".specify" / "templates" / f"{TEMPLATE}.md" + core.parent.mkdir(parents=True) + core.write_bytes(b"# Core\n") + extensions = repo / ".specify" / "extensions" + templates = extensions / "example" / "templates" + templates.mkdir(parents=True) + (templates / f"{TEMPLATE}.md").write_bytes(b"# External content\n") + registry = extensions / ".registry" + registry.write_text('{"extensions": {"example": {"enabled": false}}}', encoding="utf-8") + external = tmp_path / "external" + if storage == "missing": + shutil.rmtree(extensions) + elif storage.startswith("registry-"): + registry.unlink() + if storage == "registry-live": + external.write_text('{"extensions": {}}', encoding="utf-8") + try: + registry.symlink_to(external) + except OSError: + pytest.skip("Symlink creation unavailable") + elif storage.startswith("extensions-"): + if storage == "extensions-dangling": + shutil.rmtree(extensions) + else: + extensions.rename(external) + if storage == "extensions-junction": + if os.name != "nt": + pytest.skip("Native Windows junction required") + created = subprocess.run( + ["cmd", "/c", "mklink", "/J", str(extensions), str(external)], + capture_output=True, text=True, check=False, + ) + assert created.returncode == 0, created.stderr + else: + try: + extensions.symlink_to(external, target_is_directory=True) + except OSError: + pytest.skip("Symlink creation unavailable") + elif storage == "specify-live": + (repo / ".specify").rename(external) + try: + (repo / ".specify").symlink_to(external, target_is_directory=True) + except OSError: + pytest.skip("Symlink creation unavailable") + + if runtime == "python": + function = "resolve_template" if mode == "path" else "resolve_template_content" + command = [ + sys.executable, "-c", + ( + "import sys; from pathlib import Path; sys.path.insert(0, sys.argv[1]); " + "import common; print(getattr(common, sys.argv[2])(sys.argv[3], Path(sys.argv[4])))" + ), + str(repo / ".specify/scripts/python"), function, TEMPLATE, str(repo), + ] + elif runtime == "bash": + function = "resolve_template" if mode == "path" else "resolve_template_content" + command = [ + "bash", "-c", 'source "$1"; "$2" "$3" "$4"', "bash", + _bash_posix_path(repo / ".specify/scripts/bash/common.sh"), + function, TEMPLATE, _bash_posix_path(repo), + ] + else: + function = "Resolve-Template" if mode == "path" else "Resolve-TemplateContent" + common = str(repo / ".specify/scripts/powershell/common.ps1").replace("'", "''") + root = str(repo).replace("'", "''") + executable = WINDOWS_POWERSHELL if runtime == "windows-powershell51" else ps_cmd(repo, SCRIPT)[0] + command = [ + executable, "-NoProfile", "-Command", + ( + f"$ErrorActionPreference = 'Stop'; . '{common}'; " + f"[Console]::Out.Write(({function} -TemplateName '{TEMPLATE}' -RepoRoot '{root}'))" + ), + ] + + result = run(command, repo, timeout=30) + if storage in ("ordinary", "missing"): + assert result.returncode == 0, result.stderr + assert result.stderr == "" + assert ("constitution-template.md" if mode == "path" else "# Core") in result.stdout + else: + assert result.returncode != 0, result.stdout + assert result.stdout == "" + assert any(word in result.stderr.lower() for word in ("symlink", "reparse", "registry")) + from specify_cli.presets import PresetResolver, PresetValidationError + with pytest.raises(PresetValidationError): + PresetResolver(repo).resolve(TEMPLATE) + +@requires_bash +@pytest.mark.parametrize("function", ["resolve_template", "resolve_template_content"]) +def test_bash_without_python_rejects_linked_extension_storage(tmp_path, function): + repo = make_repo(tmp_path) + install_scripts(repo, SCRIPT) + external = tmp_path / "external" + templates = external / "example" / "templates" + templates.mkdir(parents=True) + (templates / f"{TEMPLATE}.md").write_bytes(b"# External content\n") + try: + (repo / ".specify" / "extensions").symlink_to(external, target_is_directory=True) + except OSError: + pytest.skip("Symlink creation unavailable") + + result = run([ + "bash", "-c", + 'source "$1"; _python3_command() { return 1; }; "$2" "$3" "$4"', + "bash", _bash_posix_path(repo / ".specify/scripts/bash/common.sh"), + function, TEMPLATE, _bash_posix_path(repo), + ], repo) + + assert result.returncode != 0 + assert result.stdout == "" + assert "symlink" in result.stderr.lower() + + +@pytest.mark.skipif(not WINDOWS_POWERSHELL, reason="Windows PowerShell 5.1 required") +def test_native_template_validation_uses_windows_powershell51(tmp_path): + result = run( + [WINDOWS_POWERSHELL, "-NoProfile", "-Command", "$PSVersionTable.PSVersion.ToString()"], + tmp_path, + ) + assert result.returncode == 0, result.stderr + assert result.stdout.strip().startswith("5.1.") + def test_python_json_preserves_constitution_on_legacy_stdout(tmp_path: Path) -> None: repo = make_repo(tmp_path)