From c65ca4cff1770154136e05059fbabaa9a698e422 Mon Sep 17 00:00:00 2001 From: Markus Date: Thu, 24 Sep 2026 09:03:08 +0200 Subject: [PATCH 1/2] feat(workflows): install custom step types from local dirs and archives (#4695) `specify workflow step add` gains `--dev ` and `--from ` alongside the existing catalog source. All three converge on a new `step/installer.py` domain module that owns package validation (shape, symlink/special-file rejection, 512-file/50 MiB limits), same-filesystem staging with revalidation, atomic commit, `--force` replacement, and source-kind-only registry provenance. Direct URLs require a default-deny trust prompt before any request. Docs document the local-authoring flow and the deferred bundle-local limitation. Assisted-by: opencode (model: deepseek-v4.1-flash, autonomous) --- docs/reference/bundles.md | 2 + docs/reference/workflows.md | 234 ++++++- src/specify_cli/workflows/step/_helpers.py | 120 +--- src/specify_cli/workflows/step/command_add.py | 588 +++++++++------- .../workflows/step/command_info.py | 20 + src/specify_cli/workflows/step/installer.py | 612 +++++++++++++++++ .../workflows/step/test_command_add.py | 565 +++++++++++++++- .../workflows/step/test_command_info.py | 44 ++ .../workflows/step/test_installer.py | 626 ++++++++++++++++++ 9 files changed, 2458 insertions(+), 353 deletions(-) create mode 100644 src/specify_cli/workflows/step/installer.py create mode 100644 tests/specify_cli/workflows/step/test_installer.py diff --git a/docs/reference/bundles.md b/docs/reference/bundles.md index 139028a377..f95cc7fdba 100644 --- a/docs/reference/bundles.md +++ b/docs/reference/bundles.md @@ -81,6 +81,8 @@ The source may also be a bundle directory or `.zip` artifact. Refresh uses the s A local bundle source supplies the manifest, not its component payloads. Components resolved through catalogs still require network access to refresh, even when already installed. Add `--offline` only when the components being installed or refreshed ship with Spec Kit; otherwise the command reports which component needs network access. Re-run without `--offline` to fetch that component through its catalog. +> **Step payloads resolve through the step catalog only.** A bundle's `provides.steps` entries still resolve exclusively through the active step catalogs. Bundle-local `steps//` payloads and relative `provides.steps[].source` overrides are **not** resolved in this release, so a step declared that way cannot be installed offline. To ship a step with a bundle today, publish it to a step catalog the bundle's users can reach. + ## Update Bundles ```bash diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index 811ab4ebf4..9de667b401 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -1,6 +1,6 @@ # Workflows -Workflows automate multi-step Spec-Driven Development processes — chaining commands, prompts, shell steps, and human checkpoints into repeatable sequences. They support conditional logic, loops, fan-out/fan-in, and can be paused and resumed from the exact point of interruption. +Workflows automate multi-step Spec-Driven Development processes — chaining commands, prompts, shell steps, and human checkpoints into repeatable sequences. They support conditional logic, loops, fan-out/fan-in, composition (running another installed workflow as a scoped subtree), and can be paused and resumed from the exact point of interruption. ## Run a Workflow @@ -335,7 +335,6 @@ When an installed workflow is refreshed or reinstalled, project overlays in `.sp - An overlay that targets a step id that does not exist in the base workflow will raise a validation error when the workflow is resolved. - Overlays cannot target steps added by other overlays. - Overlays cannot add new inputs or change the input schema of the base workflow. - ## Update Workflows ```bash @@ -540,9 +539,240 @@ specify workflow run speckit -i spec="Build a kanban board with drag-and-drop ta | `do-while` | Execute at least once, then loop on condition | | `fan-out` | Dispatch a step for each item in a list | | `fan-in` | Aggregate results from a fan-out step | +| `workflow` | Run an installed workflow as a scoped subtree | > **Security note:** a `shell` step runs a local command with **your** privileges. There is no capability sandbox — `requires` is an advisory pre-condition block (spec-kit version, integrations), not a runtime gate, so it does **not** restrict what a step can do. In particular there is no `requires.permissions` capability gate: it is rejected by validation precisely because it would imply a sandbox that does not exist. Review any catalog or downloaded workflow before running it, and use a `gate` step to require explicit approval before sensitive or destructive shell commands. +### Custom step packages + +Custom step types are installed with `specify workflow step`. A step is a +directory package containing metadata and executable Python: + +```text +my-step/ +├── step.yml # required, at the package root +├── __init__.py # required, at the package root +└── helpers.py # optional nested modules and data files +``` + +`step.yml` declares the step's identity. `step.type_key` must exactly match the +`` passed on the command line — the ID is never inferred from package +content: + +```yaml +step: + type_key: my-step + name: My Step + version: 0.1.0 + author: you + description: What this step does +``` + +`__init__.py` must define a `StepBase` subclass whose `type_key` matches: + +```python +from specify_cli.workflows.base import StepBase, StepResult + + +class MyStep(StepBase): + type_key = "my-step" + + def execute(self, config, context): + return StepResult(output={"ok": True}) +``` + +#### Install from a local directory + +```bash +specify workflow step add my-step --dev /path/to/my-step +``` + +`--dev` takes a **directory** (not an archive, not a bare `step.yml`) that is a +complete package. This needs no catalog, server, or network, which makes it the +supported local-authoring loop: + +```bash +specify workflow step add my-step --dev ./my-step +specify workflow step list +specify workflow step info my-step +# edit ./my-step, then replace the installed copy: +specify workflow step add my-step --dev ./my-step --force +specify workflow step remove my-step +``` + +#### Install from an archive URL + +```bash +specify workflow step add my-step --from https://example.com/my-step.zip +``` + +`--from` accepts a `.zip`, `.tar.gz`, or `.tgz` archive (a bare `step.yml` +URL is **not** a package). The archive may place `step.yml` and `__init__.py` +at its root or under exactly one top-level directory; unrelated top-level +siblings are rejected. Because a step package contains executable Python, a +direct URL install shows a default-deny trust confirmation before any network +request; declining cancels with no request and no error. HTTPS is required +(HTTP is permitted only for loopback hosts), redirects must remain secure, and +downloads are size-bounded. + +#### Install from the catalog + +```bash +specify workflow step add my-step +``` + +Catalog installs resolve individual file URLs from the active step catalogs and +then go through the same validation and commit path as `--dev` and `--from`. +Discovery-only catalogs cannot be installed from. + +#### Replacement and force + +```bash +specify workflow step add my-step --dev ./my-step --force +specify workflow step add my-step --from https://example.com/my-step.zip --force +``` + +`--force` first stages and validates the replacement before touching the +existing installation, and can replace both a registered install and a leftover +unregistered directory. Validation and staging failures leave the previous +package untouched. If a replacement commit fails after the previous package is +removed — removing the old directory, publishing the new one, or writing the +registry — the installation is left incomplete: rerun the command with the +original source and `--force` to reinstall. No automatic rollback is attempted. + +#### Package validation + +Every source is validated identically before anything is committed: + +- `step.yml` and `__init__.py` must be regular, non-symlink files at the package + root. +- The package tree is copied recursively (relative imports, nested helper + modules, and data files are supported). A symlinked package root, any + descendant symlink, and any filesystem object that is not a regular file or + directory are rejected — including inside excluded directories. +- `.git`, `__pycache__`, and `.DS_Store` entries are excluded from the copy and + from the limits. +- A package may contain at most **512 files** and **50 MiB** in total. +- `__init__.py` is **not imported** during installation; it is loaded only when + the step runs. + +> **Security note:** Installing a custom step runs its Python with **your** +> privileges. Only install step packages from sources you trust. + +#### Listing, running, and removing + +Installed custom steps appear in `specify workflow step list` and are loaded +automatically by `workflow add`, `workflow run`, and `workflow resume`. Remove +one with: + +```bash +specify workflow step remove my-step +``` + +#### Registry provenance + +Each installed step records only the *kind* of its source — `catalog` +(optionally with the catalog name), `local`, or `url`. Local paths and source +URLs are never persisted. `specify workflow step info ` shows the source. + +#### Bundle-local limitation + +A bundle's `provides.steps` still resolves only through the active step +catalogs. Bundle-local `steps//` payloads and relative +`provides.steps[].source` overrides are **not** resolved in this release, so +such steps are not installable offline. See the [Bundles reference](bundles.md). + +### Workflow composition (`type: workflow`) + +A `workflow` step runs an installed workflow as a **scoped subtree of the +current run** — there is one run, one run directory, and one process. The +included workflow behaves like a function call: values cross the boundary only +through its declared `inputs` and `outputs`. + +```yaml +steps: + - id: triage + type: prompt + prompt: "Select the workflow to run" + + - id: run-selected + type: workflow + workflow: "{{ steps.triage.output.stdout }}" + input: + report: "{{ inputs.report }}" + slug: "{{ inputs.slug }}" +``` + +| Field | Required | Description | +| ---------- | -------- | ----------- | +| `workflow` | yes | Installed workflow ID, or an expression evaluated in the caller's scope. The resolved value must be a valid ID of a registered, installed, and enabled workflow. Literal IDs are validated at definition time. | +| `input` | no | Mapping of the target's declared input names to values evaluated in the caller's scope. An undeclared name is rejected. Defaults, required, type, and enum rules apply. | + +`type: workflow` is an engine facility (like `fan-out`), not a custom-step API. +The engine owns the nested scope tree; custom steps still receive only a +`StepContext`. + +#### Scope isolation + +The included workflow receives a separate expression scope: + +- `inputs` contains only the resolved, declared, and validated mapped inputs. +- `steps` contains only the included workflow's own step results. +- Caller inputs and caller step results are **not** visible unless explicitly + passed through the `input` mapping. +- Project root, integration defaults, and the run ID remain available as + execution infrastructure. + +#### Declared outputs + +An included workflow exposes values back to its caller only through a top-level +`outputs` block. Each entry requires a `value` expression evaluated in the +included workflow's local scope once it completes: + +```yaml +outputs: + result: + value: "{{ steps.fix.output.stdout }}" + tested: + value: "{{ steps.test.output.exit_code == 0 }}" +``` + +The caller reads them from the workflow step's output: + +```yaml +"{{ steps.run-selected.output.result }}" +``` + +Output names must be safe lowercase identifiers and cannot use the reserved +names `workflow`, `status`, `error`, `aborted`, `integration`, `model`, +`options`, or `input`. A whole expression preserves its resolved type; +interpolation mixed with text produces a string. Paused, failed, and aborted +scopes do not evaluate outputs. + +#### Lifecycle and failure handling + +The workflow step reports the aggregate outcome of its subtree: all required +steps complete → `completed`; an included step pauses → the run pauses; an +included failure (unhandled) → the run fails; an included gate abort → the run +aborts (`output.aborted: true`). `continue_on_error: true` on the workflow step +lets the caller continue past an otherwise unhandled included failure; it never +overrides an abort or bypasses a pause. + +#### Resume and composition limits + +The resolved target, its composed definition snapshot (including overlays), and +the validated inputs are persisted with the run. On resume the engine reuses the +snapshot and resumes at the included scope's local step index; it does not +re-resolve the target. Editing an installed workflow affects new invocations, +not a scope already bound within a persisted run. `workflow resume --input` +updates the **root** workflow's inputs; a composing workflow forwards them by +mapping them into the child's declared inputs. + +Recursive composition is allowed, but cycles are rejected by path (`A -> B -> A` +fails while `A -> B -> D` and `A -> C -> D` is a legal diamond). Composition is +limited to 16 included levels; the root is depth 0 and entering depth 17 is +rejected. + ### Per-Step Integration Configuration Command steps may pass structured runtime configuration to integrations that diff --git a/src/specify_cli/workflows/step/_helpers.py b/src/specify_cli/workflows/step/_helpers.py index 45250ceffc..34ada2f306 100644 --- a/src/specify_cli/workflows/step/_helpers.py +++ b/src/specify_cli/workflows/step/_helpers.py @@ -1,107 +1,47 @@ -"""Shared validation helpers for workflow step commands.""" +"""Shared validation helpers for workflow step commands. + +This module preserves the CLI-coupled ``*_or_exit`` entry points used by the +registered step commands. The behavior now lives in +:mod:`specify_cli.workflows.step.installer`; these wrappers print the shared +error prefix and exit, while the domain module stays CLI-independent. +""" from __future__ import annotations from .. import _commands as cli - -# Custom step packages contain executable Python, metadata, and optional helper -# files downloaded one-by-one rather than as an archive. Mirror the archive -# ceilings so a catalog cannot turn individually valid files into an unbounded -# aggregate download. -_MAX_STEP_PACKAGE_FILES = 512 -_MAX_STEP_PACKAGE_BYTES = 50 * 1024 * 1024 # 50 MiB - -_RESERVED_STEP_IDS: frozenset[str] = frozenset({".cache", "step-registry.json"}) - -_WINDOWS_RESERVED_NAMES: frozenset[str] = frozenset( - { - "con", - "prn", - "aux", - "nul", - "com1", - "com2", - "com3", - "com4", - "com5", - "com6", - "com7", - "com8", - "com9", - "lpt1", - "lpt2", - "lpt3", - "lpt4", - "lpt5", - "lpt6", - "lpt7", - "lpt8", - "lpt9", - } +from .installer import ( + _MAX_STEP_PACKAGE_BYTES, + _MAX_STEP_PACKAGE_FILES, + StepInstallError, + resolve_steps_base_dir, + validate_step_id, ) -_WINDOWS_INVALID_CHARS: frozenset[str] = frozenset('<>:"|?*') +__all__ = [ + "_MAX_STEP_PACKAGE_BYTES", + "_MAX_STEP_PACKAGE_FILES", + "StepInstallError", + "resolve_steps_base_dir", + "validate_step_id", +] def _validate_step_id_or_exit(step_id: str) -> None: """Validate that ``step_id`` is a single safe path component. - Rejects empty strings, whitespace-only strings, leading/trailing whitespace, - path separators, ``.``/``..`` components, dotfile prefixes, reserved names, - Windows-invalid filename characters, trailing dots/spaces, and Windows - reserved device names. Exits with code 1 on failure. + Exits with code 1 on failure. """ - # Strip the stem (before first dot) for Windows reserved-name check - stem = step_id.split(".")[0].lower() if step_id else "" - if ( - not step_id - or not step_id.strip() - or step_id != step_id.strip() - or "/" in step_id - or "\\" in step_id - or step_id in (".", "..") - or step_id.startswith(".") - or step_id.endswith(".") - or step_id.endswith(" ") - or step_id.lower() in _RESERVED_STEP_IDS - or stem in _WINDOWS_RESERVED_NAMES - or any(c in _WINDOWS_INVALID_CHARS for c in step_id) - or any(ord(c) < 32 for c in step_id) - ): - cli.console.print( - f"[red]Error:[/red] Invalid step id '{step_id}': must be a single safe " - "path component (no separators, no leading dot, not a reserved name, " - "no invalid filename characters)" - ) - raise cli.typer.Exit(1) + try: + validate_step_id(step_id) + except StepInstallError as exc: + cli.console.print(f"[red]Error:[/red] {exc}") + raise cli.typer.Exit(1) from exc def _resolve_steps_base_dir_or_exit(project_root: cli.Path) -> cli.Path: """Resolve .specify/workflows/steps while refusing symlinked parent directories.""" - project_root_resolved = project_root.resolve() - steps_base_dir_unresolved = project_root / ".specify" / "workflows" / "steps" - - current = project_root - for part in (".specify", "workflows", "steps"): - current = current / part - if current.is_symlink(): - cli.console.print( - f"[red]Error:[/red] Refusing to use symlinked step directory '{current}'" - ) - raise cli.typer.Exit(1) - if current.exists() and not current.is_dir(): - cli.console.print( - f"[red]Error:[/red] Step directory path is not a directory: '{current}'" - ) - raise cli.typer.Exit(1) - - steps_base_dir = steps_base_dir_unresolved.resolve() try: - steps_base_dir.relative_to(project_root_resolved) - except ValueError: - cli.console.print( - f"[red]Error:[/red] Step directory escapes project root: '{steps_base_dir}'" - ) - raise cli.typer.Exit(1) - - return steps_base_dir + return resolve_steps_base_dir(project_root) + except StepInstallError as exc: + cli.console.print(f"[red]Error:[/red] {exc}") + raise cli.typer.Exit(1) from exc diff --git a/src/specify_cli/workflows/step/command_add.py b/src/specify_cli/workflows/step/command_add.py index 7a3a3b8cad..3aba60bce3 100644 --- a/src/specify_cli/workflows/step/command_add.py +++ b/src/specify_cli/workflows/step/command_add.py @@ -1,109 +1,288 @@ -"""Command handler for ``specify workflow step add``.""" +"""Command handler for ``specify workflow step add``. + +The registered handler stays a thin orchestrator: it parses options, validates +them, and dispatches to a per-source private helper (catalog, ``--dev`` local +directory, and ``--from`` archive URL). All three sources converge on +``step/installer.py``'s single validation + staged-commit path. +""" from __future__ import annotations +from typing import Annotated + from .. import _commands as cli +from . import _helpers as step_helpers from . import step_app -from . import _helpers as step_helpers +def _cleanup_download_tmp_path(tmp_path: cli.Path | None) -> None: + """Best-effort unlink of a partially-downloaded step archive temp file. -@step_app.command("add") -def workflow_step_add( - step_id: str = cli.typer.Argument(..., help="Step type ID from catalog"), -): - """Install a custom step type from the step catalog.""" - from .catalog import ( - StepCatalog, - StepCatalogError, - StepRegistry, - StepValidationError, + A cleanup ``OSError`` must never replace/mask whatever error or interrupt is + already propagating -- warn about it and keep going. + """ + if tmp_path is None: + return + try: + tmp_path.unlink(missing_ok=True) + except OSError as cleanup_exc: + cli.console.print( + "[yellow]Warning:[/yellow] Could not remove temporary " + f"step download file: {cli._escape_markup(str(cleanup_exc))} " + f"(path: {cli._escape_markup(str(tmp_path))})" + ) + + +def _print_installed(step_id: str, entry: dict) -> None: + step_name = entry.get("name") or step_id + cli.console.print(f"[green]✓[/green] Step type '{step_name}' ({step_id}) installed") + cli.console.print( + " Use [cyan]specify workflow step list[/cyan] to verify the installation." ) - project_root = cli._require_specify_project() + +def _install_from_dev( + project_root: cli.Path, step_id: str, dev: str, *, force: bool +) -> None: + """Install a complete step package from a local directory.""" + from . import installer + + dev_path = cli.Path(dev).expanduser() + if dev_path.is_symlink(): + raise installer.StepInstallError( + f"Refusing to install from a symlinked source directory: '{dev_path}'" + ) + if not dev_path.is_dir(): + raise installer.StepInstallError( + "--dev source must be a directory containing step.yml and " + f"__init__.py: '{dev_path}'" + ) + + entry = installer.install_step_package( + project_root, step_id, dev_path, source="local", force=force + ) + _print_installed(step_id, entry) + + +def _install_from_url( + project_root: cli.Path, step_id: str, from_url: str, *, force: bool +) -> None: + """Install a step package archive from a direct URL.""" + import tempfile + from urllib.parse import urlparse + + from rich.panel import Panel + + from specify_cli.authentication.github_http import ( + resolve_github_release_asset_api_url as _resolve_gh_asset, + ) + from specify_cli.authentication.http import ( + github_provider_hosts as _github_provider_hosts, + ) + from specify_cli.authentication.http import open_url as _open_url + + from . import installer + + try: + parsed = urlparse(from_url) + hostname = parsed.hostname + _ = parsed.port + except ValueError: + raise installer.StepInstallError( + f"Invalid URL: {cli._escape_markup(from_url)}" + ) from None + if not hostname: + raise installer.StepInstallError( + f"Invalid URL: {cli._escape_markup(from_url)}" + ) + if not cli.is_https_or_localhost_http(from_url): + raise installer.StepInstallError( + "URL must use HTTPS for security. HTTP is only allowed for " + "loopback URLs." + ) + + # Reject before the trust prompt and before any network request. + installer.check_installable(project_root, step_id, force=force) + + # Prompt BEFORE any request (and before any spinner) so the user can see + # and answer it; a declined prompt issues no request and exits 0. + cli.console.print() + cli.console.print( + Panel( + "[bold]You are installing a workflow step type directly from an " + "external URL.\nA step package contains executable Python.[/bold]\n\n" + f"URL: {cli._escape_markup(from_url)}\n\n" + "Only install step packages from sources you trust.", + title="[bold yellow]⚠ Untrusted Source[/bold yellow]", + border_style="yellow", + padding=(1, 2), + ) + ) + cli.console.print() + if not cli.typer.confirm("Continue with installation?", default=False): + cli.console.print("Cancelled") + raise cli.typer.Exit(0) + + download_url = from_url + extra_headers = None + tmp_path: cli.Path | None = None + try: + resolved_url = _resolve_gh_asset( + from_url, + _open_url, + timeout=30, + github_hosts=_github_provider_hosts(), + redirect_validator=cli._reject_insecure_download_redirect, + ) + if resolved_url: + download_url = resolved_url + extra_headers = {"Accept": "application/octet-stream"} + + with _open_url( + download_url, + timeout=30, + extra_headers=extra_headers, + redirect_validator=cli._reject_insecure_download_redirect, + ) as resp: + final_url = resp.geturl() + if not cli.is_https_or_localhost_http(final_url): + raise installer.StepInstallError( + f"URL redirected to non-HTTPS: {cli._escape_markup(final_url)}" + ) + content_type = ( + resp.getheader("Content-Type") + if hasattr(resp, "getheader") + else None + ) + archive_format = ( + cli.archive_format_from_name(final_url) + or cli.archive_format_from_name(from_url) + or cli.archive_format_from_content_type(content_type) + ) + if archive_format is None: + raise installer.StepInstallError( + "URL does not reference a supported archive " + "(.zip, .tar.gz, or .tgz)" + ) + downloaded = cli.read_response_limited( + resp, + error_type=ValueError, + label="step archive download", + ) + + with tempfile.NamedTemporaryFile( + suffix=cli.archive_suffix(archive_format), delete=False + ) as tmp: + tmp_path = cli.Path(tmp.name) + tmp.write(downloaded) + + with tempfile.TemporaryDirectory( + prefix="speckit-step-archive-" + ) as extract_dir: + extracted_root = cli.Path(extract_dir) + # safe_extract_archive re-detects and confirms the archive bytes. + cli.safe_extract_archive( + tmp_path, + extracted_root, + source_name=final_url, + content_type=content_type, + ) + package_root = installer.resolve_package_root(extracted_root) + entry = installer.install_step_package( + project_root, + step_id, + package_root, + source="url", + force=force, + ) + except cli.typer.Exit: + raise + except installer.StepInstallError: + raise + except Exception as exc: + raise installer.StepInstallError( + f"Failed to install step from URL: {cli._escape_markup(str(exc))}" + ) from exc + finally: + _cleanup_download_tmp_path(tmp_path) + + _print_installed(step_id, entry) + + +def _install_from_catalog(project_root: cli.Path, step_id: str, *, force: bool) -> None: + """Install a step package from the step catalog. + + The catalog fetch (URL/derivation/count preflight) stays a catalog concern; + the materialized files are then handed to the shared installer. + """ + import tempfile + + from . import installer + from .catalog import StepCatalog, StepCatalogError catalog = StepCatalog(project_root) try: info = catalog.get_step_info(step_id) except StepCatalogError as exc: - cli.console.print(f"[red]Error:[/red] {exc}") - raise cli.typer.Exit(1) + raise installer.StepInstallError(str(exc)) from exc if not info: - cli.console.print( - f"[red]Error:[/red] Step type '{step_id}' not found in catalog" + raise installer.StepInstallError( + f"Step type '{step_id}' not found in catalog" ) - raise cli.typer.Exit(1) if not info.get("_install_allowed", True): cli.console.print( - f"[yellow]Warning:[/yellow] Step type '{step_id}' is from a discovery-only catalog" + f"[yellow]Warning:[/yellow] Step type '{step_id}' is from a " + "discovery-only catalog" ) cli.console.print("Direct installation is not enabled for this catalog source.") raise cli.typer.Exit(1) - # Reject step IDs that collide with built-in step types - from .. import STEP_REGISTRY as _step_reg - - if step_id in _step_reg: - cli.console.print( - f"[red]Error:[/red] Step type '{step_id}' conflicts with a built-in step type" - ) - raise cli.typer.Exit(1) - - # Reject if already installed - registry = StepRegistry(project_root) - if registry.is_installed(step_id): - cli.console.print( - f"[red]Error:[/red] Step type '{step_id}' is already installed. " - "Remove it first with: [cyan]specify workflow step remove " - f"{step_id}[/cyan]" - ) - raise cli.typer.Exit(1) + # Reject built-in collisions and duplicates before any download. + installer.check_installable(project_root, step_id, force=force) declared_step_yml_url = info.get("step_yml_url") if declared_step_yml_url is not None and not isinstance(declared_step_yml_url, str): - cli.console.print( - f"[red]Error:[/red] Catalog entry for '{step_id}' has a malformed " - "step.yml URL; expected a non-empty string" + raise installer.StepInstallError( + f"Catalog entry for '{step_id}' has a malformed step.yml URL; " + "expected a non-empty string" ) - raise cli.typer.Exit(1) step_yml_url = declared_step_yml_url or info.get("url") if step_yml_url is None or ( isinstance(step_yml_url, str) and not step_yml_url.strip() ): - cli.console.print(f"[red]Error:[/red] Catalog entry for '{step_id}' has no URL") - raise cli.typer.Exit(1) + raise installer.StepInstallError( + f"Catalog entry for '{step_id}' has no URL" + ) if not isinstance(step_yml_url, str): - cli.console.print( - f"[red]Error:[/red] Catalog entry for '{step_id}' has a malformed " - "step.yml URL; expected a non-empty string" + raise installer.StepInstallError( + f"Catalog entry for '{step_id}' has a malformed step.yml URL; " + "expected a non-empty string" ) - raise cli.typer.Exit(1) - # Derive __init__.py URL: replace trailing step.yml with __init__.py - # or use explicit init_url if provided. + # Derive __init__.py URL: replace trailing step.yml with __init__.py or use + # explicit init_url if provided. init_url = info.get("init_url") if init_url is not None and (not isinstance(init_url, str) or not init_url.strip()): - cli.console.print( - f"[red]Error:[/red] Catalog entry for '{step_id}' has a malformed " - "__init__.py URL; expected a non-empty string" + raise installer.StepInstallError( + f"Catalog entry for '{step_id}' has a malformed __init__.py URL; " + "expected a non-empty string" ) - raise cli.typer.Exit(1) if not init_url: if step_yml_url.endswith("step.yml"): init_url = step_yml_url[: -len("step.yml")] + "__init__.py" else: - cli.console.print( - f"[red]Error:[/red] Cannot derive __init__.py URL from '{step_yml_url}'. " - "Catalog entry should provide 'init_url' or a 'url' ending in 'step.yml'." + raise installer.StepInstallError( + f"Cannot derive __init__.py URL from '{step_yml_url}'. " + "Catalog entry should provide 'init_url' or a 'url' ending in " + "'step.yml'." ) - raise cli.typer.Exit(1) # Preflight the declared file count before creating a staging directory or - # issuing any request. The two required files are always part of the package; - # duplicate declarations for them in extra_files are ignored below and do - # not count twice. + # issuing any request. The two required files are always part of the + # package; duplicate declarations for them in extra_files are ignored below + # and do not count twice. extra_files = info.get("extra_files") if extra_files is not None and not isinstance(extra_files, dict): cli.console.print( @@ -126,12 +305,11 @@ def _is_required_package_file(rel_path: object) -> bool: 1 for rel_path in (extra_files or {}) if not _is_required_package_file(rel_path) ) package_file_count = 2 + declared_extra_count - if package_file_count > step_helpers._MAX_STEP_PACKAGE_FILES: - cli.console.print( - f"[red]Error:[/red] Step package declares {package_file_count} files, " - f"exceeding the {step_helpers._MAX_STEP_PACKAGE_FILES}-file limit" + if package_file_count > installer._MAX_STEP_PACKAGE_FILES: + raise installer.StepInstallError( + f"Step package declares {package_file_count} files, exceeding the " + f"{installer._MAX_STEP_PACKAGE_FILES}-file limit" ) - raise cli.typer.Exit(1) from specify_cli.authentication.http import open_url as _open_url @@ -146,231 +324,127 @@ def _safe_fetch(url: str) -> bytes: raise ValueError(f"Redirect to non-HTTPS URL: {final_url}") return cli._read_response_within_limit(resp) - step_helpers._validate_step_id_or_exit(step_id) - - steps_base_dir = step_helpers._resolve_steps_base_dir_or_exit(project_root) - step_dir = (steps_base_dir / step_id).resolve() - # Defense-in-depth: ensure the resolved directory is a direct child of - # steps_base_dir even after symlink resolution. - try: - rel_parts = step_dir.relative_to(steps_base_dir).parts - except ValueError: - cli.console.print(f"[red]Error:[/red] Invalid step id '{step_id}'") - raise cli.typer.Exit(1) - if rel_parts != (step_id,): - cli.console.print(f"[red]Error:[/red] Invalid step id '{step_id}'") - raise cli.typer.Exit(1) - - import shutil - import tempfile - - # Refuse if step_dir already exists (e.g. leftover from a previous failed/manual - # install that wasn't registered). The user should remove it before retrying. - if step_dir.exists(): - cli.console.print( - f"[red]Error:[/red] Step directory already exists at '{step_dir}'. " - f"Remove it manually or use: [cyan]specify workflow step remove {step_id}[/cyan]" - ) - raise cli.typer.Exit(1) - - # Create steps_base_dir now so the staging temp dir is on the same filesystem, - # enabling a truly atomic os.rename() below. - try: - steps_base_dir.mkdir(parents=True, exist_ok=True) - tmp_path = cli.Path( - tempfile.mkdtemp(prefix="speckit_step_tmp_", dir=steps_base_dir) - ) - except OSError as exc: - cli.console.print( - f"[red]Error:[/red] Failed to create staging directory: {exc}" - ) - raise cli.typer.Exit(1) - try: + with tempfile.TemporaryDirectory(prefix="speckit-step-package-") as package_tmp: + package_dir = cli.Path(package_tmp) try: step_yml_content = _safe_fetch(step_yml_url) init_py_content = _safe_fetch(init_url) except Exception as exc: - cli.console.print(f"[red]Error:[/red] Failed to download step files: {exc}") - raise cli.typer.Exit(1) + raise installer.StepInstallError( + f"Failed to download step files: {exc}" + ) from exc package_bytes = len(step_yml_content) + len(init_py_content) - if package_bytes > step_helpers._MAX_STEP_PACKAGE_BYTES: - cli.console.print( - f"[red]Error:[/red] Step package exceeds the " - f"{step_helpers._MAX_STEP_PACKAGE_BYTES}-byte total size limit" - ) - raise cli.typer.Exit(1) - - # Validate step.yml - try: - import yaml as _yaml - - step_yml_text = step_yml_content.decode("utf-8") - # ``safe_load`` returns None for BOTH an empty document and an - # explicit null scalar (``null``, ``~``, ``NULL``), so it cannot - # tell them apart on its own. ``compose`` yields no node only for - # a genuinely empty document. - node = _yaml.compose(step_yml_text) - meta = _yaml.safe_load(step_yml_text) - is_empty_document = node is None or ( - meta is None - and isinstance(node, _yaml.nodes.ScalarNode) - and node.value == "" - and node.start_mark.index == node.end_mark.index - ) - except Exception as exc: - cli.console.print(f"[red]Error:[/red] Invalid step.yml: {exc}") - raise cli.typer.Exit(1) - - # Do NOT coerce with ``or {}`` here: that also turns a FALSY non-mapping - # (top-level ``[]``, ``false``, ``0``, ``''``, or an explicit ``null``) - # into ``{}`` and silently bypasses this shape check, surfacing the - # unrelated "missing 'step.type_key'" error below instead of the real - # problem. Only a genuinely empty document defaults to ``{}``. - if meta is None and is_empty_document: - meta = {} - elif not isinstance(meta, dict): - cli.console.print("[red]Error:[/red] step.yml must be a YAML mapping") - raise cli.typer.Exit(1) - - step_meta = meta.get("step", {}) - if not isinstance(step_meta, dict): - cli.console.print( - "[red]Error:[/red] step.yml 'step' field must be a mapping" + if package_bytes > installer._MAX_STEP_PACKAGE_BYTES: + raise installer.StepInstallError( + f"Step package exceeds the " + f"{installer._MAX_STEP_PACKAGE_BYTES}-byte total size limit" ) - raise cli.typer.Exit(1) - type_key = step_meta.get("type_key", "") - if not type_key: - cli.console.print( - "[red]Error:[/red] step.yml missing 'step.type_key' field" - ) - raise cli.typer.Exit(1) - - if type_key != step_id: - cli.console.print( - f"[red]Error:[/red] step.yml type_key ({type_key!r}) does not match " - f"catalog ID ({step_id!r})" - ) - raise cli.typer.Exit(1) - # Write the two required files. try: - (tmp_path / "step.yml").write_bytes(step_yml_content) - (tmp_path / "__init__.py").write_bytes(init_py_content) + (package_dir / "step.yml").write_bytes(step_yml_content) + (package_dir / "__init__.py").write_bytes(init_py_content) except OSError as exc: - cli.console.print( - f"[red]Error:[/red] Failed to write step files to staging directory: {exc}" - ) - raise cli.typer.Exit(1) - - # Optionally download additional package files declared in the catalog entry - # (e.g. helper modules). Each entry in ``extra_files`` is a mapping of - # relative-path → URL. step.yml and __init__.py are ignored here (already - # written). Paths are validated to stay within the step package directory to - # prevent path-traversal attacks. + raise installer.StepInstallError( + f"Failed to write step files to staging directory: {exc}" + ) from exc + + # Optionally download additional package files declared in the catalog + # entry (e.g. helper modules). Each entry in ``extra_files`` is a mapping + # of relative-path → URL. Paths are validated to stay within the step + # package directory to prevent path-traversal attacks. for rel_path, file_url in (extra_files or {}).items(): if not isinstance(rel_path, str) or not rel_path.strip(): - cli.console.print( - "[red]Error:[/red] Catalog entry 'extra_files' contains an " - "empty or non-string path key" + raise installer.StepInstallError( + "Catalog entry 'extra_files' contains an empty or non-string " + "path key" ) - raise cli.typer.Exit(1) if _is_required_package_file(rel_path): continue # already written above - # Reject dot-path segments ('', '.', '..') that would refer to the - # package directory itself (IsADirectoryError) or escape it. - rel_parts = cli.Path(rel_path).parts - if not rel_parts or any(seg in ("", ".", "..") for seg in rel_parts): - cli.console.print( - f"[red]Error:[/red] extra_files path '{rel_path}' is not a " - "valid relative file path" + path_parts = cli.Path(rel_path).parts + if not path_parts or any(seg in ("", ".", "..") for seg in path_parts): + raise installer.StepInstallError( + f"extra_files path '{rel_path}' is not a valid relative file path" ) - raise cli.typer.Exit(1) if not isinstance(file_url, str) or not file_url.strip(): - cli.console.print( - f"[red]Error:[/red] extra_files entry '{rel_path}' has an " - "empty or non-string URL" + raise installer.StepInstallError( + f"extra_files entry '{rel_path}' has an empty or non-string URL" ) - raise cli.typer.Exit(1) - # Resolve both destination and base to handle any symlinks in tmp_path itself, - # ensuring the traversal check is robust even on non-canonical paths. - resolved_base = tmp_path.resolve() - dest = (tmp_path / rel_path).resolve() + resolved_base = package_dir.resolve() + dest = (package_dir / rel_path).resolve() try: dest.relative_to(resolved_base) except ValueError: - cli.console.print( - f"[red]Error:[/red] extra_files path '{rel_path}' is outside " - "the step package directory" - ) - raise cli.typer.Exit(1) + raise installer.StepInstallError( + f"extra_files path '{rel_path}' is outside the step package " + "directory" + ) from None try: file_content = _safe_fetch(file_url) except Exception as exc: - cli.console.print( - f"[red]Error:[/red] Failed to download extra file '{rel_path}': {exc}" - ) - raise cli.typer.Exit(1) + raise installer.StepInstallError( + f"Failed to download extra file '{rel_path}': {exc}" + ) from exc package_bytes += len(file_content) - if package_bytes > step_helpers._MAX_STEP_PACKAGE_BYTES: - cli.console.print( - f"[red]Error:[/red] Step package exceeds the " - f"{step_helpers._MAX_STEP_PACKAGE_BYTES}-byte total size limit" + if package_bytes > installer._MAX_STEP_PACKAGE_BYTES: + raise installer.StepInstallError( + f"Step package exceeds the " + f"{installer._MAX_STEP_PACKAGE_BYTES}-byte total size limit" ) - raise cli.typer.Exit(1) try: dest.parent.mkdir(parents=True, exist_ok=True) dest.write_bytes(file_content) except OSError as exc: - cli.console.print( - f"[red]Error:[/red] Failed to write extra file '{rel_path}': {exc}" - ) - raise cli.typer.Exit(1) + raise installer.StepInstallError( + f"Failed to write extra file '{rel_path}': {exc}" + ) from exc - # Atomically rename the staging directory to the final location. - # Both paths are under steps_base_dir (same filesystem), so os.rename() - # is atomic on POSIX and won't leave a partially-written directory at - # step_dir on failure. - try: - cli.os.rename(tmp_path, step_dir) - except OSError as exc: - cli.console.print( - f"[red]Error:[/red] Failed to install step '{step_id}': {exc}" - ) - raise cli.typer.Exit(1) - finally: - # Clean up if the rename hasn't moved tmp_path yet (i.e. on any failure). - shutil.rmtree(tmp_path, ignore_errors=True) + entry = installer.install_step_package( + project_root, + step_id, + package_dir, + source="catalog", + catalog_name=info.get("_catalog_name", ""), + catalog_metadata=info, + force=force, + ) - step_name = info.get("name") or step_id - step_version = info.get("version") or step_meta.get("version") or "0.0.0" + _print_installed(step_id, entry) - # Register in step registry - registry = StepRegistry(project_root) - try: - registry.add( - step_id, - { - "name": step_name, - "version": step_version, - "description": info.get( - "description", step_meta.get("description", "") - ), - "author": info.get("author", step_meta.get("author", "")), - "source": "catalog", - "catalog_name": info.get("_catalog_name", ""), - "type_key": type_key, - }, + +@step_app.command("add") +def workflow_step_add( + step_id: str = cli.typer.Argument(..., help="Step type ID"), + dev: Annotated[str | None, cli.typer.Option("--dev", help="Install from a local step package directory")] = None, + from_url: Annotated[str | None, cli.typer.Option("--from", help="Install from a .zip/.tar.gz/.tgz archive URL")] = None, + force: Annotated[bool, cli.typer.Option("--force", help="Replace an existing installation")] = False, +): + """Install a custom step type from the catalog, a local directory, or a URL.""" + from . import installer + + project_root = cli._require_specify_project() + + if dev is not None and from_url is not None: + cli.console.print( + "[red]Error:[/red] --dev and --from are mutually exclusive" ) - except StepValidationError as exc: - # Roll back the just-installed directory so the system isn't left with - # an unregistered step package on disk after a registry write failure - # (e.g. read-only filesystem, permission denied). - shutil.rmtree(step_dir, ignore_errors=True) - cli.console.print(f"[red]Error:[/red] {exc}") + raise cli.typer.Exit(1) + if dev is not None and not dev.strip(): + cli.console.print("[red]Error:[/red] --dev value must not be empty") + raise cli.typer.Exit(1) + if from_url is not None and not from_url.strip(): + cli.console.print("[red]Error:[/red] --from value must not be empty") raise cli.typer.Exit(1) - cli.console.print(f"[green]✓[/green] Step type '{step_name}' ({step_id}) installed") - cli.console.print( - " Use [cyan]specify workflow step list[/cyan] to verify the installation." - ) + step_helpers._validate_step_id_or_exit(step_id) + + try: + if dev is not None: + _install_from_dev(project_root, step_id, dev, force=force) + elif from_url is not None: + _install_from_url(project_root, step_id, from_url, force=force) + else: + _install_from_catalog(project_root, step_id, force=force) + except installer.StepInstallError as exc: + cli.console.print(f"[red]Error:[/red] {exc}") + raise cli.typer.Exit(1) from exc diff --git a/src/specify_cli/workflows/step/command_info.py b/src/specify_cli/workflows/step/command_info.py index c98072d8c0..6f70a0403b 100644 --- a/src/specify_cli/workflows/step/command_info.py +++ b/src/specify_cli/workflows/step/command_info.py @@ -6,6 +6,23 @@ from . import step_app +def _format_source(installed_meta: dict) -> str: + """Render a registry entry's provenance as a human-facing source label. + + Local and URL installs deliberately store no path/URL, so only the source + kind is shown. + """ + source = installed_meta.get("source") + if source == "catalog": + catalog_name = installed_meta.get("catalog_name") + if catalog_name: + return f"catalog ({cli._escape_markup(str(catalog_name))})" + return "catalog" + if source in ("local", "url"): + return str(source) + return "" + + @step_app.command("info") def workflow_step_info( step_id: str = cli.typer.Argument(..., help="Step type ID"), @@ -46,6 +63,9 @@ def workflow_step_info( f" Description: " f"{cli._escape_markup(str(installed_meta['description']))}" ) + source_label = _format_source(installed_meta) + if source_label: + cli.console.print(f" Source: {source_label}") cli.console.print(" [green]Installed[/green]") return diff --git a/src/specify_cli/workflows/step/installer.py b/src/specify_cli/workflows/step/installer.py new file mode 100644 index 0000000000..1e37e2b57c --- /dev/null +++ b/src/specify_cli/workflows/step/installer.py @@ -0,0 +1,612 @@ +"""Domain install/validation for custom workflow step packages. + +This module owns the source-independent behavior shared by every +``specify workflow step add`` source mode (catalog, ``--dev`` local directory, +and ``--from`` archive URL): step-id and base-directory validation, package +shape/symlink/limit validation, staging, atomic commit, and registry +provenance. It is deliberately CLI-independent -- it never prints and never +raises ``typer.Exit``. Callers receive :class:`StepInstallError` and decide how +to surface it. +""" + +from __future__ import annotations + +import os +import shutil +import stat +import tempfile +from collections.abc import Mapping +from pathlib import Path +from typing import Any + +import yaml + +# Custom step packages contain executable Python, metadata, and optional helper +# files. These ceilings apply uniformly to catalog, local, and archive sources. +_MAX_STEP_PACKAGE_FILES = 512 +_MAX_STEP_PACKAGE_BYTES = 50 * 1024 * 1024 # 50 MiB + +# Files/dirs never copied into (or counted as part of) an installed step +# package. Mirrors ``bundles/packager.py`` ``EXCLUDE_NAMES``. +EXCLUDE_NAMES: frozenset[str] = frozenset({".git", "__pycache__", ".DS_Store"}) + +# Prefix for the private same-filesystem working directory created beneath the +# steps base directory. The leading dot keeps it out of the way, and because it +# contains only a ``staged/`` child (never ``step.yml``/``__init__.py`` at its +# root) the runtime loader never mistakes it for an installable package. +_WORK_DIR_PREFIX = ".speckit-step-install-" + +_RESERVED_STEP_IDS: frozenset[str] = frozenset({".cache", "step-registry.json"}) + +_WINDOWS_RESERVED_NAMES: frozenset[str] = frozenset( + { + "con", + "prn", + "aux", + "nul", + "com1", + "com2", + "com3", + "com4", + "com5", + "com6", + "com7", + "com8", + "com9", + "lpt1", + "lpt2", + "lpt3", + "lpt4", + "lpt5", + "lpt6", + "lpt7", + "lpt8", + "lpt9", + } +) + +_WINDOWS_INVALID_CHARS: frozenset[str] = frozenset('<>:"|?*') + + +class StepInstallError(Exception): + """User-facing step package install/validation failure.""" + + +# --------------------------------------------------------------------------- +# Step id + base directory validation +# --------------------------------------------------------------------------- + + +def validate_step_id(step_id: str) -> None: + """Validate that ``step_id`` is a single safe path component. + + Rejects empty strings, whitespace-only strings, leading/trailing + whitespace, path separators, ``.``/``..`` components, dotfile prefixes, + reserved names, Windows-invalid filename characters, trailing dots/spaces, + and Windows reserved device names. + """ + stem = step_id.split(".")[0].lower() if step_id else "" + if ( + not step_id + or not step_id.strip() + or step_id != step_id.strip() + or "/" in step_id + or "\\" in step_id + or step_id in (".", "..") + or step_id.startswith(".") + or step_id.endswith((".", " ")) + or step_id.lower() in _RESERVED_STEP_IDS + or stem in _WINDOWS_RESERVED_NAMES + or any(c in _WINDOWS_INVALID_CHARS for c in step_id) + or any(ord(c) < 32 for c in step_id) + ): + raise StepInstallError( + f"Invalid step id '{step_id}': must be a single safe " + "path component (no separators, no leading dot, not a reserved name, " + "no invalid filename characters)" + ) + + +def resolve_steps_base_dir(project_root: Path) -> Path: + """Resolve ``.specify/workflows/steps`` refusing symlinked parent dirs.""" + project_root = Path(project_root) + project_root_resolved = project_root.resolve() + steps_base_dir_unresolved = project_root / ".specify" / "workflows" / "steps" + + current = project_root + for part in (".specify", "workflows", "steps"): + current = current / part + if current.is_symlink(): + raise StepInstallError( + f"Refusing to use symlinked step directory '{current}'" + ) + if current.exists() and not current.is_dir(): + raise StepInstallError( + f"Step directory path is not a directory: '{current}'" + ) + + steps_base_dir = steps_base_dir_unresolved.resolve() + try: + steps_base_dir.relative_to(project_root_resolved) + except ValueError: + raise StepInstallError( + f"Step directory escapes project root: '{steps_base_dir}'" + ) from None + + return steps_base_dir + + +def _resolve_step_dir(steps_base_dir: Path, step_id: str) -> Path: + """Return the canonical destination directory for ``step_id``.""" + step_dir = steps_base_dir / step_id + try: + rel_parts = step_dir.relative_to(steps_base_dir).parts + except ValueError: + raise StepInstallError(f"Invalid step id '{step_id}'") from None + if rel_parts != (step_id,): + raise StepInstallError(f"Invalid step id '{step_id}'") + return step_dir + + +def _reject_unsafe_destination(step_dir: Path) -> None: + """Refuse a symlink (including dangling) or non-directory destination.""" + if step_dir.is_symlink(): + raise StepInstallError( + f"Refusing to install step through a symlinked path: '{step_dir}'" + ) + if step_dir.exists() and not step_dir.is_dir(): + raise StepInstallError( + f"Step install path exists but is not a directory: '{step_dir}'" + ) + + +# --------------------------------------------------------------------------- +# Package shape + safety validation +# --------------------------------------------------------------------------- + + +def _walk_package_tree(package_dir: Path): + """Yield ``(path, is_dir, excluded)`` for every descendant of *package_dir*. + + Descends into excluded directories so a symlink or special file hiding + inside ``.git``/``__pycache__`` is still rejected, but never follows a + symlink. Raises :class:`StepInstallError` on any symlink or object that is + neither a regular file nor a directory. + """ + + def _walk(current: Path, excluded_prefix: bool): + try: + entries = sorted(os.scandir(current), key=lambda entry: entry.name) + except OSError as exc: + raise StepInstallError( + f"Failed to read step package directory '{current}': {exc}" + ) from exc + for entry in entries: + try: + mode = entry.stat(follow_symlinks=False).st_mode + except OSError as exc: + raise StepInstallError( + f"Failed to inspect step package entry '{entry.path}': {exc}" + ) from exc + path = Path(entry.path) + if stat.S_ISLNK(mode): + raise StepInstallError(f"Step package contains symlink: {path}") + excluded = excluded_prefix or entry.name in EXCLUDE_NAMES + if stat.S_ISDIR(mode): + yield path, True, excluded + yield from _walk(path, excluded) + elif stat.S_ISREG(mode): + yield path, False, excluded + else: + raise StepInstallError( + f"Step package contains unsupported file: {path}" + ) + + yield from _walk(package_dir, False) + + +def _parse_step_metadata(step_yml_text: str, step_id: str) -> dict[str, Any]: + """Parse and validate ``step.yml``, returning the ``step`` mapping.""" + try: + # ``safe_load`` returns None for BOTH an empty document and an explicit + # null scalar (``null``, ``~``, ``NULL``), so it cannot tell them apart + # on its own. ``compose`` yields no node only for a genuinely empty + # document. + node = yaml.compose(step_yml_text) + meta = yaml.safe_load(step_yml_text) + is_empty_document = node is None or ( + meta is None + and isinstance(node, yaml.nodes.ScalarNode) + and node.value == "" + and node.start_mark.index == node.end_mark.index + ) + except Exception as exc: + raise StepInstallError(f"Invalid step.yml: {exc}") from exc + + # Do NOT coerce with ``or {}`` here: that also turns a FALSY non-mapping + # (top-level ``[]``, ``false``, ``0``, ``''``, or an explicit ``null``) + # into ``{}`` and silently bypasses this shape check. Only a genuinely + # empty document defaults to ``{}``. + if meta is None and is_empty_document: + meta = {} + elif not isinstance(meta, dict): + raise StepInstallError("step.yml must be a YAML mapping") + + step_meta = meta.get("step", {}) + if not isinstance(step_meta, dict): + raise StepInstallError("step.yml 'step' field must be a mapping") + type_key = step_meta.get("type_key", "") + if not type_key: + raise StepInstallError("step.yml missing 'step.type_key' field") + if type_key != step_id: + raise StepInstallError( + f"step.yml type_key ({type_key!r}) does not match step ID ({step_id!r})" + ) + return step_meta + + +def validate_step_package(package_dir: Path, step_id: str) -> dict[str, Any]: + """Validate a materialized step package directory. + + Returns the validated ``step.yml`` ``step`` mapping. Raises + :class:`StepInstallError` on any shape, symlink, limit, path, or identity + violation. Never imports or executes ``__init__.py``. + """ + package_dir = Path(package_dir) + + if package_dir.is_symlink(): + raise StepInstallError( + f"Refusing to install from a symlinked package directory: '{package_dir}'" + ) + if not package_dir.is_dir(): + raise StepInstallError(f"Step package directory not found: '{package_dir}'") + + for required in ("step.yml", "__init__.py"): + required_path = package_dir / required + if required_path.is_symlink(): + raise StepInstallError( + f"Step package '{required}' must be a regular file, not a symlink" + ) + if not required_path.is_file(): + raise StepInstallError( + f"Step package is missing required file '{required}' at its root" + ) + + retained_files = 0 + retained_bytes = 0 + for path, is_dir, excluded in _walk_package_tree(package_dir): + if is_dir or excluded: + continue + retained_files += 1 + try: + retained_bytes += path.lstat().st_size + except OSError as exc: + raise StepInstallError( + f"Failed to inspect step package file '{path}': {exc}" + ) from exc + + if retained_files > _MAX_STEP_PACKAGE_FILES: + raise StepInstallError( + f"Step package contains {retained_files} files, exceeding the " + f"{_MAX_STEP_PACKAGE_FILES}-file limit" + ) + if retained_bytes > _MAX_STEP_PACKAGE_BYTES: + raise StepInstallError( + f"Step package exceeds the {_MAX_STEP_PACKAGE_BYTES}-byte total " + "size limit" + ) + + try: + step_yml_text = (package_dir / "step.yml").read_text(encoding="utf-8") + except (OSError, UnicodeDecodeError) as exc: + raise StepInstallError(f"Invalid step.yml: {exc}") from exc + + return _parse_step_metadata(step_yml_text, step_id) + + +def resolve_package_root(extracted_root: Path) -> Path: + """Resolve a root-level or single-nested step package directory.""" + extracted_root = Path(extracted_root) + root_manifest = extracted_root / "step.yml" + if root_manifest.is_file() and not root_manifest.is_symlink(): + return extracted_root + try: + entries = list(extracted_root.iterdir()) + except OSError as exc: + raise StepInstallError( + f"Failed to inspect archive contents: {exc}" + ) from exc + if len(entries) == 1: + candidate = entries[0] + candidate_manifest = candidate / "step.yml" + if ( + candidate.is_dir() + and not candidate.is_symlink() + and candidate_manifest.is_file() + and not candidate_manifest.is_symlink() + ): + return candidate + raise StepInstallError( + "archive must contain step.yml at its root or in exactly one top-level " + "directory" + ) + + +# --------------------------------------------------------------------------- +# Collision / duplicate preflight +# --------------------------------------------------------------------------- + + +def _reject_builtin_collision(step_id: str) -> None: + from .. import BUILTIN_STEP_TYPES + + if step_id in BUILTIN_STEP_TYPES: + raise StepInstallError( + f"Step type '{step_id}' conflicts with a built-in step type" + ) + + +def _check_duplicate( + registry: Any, step_id: str, step_dir: Path, *, force: bool +) -> None: + if force: + return + if registry.is_installed(step_id): + raise StepInstallError( + f"Step type '{step_id}' is already installed. Remove it first with: " + f"[cyan]specify workflow step remove {step_id}[/cyan]" + ) + if step_dir.exists(): + raise StepInstallError( + f"Step directory already exists at '{step_dir}'. Remove it manually " + f"or use: [cyan]specify workflow step remove {step_id}[/cyan]" + ) + + +def check_installable(project_root: Path, step_id: str, *, force: bool = False) -> Path: + """Advisory preflight shared by all sources. + + Validates the id, base directory, built-in collision, and + duplicate/orphan-destination state without touching the package. The CLI + uses this to reject before a download; :func:`install_step_package` + re-runs the same checks as defense-in-depth. + """ + from .catalog import StepRegistry + + validate_step_id(step_id) + steps_base_dir = resolve_steps_base_dir(project_root) + step_dir = _resolve_step_dir(steps_base_dir, step_id) + _reject_unsafe_destination(step_dir) + _reject_builtin_collision(step_id) + registry = StepRegistry(project_root) + _check_duplicate(registry, step_id, step_dir, force=force) + return step_dir + + +# --------------------------------------------------------------------------- +# Staging + commit +# --------------------------------------------------------------------------- + + +def _build_entry( + step_id: str, + step_meta: Mapping[str, Any], + *, + source: str, + catalog_name: str, + catalog_metadata: Mapping[str, Any] | None, +) -> dict[str, Any]: + catalog_metadata = catalog_metadata or {} + entry: dict[str, Any] = { + "name": catalog_metadata.get("name") + or step_meta.get("name") + or step_id, + "version": catalog_metadata.get("version") + or step_meta.get("version") + or "0.0.0", + "description": catalog_metadata.get( + "description", step_meta.get("description", "") + ), + "author": catalog_metadata.get("author", step_meta.get("author", "")), + "type_key": step_meta["type_key"], + "source": source, + } + if source == "catalog": + entry["catalog_name"] = catalog_name + return entry + + +def _copy_package_tree(source_dir: Path, target_dir: Path) -> None: + """Recursively copy *source_dir* into *target_dir*, skipping excludes. + + Refuses to follow a symlink encountered mid-copy so a source swapped after + validation cannot smuggle external content into the staged package. + """ + + def _copy(current: Path, destination: Path) -> None: + try: + destination.mkdir(parents=True, exist_ok=True) + except OSError as exc: + raise StepInstallError( + f"Failed to stage step package: {exc}" + ) from exc + try: + entries = sorted(os.scandir(current), key=lambda entry: entry.name) + except OSError as exc: + raise StepInstallError( + f"Failed to stage step package: {exc}" + ) from exc + for entry in entries: + if entry.name in EXCLUDE_NAMES: + continue + try: + mode = entry.stat(follow_symlinks=False).st_mode + except OSError as exc: + raise StepInstallError(f"Failed to stage step package: {exc}") from exc + target = destination / entry.name + if stat.S_ISLNK(mode): + raise StepInstallError( + f"Step package contains symlink: {entry.path}" + ) + if stat.S_ISDIR(mode): + _copy(Path(entry.path), target) + elif stat.S_ISREG(mode): + try: + shutil.copyfile(entry.path, target) + except OSError as exc: + raise StepInstallError( + f"Failed to stage step package: {exc}" + ) from exc + else: + raise StepInstallError( + f"Step package contains unsupported file: {entry.path}" + ) + + _copy(source_dir, target_dir) + + +def _replace_install( + step_dir: Path, + staged_dir: Path, + registry: Any, + step_id: str, + entry: dict[str, Any], + *, + force: bool, +) -> None: + """Publish the staged package and record its registry entry.""" + from .catalog import StepValidationError + + if step_dir.exists(): + # --force replacement: the replacement is fully staged and validated, + # so it is safe to remove the previous installation now. + try: + shutil.rmtree(step_dir) + except OSError as exc: + raise StepInstallError( + f"Failed to remove the existing step installation at " + f"'{step_dir}': {exc}. Reinstall from the original source with " + "--force." + ) from exc + try: + os.replace(staged_dir, step_dir) + except OSError as exc: + raise StepInstallError( + f"Failed to publish the replacement for step type '{step_id}': " + f"{exc}. The previous installation was removed; reinstall from " + "the original source with --force." + ) from exc + try: + registry.add(step_id, entry) + except (StepValidationError, OSError, TypeError, ValueError) as exc: + raise StepInstallError( + f"Failed to update the step registry for '{step_id}': {exc}. The " + "step directory was replaced but is not registered; reinstall " + "from the original source with --force." + ) from exc + return + + try: + os.replace(staged_dir, step_dir) + except OSError as exc: + raise StepInstallError( + f"Failed to install step '{step_id}': {exc}" + ) from exc + + try: + registry.add(step_id, entry) + except (StepValidationError, OSError, TypeError, ValueError) as exc: + # Fresh install: roll back the just-published directory so the system + # is not left with an unregistered step package on disk. + shutil.rmtree(step_dir, ignore_errors=True) + raise StepInstallError(str(exc)) from exc + + +def install_step_package( + project_root: Path, + step_id: str, + package_dir: Path, + *, + source: str, + catalog_name: str = "", + catalog_metadata: Mapping[str, Any] | None = None, + force: bool = False, +) -> dict[str, Any]: + """Validate, stage, and commit a step package from any source. + + ``source`` is exactly ``"catalog"``, ``"local"``, or ``"url"``. Returns the + registry entry that was persisted. + """ + from .catalog import StepRegistry + + package_dir = Path(package_dir) + if package_dir.is_symlink(): + raise StepInstallError( + f"Refusing to install from a symlinked package directory: '{package_dir}'" + ) + if not package_dir.is_dir(): + raise StepInstallError(f"Step package directory not found: '{package_dir}'") + + validate_step_id(step_id) + steps_base_dir = resolve_steps_base_dir(project_root) + step_dir = _resolve_step_dir(steps_base_dir, step_id) + _reject_unsafe_destination(step_dir) + + # Reject a source that resolves to (or contains) the install destination: + # a --force replacement would otherwise delete the source before it can be + # copied. + try: + source_resolved = package_dir.resolve() + dest_resolved = step_dir.resolve() + except OSError as exc: + raise StepInstallError(f"Failed to resolve step package path: {exc}") from exc + if source_resolved == dest_resolved or dest_resolved.is_relative_to( + source_resolved + ): + raise StepInstallError( + f"Step package source resolves to the install destination: " + f"'{package_dir}'" + ) + + _reject_builtin_collision(step_id) + registry = StepRegistry(project_root) + _check_duplicate(registry, step_id, step_dir, force=force) + + step_meta = validate_step_package(package_dir, step_id) + entry = _build_entry( + step_id, + step_meta, + source=source, + catalog_name=catalog_name, + catalog_metadata=catalog_metadata, + ) + + try: + steps_base_dir.mkdir(parents=True, exist_ok=True) + work_dir = Path( + tempfile.mkdtemp(prefix=_WORK_DIR_PREFIX, dir=steps_base_dir) + ) + except OSError as exc: + raise StepInstallError(f"Failed to create staging directory: {exc}") from exc + + staged_dir = work_dir / "staged" + try: + _copy_package_tree(package_dir, staged_dir) + # Re-validate the complete staged copy: the source may have changed + # while it was copied. + validate_step_package(staged_dir, step_id) + # Recheck the destination immediately before commit (TOCTOU). + _reject_unsafe_destination(step_dir) + _check_duplicate(registry, step_id, step_dir, force=force) + _replace_install( + step_dir, + staged_dir, + registry, + step_id, + entry, + force=force, + ) + finally: + shutil.rmtree(work_dir, ignore_errors=True) + + return entry diff --git a/tests/specify_cli/workflows/step/test_command_add.py b/tests/specify_cli/workflows/step/test_command_add.py index e19c53c6b1..3250057e74 100644 --- a/tests/specify_cli/workflows/step/test_command_add.py +++ b/tests/specify_cli/workflows/step/test_command_add.py @@ -325,11 +325,11 @@ def test_add_rejects_too_many_package_files_before_network( from specify_cli import app from specify_cli.authentication import http as auth_http - from specify_cli.workflows.step import _helpers as step_helpers + from specify_cli.workflows.step import installer from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) - monkeypatch.setattr(step_helpers, "_MAX_STEP_PACKAGE_FILES", 3) + monkeypatch.setattr(installer, "_MAX_STEP_PACKAGE_FILES", 3) monkeypatch.setattr( StepCatalog, "get_step_info", @@ -371,11 +371,11 @@ def test_add_rejects_package_over_cumulative_size_and_cleans_staging( from specify_cli import app from specify_cli.authentication import http as auth_http - from specify_cli.workflows.step import _helpers as step_helpers + from specify_cli.workflows.step import installer from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) - monkeypatch.setattr(step_helpers, "_MAX_STEP_PACKAGE_BYTES", 40) + monkeypatch.setattr(installer, "_MAX_STEP_PACKAGE_BYTES", 40) monkeypatch.setattr( StepCatalog, "get_step_info", @@ -604,3 +604,560 @@ def _fake_open_url(url, timeout=30, redirect_validator=None): assert result.exit_code != 0 assert "empty or non-string URL" in result.output + + +def _write_package(base, type_key="my-step", *, init_body="# init\n"): + package_dir = base / f"{type_key}-pkg" + package_dir.mkdir(parents=True, exist_ok=True) + (package_dir / "step.yml").write_text( + f"step:\n type_key: {type_key}\n name: My Step\n version: 0.1.0\n", + encoding="utf-8", + ) + (package_dir / "__init__.py").write_text(init_body, encoding="utf-8") + return package_dir + + +def _valid_init_body(type_key: str) -> str: + return ( + "from specify_cli.workflows.base import StepBase, StepResult\n\n\n" + "class CustomStep(StepBase):\n" + f" type_key = {type_key!r}\n\n" + " def execute(self, config, context):\n" + " return StepResult(output={'ok': True})\n" + ) + + +def _make_zip(files): + import io + import zipfile + + buffer = io.BytesIO() + with zipfile.ZipFile(buffer, "w") as archive: + for rel, body in files.items(): + data = body.encode("utf-8") if isinstance(body, str) else body + archive.writestr(rel, data) + return buffer.getvalue() + + +def _make_tar_gz(files): + import io + import tarfile + + buffer = io.BytesIO() + with tarfile.open(fileobj=buffer, mode="w:gz") as archive: + for rel, body in files.items(): + data = body.encode("utf-8") if isinstance(body, str) else body + info = tarfile.TarInfo(rel) + info.size = len(data) + archive.addfile(info, io.BytesIO(data)) + return buffer.getvalue() + + +class _ArchiveResponse: + def __init__(self, url, body=b"", content_type=None): + self.url = url + self.body = body + self.content_type = content_type + self.offset = 0 + + def __enter__(self): + return self + + def __exit__(self, exc_type, exc, tb): + return False + + def getheader(self, name): + if name.lower() == "content-type": + return self.content_type + return None + + def geturl(self): + return self.url + + def read(self, size=-1): + if size < 0: + size = len(self.body) - self.offset + chunk = self.body[self.offset : self.offset + size] + self.offset += len(chunk) + return chunk + + +def _valid_archive_files(type_key="my-step"): + return { + "step.yml": f"step:\n type_key: {type_key}\n name: My Step\n", + "__init__.py": "# init\n", + } + + +class TestWorkflowStepAddSources: + def test_dev_installs_and_loads(self, project_dir, tmp_path, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows import STEP_REGISTRY, load_custom_steps + + package = _write_package( + tmp_path, type_key="dev-load-step", init_body=_valid_init_body("dev-load-step") + ) + monkeypatch.chdir(project_dir) + runner = CliRunner() + result = runner.invoke( + app, ["workflow", "step", "add", "dev-load-step", "--dev", str(package)] + ) + + assert result.exit_code == 0, result.output + assert "installed" in result.output + installed = project_dir / ".specify" / "workflows" / "steps" / "dev-load-step" + assert (installed / "step.yml").is_file() + + loaded = load_custom_steps(project_dir) + assert "dev-load-step" in loaded + assert "dev-load-step" in STEP_REGISTRY + + def test_dev_install_list_and_remove(self, project_dir, tmp_path, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + package = _write_package(tmp_path, type_key="dev-step") + monkeypatch.chdir(project_dir) + runner = CliRunner() + assert ( + runner.invoke( + app, ["workflow", "step", "add", "dev-step", "--dev", str(package)] + ).exit_code + == 0 + ) + + listed = runner.invoke(app, ["workflow", "step", "list"]) + assert listed.exit_code == 0 + assert "dev-step" in listed.output + + removed = runner.invoke(app, ["workflow", "step", "remove", "dev-step"]) + assert removed.exit_code == 0 + assert not ( + project_dir / ".specify" / "workflows" / "steps" / "dev-step" + ).exists() + + def test_dev_rejects_missing_init(self, project_dir, tmp_path, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + package = _write_package(tmp_path, type_key="dev-step") + (package / "__init__.py").unlink() + monkeypatch.chdir(project_dir) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "dev-step", "--dev", str(package)] + ) + assert result.exit_code != 0 + assert "__init__.py" in result.output + + def test_dev_rejects_symlinked_source_root(self, project_dir, tmp_path, monkeypatch): + if not hasattr(os, "symlink"): + pytest.skip("symlinks are unavailable") + from typer.testing import CliRunner + from specify_cli import app + + package = _write_package(tmp_path, type_key="dev-step") + link = tmp_path / "linked" + link.symlink_to(package, target_is_directory=True) + monkeypatch.chdir(project_dir) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "dev-step", "--dev", str(link)] + ) + assert result.exit_code != 0 + assert "symlink" in result.output.lower() + + def test_dev_and_from_are_mutually_exclusive(self, project_dir, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + monkeypatch.chdir(project_dir) + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "dev-step", + "--dev", + "somewhere", + "--from", + "https://example.com/pkg.zip", + ], + ) + assert result.exit_code != 0 + assert "mutually exclusive" in result.output + + @pytest.mark.parametrize("option", ["--dev", "--from"]) + def test_empty_source_value_rejected(self, project_dir, monkeypatch, option): + from typer.testing import CliRunner + from specify_cli import app + + monkeypatch.chdir(project_dir) + result = CliRunner().invoke( + app, ["workflow", "step", "add", "dev-step", option, " "] + ) + assert result.exit_code != 0 + + def test_force_replaces_installed_package(self, project_dir, tmp_path, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + package = _write_package(tmp_path, type_key="dev-step", init_body="# old\n") + monkeypatch.chdir(project_dir) + runner = CliRunner() + assert ( + runner.invoke( + app, ["workflow", "step", "add", "dev-step", "--dev", str(package)] + ).exit_code + == 0 + ) + + # A second install without --force is rejected. + duplicate = runner.invoke( + app, ["workflow", "step", "add", "dev-step", "--dev", str(package)] + ) + assert duplicate.exit_code != 0 + assert "already installed" in duplicate.output + + (package / "__init__.py").write_text("# new\n", encoding="utf-8") + forced = runner.invoke( + app, + ["workflow", "step", "add", "dev-step", "--dev", str(package), "--force"], + ) + assert forced.exit_code == 0, forced.output + installed = ( + project_dir / ".specify" / "workflows" / "steps" / "dev-step" / "__init__.py" + ) + assert installed.read_text(encoding="utf-8") == "# new\n" + + def test_force_replaces_orphaned_directory(self, project_dir, tmp_path, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + orphan = ( + project_dir / ".specify" / "workflows" / "steps" / "dev-step" + ) + orphan.mkdir(parents=True) + (orphan / "step.yml").write_text("step:\n type_key: dev-step\n", encoding="utf-8") + (orphan / "__init__.py").write_text("# old\n", encoding="utf-8") + + package = _write_package(tmp_path, type_key="dev-step", init_body="# new\n") + monkeypatch.chdir(project_dir) + + result = CliRunner().invoke( + app, + ["workflow", "step", "add", "dev-step", "--dev", str(package), "--force"], + ) + assert result.exit_code == 0, result.output + assert (orphan / "__init__.py").read_text(encoding="utf-8") == "# new\n" + + def test_from_denied_confirmation_issues_no_request( + self, project_dir, monkeypatch + ): + import typer + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.authentication import http as auth_http + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: False) + monkeypatch.setattr( + auth_http, + "open_url", + lambda *a, **k: (_ for _ in ()).throw( + AssertionError("network request must not be issued") + ), + ) + + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "dev-step", + "--from", + "https://example.com/pkg.zip", + ], + ) + assert result.exit_code == 0 + assert "Cancelled" in result.output + assert not ( + project_dir / ".specify" / "workflows" / "steps" / "dev-step" + ).exists() + + @pytest.mark.parametrize( + ("url", "body_factory", "content_type"), + [ + ("https://example.com/pkg.zip", _make_zip, "application/zip"), + ("https://example.com/pkg.tar.gz", _make_tar_gz, "application/gzip"), + ], + ) + def test_from_archive_installs( + self, project_dir, monkeypatch, url, body_factory, content_type + ): + import typer + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.authentication import http as auth_http + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: True) + body = body_factory(_valid_archive_files()) + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None, extra_headers=None: ( + _ArchiveResponse(url, body, content_type) + ), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step", "--from", url] + ) + assert result.exit_code == 0, result.output + assert ( + project_dir / ".specify" / "workflows" / "steps" / "my-step" / "step.yml" + ).is_file() + + def test_from_rejects_non_https(self, project_dir, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + monkeypatch.chdir(project_dir) + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "my-step", + "--from", + "http://example.com/pkg.zip", + ], + ) + assert result.exit_code != 0 + assert "HTTPS" in result.output + + def test_from_rejects_malformed_url(self, project_dir, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + + monkeypatch.chdir(project_dir) + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "my-step", + "--from", + "https://[not-an-ip]/pkg.zip", + ], + ) + assert result.exit_code != 0 + assert "Invalid URL" in result.output + + def test_from_rejects_redirect_to_non_https(self, project_dir, monkeypatch): + import typer + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.authentication import http as auth_http + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: True) + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None, extra_headers=None: ( + _ArchiveResponse("http://evil.example.com/pkg.zip", b"", "application/zip") + ), + ) + + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "my-step", + "--from", + "https://example.com/pkg.zip", + ], + ) + assert result.exit_code != 0 + assert "non-HTTPS" in result.output + + def test_from_rejects_non_archive_body(self, project_dir, monkeypatch): + import typer + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.authentication import http as auth_http + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: True) + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None, extra_headers=None: ( + _ArchiveResponse(url, b"step:\n type_key: my-step\n", "text/yaml") + ), + ) + + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "my-step", + "--from", + "https://example.com/step.yml", + ], + ) + assert result.exit_code != 0 + assert "supported archive" in result.output + + def test_from_rejects_archive_with_unrelated_siblings( + self, project_dir, monkeypatch + ): + import typer + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.authentication import http as auth_http + + files = { + "inner/step.yml": "step:\n type_key: my-step\n", + "inner/__init__.py": "# init\n", + "README.md": "readme\n", + } + body = _make_zip(files) + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: True) + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None, extra_headers=None: ( + _ArchiveResponse(url, body, "application/zip") + ), + ) + + result = CliRunner().invoke( + app, + [ + "workflow", + "step", + "add", + "my-step", + "--from", + "https://example.com/pkg.zip", + ], + ) + assert result.exit_code != 0 + assert "exactly one top-level" in result.output + + def test_from_denied_when_already_installed_errors_before_prompt( + self, project_dir, tmp_path, monkeypatch + ): + import typer + from typer.testing import CliRunner + from specify_cli import app + + package = _write_package(tmp_path, type_key="my-step") + monkeypatch.chdir(project_dir) + runner = CliRunner() + assert ( + runner.invoke( + app, ["workflow", "step", "add", "my-step", "--dev", str(package)] + ).exit_code + == 0 + ) + + prompts = [] + monkeypatch.setattr( + typer, "confirm", lambda *a, **k: prompts.append(True) or True + ) + result = runner.invoke( + app, + [ + "workflow", + "step", + "add", + "my-step", + "--from", + "https://example.com/pkg.zip", + ], + ) + assert result.exit_code != 0 + assert "already installed" in result.output + assert prompts == [] + + def test_direct_python_call_uses_plain_defaults(self, project_dir, monkeypatch): + """The bundle delegate calls ``workflow_step_add(component.id)``.""" + import typer + + from specify_cli import workflow_step_add + from specify_cli.workflows.step.catalog import StepCatalog + + monkeypatch.chdir(project_dir) + monkeypatch.setattr( + StepCatalog, "get_step_info", lambda self, step_id: None + ) + + # A bare positional call must not raise a TypeError from leaking + # typer.Option metadata; it enters catalog mode and exits cleanly. + with pytest.raises(typer.Exit): + workflow_step_add("my-step") + + +class TestWorkflowStepAddEndToEnd: + _WORKFLOW_YAML = """ +schema_version: "1.0" +workflow: + id: "custom-step-wf" + name: "Custom Step Workflow" + version: "1.0.0" +steps: + - id: run-custom + type: dev-step +""" + + _INIT_BODY = """ +from specify_cli.workflows.base import StepBase, StepResult + + +class DevStep(StepBase): + type_key = "dev-step" + + def execute(self, config, context): + return StepResult(output={"ok": True}) +""" + + def test_dev_install_loads_runs_and_removes( + self, project_dir, tmp_path, monkeypatch + ): + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows import load_custom_steps + + package = _write_package(tmp_path, type_key="dev-step", init_body=self._INIT_BODY) + monkeypatch.chdir(project_dir) + runner = CliRunner() + + installed = runner.invoke( + app, ["workflow", "step", "add", "dev-step", "--dev", str(package)] + ) + assert installed.exit_code == 0, installed.output + + assert "dev-step" in load_custom_steps(project_dir) + + workflow_file = tmp_path / "custom-step-wf.yml" + workflow_file.write_text(self._WORKFLOW_YAML, encoding="utf-8") + run = runner.invoke(app, ["workflow", "run", str(workflow_file), "--json"]) + assert run.exit_code == 0, run.output + assert "completed" in run.output + + removed = runner.invoke(app, ["workflow", "step", "remove", "dev-step"]) + assert removed.exit_code == 0 diff --git a/tests/specify_cli/workflows/step/test_command_info.py b/tests/specify_cli/workflows/step/test_command_info.py index 663b12be47..b63d5f8f78 100644 --- a/tests/specify_cli/workflows/step/test_command_info.py +++ b/tests/specify_cli/workflows/step/test_command_info.py @@ -61,3 +61,47 @@ def test_info_escapes_missing_step_id(self, project_dir, monkeypatch): assert result.exit_code == 1, result.output assert step_id in result.output + + def test_info_prints_local_source(self, project_dir, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.step.catalog import StepRegistry + + monkeypatch.chdir(project_dir) + monkeypatch.setattr( + StepRegistry, + "get", + lambda _registry, step_id: { + "name": "Local Step", + "version": "1.0.0", + "source": "local", + }, + ) + + result = CliRunner().invoke(app, ["workflow", "step", "info", "local-step"]) + + assert result.exit_code == 0, result.output + assert "Source:" in result.output + assert "local" in result.output + + def test_info_prints_catalog_source_with_name(self, project_dir, monkeypatch): + from typer.testing import CliRunner + from specify_cli import app + from specify_cli.workflows.step.catalog import StepRegistry + + monkeypatch.chdir(project_dir) + monkeypatch.setattr( + StepRegistry, + "get", + lambda _registry, step_id: { + "name": "Catalog Step", + "version": "1.0.0", + "source": "catalog", + "catalog_name": "default", + }, + ) + + result = CliRunner().invoke(app, ["workflow", "step", "info", "catalog-step"]) + + assert result.exit_code == 0, result.output + assert "catalog (default)" in result.output diff --git a/tests/specify_cli/workflows/step/test_installer.py b/tests/specify_cli/workflows/step/test_installer.py new file mode 100644 index 0000000000..ecbacc4422 --- /dev/null +++ b/tests/specify_cli/workflows/step/test_installer.py @@ -0,0 +1,626 @@ +"""Domain-focused tests for the workflow step package installer.""" + +from __future__ import annotations + +import json +import os +from pathlib import Path + +import pytest + +from specify_cli.workflows.step import installer + + +def _write_package( + package_dir: Path, type_key: str = "my-step", *, init_body: str = "# init\n" +) -> Path: + package_dir.mkdir(parents=True, exist_ok=True) + (package_dir / "step.yml").write_text( + f"step:\n type_key: {type_key}\n name: My Step\n version: 0.1.0\n", + encoding="utf-8", + ) + (package_dir / "__init__.py").write_text(init_body, encoding="utf-8") + return package_dir + + +def _steps_dir(project_dir: Path) -> Path: + return project_dir / ".specify" / "workflows" / "steps" + + +def _register(project_dir: Path, step_id: str, **overrides) -> None: + from specify_cli.workflows.step.catalog import StepRegistry + + entry = { + "name": "My Step", + "version": "0.1.0", + "type_key": step_id, + "source": "catalog", + "catalog_name": "default", + } + entry.update(overrides) + StepRegistry(project_dir).add(step_id, entry) + + +def _registry_entry(project_dir: Path, step_id: str) -> dict: + path = _steps_dir(project_dir) / "step-registry.json" + return json.loads(path.read_text(encoding="utf-8"))["steps"][step_id] + + +# --------------------------------------------------------------------------- +# Step id validation +# --------------------------------------------------------------------------- + + +@pytest.mark.parametrize("step_id", ["my-step", "my_step", "step2", "a.b", "Step"]) +def test_validate_step_id_accepts_normal(step_id): + installer.validate_step_id(step_id) + + +@pytest.mark.parametrize( + "step_id", + [ + "", + " ", + " padded", + "padded ", + "a/b", + "a\\b", + ".", + "..", + ".hidden", + ".cache", + "step-registry.json", + "con", + "nul", + "com1", + "a:b", + "a*b", + "a= 2: + raise installer.StepInstallError("staged copy invalid") + return real_validate(package_dir, step_id) + + monkeypatch.setattr(installer, "validate_step_package", _validate) + + with pytest.raises(installer.StepInstallError): + installer.install_step_package( + project_dir, "my-step", new_pkg, source="local", force=True + ) + + assert (_steps_dir(project_dir) / "my-step" / "__init__.py").read_text( + encoding="utf-8" + ) == "# old\n" + + +def test_force_registry_failure_warns_reinstall( + tmp_path, project_dir, monkeypatch +): + from specify_cli.workflows.step.catalog import StepRegistry, StepValidationError + + _write_package(_steps_dir(project_dir) / "my-step", init_body="# old\n") + _register(project_dir, "my-step") + new_pkg = _write_package(tmp_path / "pkg", init_body="# new\n") + + def _boom(self, step_id, metadata): + raise StepValidationError("disk full") + + monkeypatch.setattr(StepRegistry, "add", _boom) + + with pytest.raises(installer.StepInstallError) as exc: + installer.install_step_package( + project_dir, "my-step", new_pkg, source="local", force=True + ) + assert "reinstall" in str(exc.value).lower() + assert (_steps_dir(project_dir) / "my-step" / "__init__.py").read_text( + encoding="utf-8" + ) == "# new\n" + + +def test_force_removal_failure_warns_reinstall(tmp_path, project_dir, monkeypatch): + target = _steps_dir(project_dir) / "my-step" + _write_package(target, init_body="# old\n") + _register(project_dir, "my-step") + new_pkg = _write_package(tmp_path / "pkg", init_body="# new\n") + + real_rmtree = installer.shutil.rmtree + + def _rmtree(path, *args, **kwargs): + if Path(path) == target: + raise OSError("cannot remove") + return real_rmtree(path, *args, **kwargs) + + monkeypatch.setattr(installer.shutil, "rmtree", _rmtree) + + with pytest.raises(installer.StepInstallError) as exc: + installer.install_step_package( + project_dir, "my-step", new_pkg, source="local", force=True + ) + assert "reinstall" in str(exc.value).lower() + assert (target / "__init__.py").read_text(encoding="utf-8") == "# old\n" + + +def test_force_publication_failure_warns_reinstall(tmp_path, project_dir, monkeypatch): + target = _steps_dir(project_dir) / "my-step" + _write_package(target, init_body="# old\n") + _register(project_dir, "my-step") + new_pkg = _write_package(tmp_path / "pkg", init_body="# new\n") + + real_replace = installer.os.replace + + def _replace(src, dst, *args, **kwargs): + if Path(dst) == target: + raise OSError("rename failed") + return real_replace(src, dst, *args, **kwargs) + + monkeypatch.setattr(installer.os, "replace", _replace) + + with pytest.raises(installer.StepInstallError) as exc: + installer.install_step_package( + project_dir, "my-step", new_pkg, source="local", force=True + ) + assert "reinstall" in str(exc.value).lower() + + +def test_force_replaces_orphaned_directory(tmp_path, project_dir): + orphan = _write_package(_steps_dir(project_dir) / "my-step", init_body="# old\n") + assert orphan.is_dir() + + new_pkg = _write_package(tmp_path / "pkg", init_body="# new\n") + installer.install_step_package( + project_dir, "my-step", new_pkg, source="local", force=True + ) + + assert (_steps_dir(project_dir) / "my-step" / "__init__.py").read_text( + encoding="utf-8" + ) == "# new\n" + + +def test_no_backup_artifacts_after_force(tmp_path, project_dir): + _write_package(_steps_dir(project_dir) / "my-step", init_body="# old\n") + _register(project_dir, "my-step") + new_pkg = _write_package(tmp_path / "pkg", init_body="# new\n") + + installer.install_step_package( + project_dir, "my-step", new_pkg, source="local", force=True + ) + + names = sorted(path.name for path in _steps_dir(project_dir).iterdir()) + assert names == ["my-step", "step-registry.json"] + + +def test_loader_does_not_discover_staging_package(project_dir): + from specify_cli.workflows import load_custom_steps + + staging = _steps_dir(project_dir) / ".speckit-step-install-abc" / "staged" + _write_package(staging, type_key="staged-only-step") + + loaded = load_custom_steps(project_dir) + assert "staged-only-step" not in loaded + + +def test_builtin_collision_uses_immutable_snapshot(tmp_path, project_dir, monkeypatch): + from specify_cli.workflows import BUILTIN_STEP_TYPES, STEP_REGISTRY + + monkeypatch.delitem(STEP_REGISTRY, "shell", raising=False) + assert "shell" in BUILTIN_STEP_TYPES + + pkg = _write_package(tmp_path / "pkg", type_key="shell") + with pytest.raises(installer.StepInstallError, match="built-in"): + installer.install_step_package(project_dir, "shell", pkg, source="local") + + +def test_check_installable_exposes_duplicate_before_install(tmp_path, project_dir): + pkg = _write_package(tmp_path / "pkg") + installer.check_installable(project_dir, "my-step") + installer.install_step_package(project_dir, "my-step", pkg, source="local") + + with pytest.raises(installer.StepInstallError, match="already installed"): + installer.check_installable(project_dir, "my-step") + # force permits the preflight. + installer.check_installable(project_dir, "my-step", force=True) + + +def _tree(root: Path) -> dict[str, bytes]: + return { + str(path.relative_to(root)): path.read_bytes() + for path in sorted(root.rglob("*")) + if path.is_file() + } + + +def test_all_sources_share_tree_and_metadata(tmp_path): + pkg = _write_package(tmp_path / "pkg", type_key="parity-step") + expected_tree = _tree(pkg) + + entries: dict[str, dict] = {} + trees: dict[str, dict] = {} + for source in ("catalog", "local", "url"): + project = tmp_path / f"proj-{source}" + project.mkdir() + entries[source] = installer.install_step_package( + project, + "parity-step", + pkg, + source=source, + catalog_name="default" if source == "catalog" else "", + catalog_metadata={"name": "Parity"} if source == "catalog" else None, + ) + trees[source] = _tree(_steps_dir(project) / "parity-step") + + assert trees["catalog"] == trees["local"] == trees["url"] == expected_tree + + for source, entry in entries.items(): + assert entry["source"] == source + assert entry["type_key"] == "parity-step" + if source == "catalog": + assert entry["catalog_name"] == "default" + else: + assert "catalog_name" not in entry + + +def test_all_sources_share_identity_rejection(tmp_path): + pkg = _write_package(tmp_path / "pkg", type_key="wrong-step") + for index, source in enumerate(("catalog", "local", "url")): + project = tmp_path / f"proj-{index}" + project.mkdir() + with pytest.raises(installer.StepInstallError, match="does not match"): + installer.install_step_package( + project, "parity-step", pkg, source=source + ) From 461494c9fec68ab5e7ce24e5bf17cfc7ea220fb8 Mon Sep 17 00:00:00 2001 From: Markus Date: Fri, 25 Sep 2026 19:06:08 +0200 Subject: [PATCH 2/2] fix(workflows): harden local step installation Assisted-by: OpenCode (model: gpt-5.6-terra, autonomous) --- docs/reference/workflows.md | 18 +- src/specify_cli/workflows/__init__.py | 22 +- .../workflows/step/catalog/_domain.py | 87 +++++- src/specify_cli/workflows/step/command_add.py | 90 +++++- .../workflows/step/command_remove.py | 89 ++---- src/specify_cli/workflows/step/installer.py | 277 ++++++++++++++++-- tests/specify_cli/bundles/test_primitives.py | 26 +- .../step/catalog/test_command_list.py | 4 +- .../workflows/step/catalog/test_registry.py | 33 +++ .../workflows/step/test_command_add.py | 97 +++++- .../workflows/step/test_command_info.py | 11 +- .../workflows/step/test_command_list.py | 8 +- .../workflows/step/test_command_remove.py | 5 +- .../workflows/step/test_command_search.py | 8 +- .../workflows/step/test_installer.py | 48 +++ .../workflows/test_custom_steps.py | 49 ++++ 16 files changed, 702 insertions(+), 170 deletions(-) create mode 100644 tests/specify_cli/workflows/step/catalog/test_registry.py create mode 100644 tests/specify_cli/workflows/test_custom_steps.py diff --git a/docs/reference/workflows.md b/docs/reference/workflows.md index 9de667b401..17a762015f 100644 --- a/docs/reference/workflows.md +++ b/docs/reference/workflows.md @@ -652,12 +652,18 @@ Every source is validated identically before anything is committed: directory are rejected — including inside excluded directories. - `.git`, `__pycache__`, and `.DS_Store` entries are excluded from the copy and from the limits. -- A package may contain at most **512 files** and **50 MiB** in total. -- `__init__.py` is **not imported** during installation; it is loaded only when - the step runs. - -> **Security note:** Installing a custom step runs its Python with **your** -> privileges. Only install step packages from sources you trust. +- The installed-package policy permits at most **512 retained files** and + **50 MiB** of retained content. Excluded entries do not consume this budget. +- Archive URLs also pass transport/extraction safety limits before package + validation: at most 512 archive entries, 50 MiB downloaded or extracted, and + 10 MiB per archive member. Catalog files have a 50 MiB per-response bound. +- Installation validates and copies the package but does **not** import or + execute `__init__.py`. Installed custom step modules are loaded during startup + of `workflow add`, `workflow run`, and `workflow resume`, before any particular + custom step necessarily executes. + +> **Security note:** Loading a custom step runs its Python with **your** +> privileges. Only install and retain step packages from sources you trust. #### Listing, running, and removing diff --git a/src/specify_cli/workflows/__init__.py b/src/specify_cli/workflows/__init__.py index 2bb3de56a5..baa3b8fa09 100644 --- a/src/specify_cli/workflows/__init__.py +++ b/src/specify_cli/workflows/__init__.py @@ -75,11 +75,27 @@ def _register_builtin_steps() -> None: # The step types Spec Kit ships, snapshotted before any community step can be # loaded. ``load_custom_steps`` adds project-installed ids to the process-global -# ``STEP_REGISTRY`` and never removes them, so ``STEP_REGISTRY`` cannot answer +# ``STEP_REGISTRY`` and refreshes them for each project, so it cannot answer # "is this bundled with Spec Kit?" in a long-lived process: a step loaded for one # project would look built-in for the next. Callers that need the immutable set # (e.g. the bundler's reference checker) must use this instead. BUILTIN_STEP_TYPES: frozenset[str] = frozenset(STEP_REGISTRY) +_CUSTOM_STEP_MODULES: set[str] = set() + + +def _unload_custom_steps() -> None: + """Clear custom registrations and synthetic imports from a prior project.""" + import sys + + for type_key in tuple(STEP_REGISTRY): + if type_key not in BUILTIN_STEP_TYPES: + del STEP_REGISTRY[type_key] + for module_name in _CUSTOM_STEP_MODULES: + sys.modules.pop(module_name, None) + prefix = module_name + "." + for loaded_name in [name for name in sys.modules if name.startswith(prefix)]: + sys.modules.pop(loaded_name, None) + _CUSTOM_STEP_MODULES.clear() def load_custom_steps(project_root: Path) -> list[str]: @@ -97,6 +113,7 @@ def load_custom_steps(project_root: Path) -> list[str]: import re as _re import sys as _sys + _unload_custom_steps() steps_dir = Path(project_root) / ".specify" / "workflows" / "steps" # Defense-in-depth: refuse to execute step code from a symlinked @@ -192,6 +209,7 @@ def load_custom_steps(project_root: Path) -> list[str]: _register_step(step_class()) loaded.append(type_key) registered = True + _CUSTOM_STEP_MODULES.add(module_name) finally: # If the step wasn't successfully registered (failed import, # no matching StepBase subclass, or registration error), remove @@ -206,7 +224,7 @@ def load_custom_steps(project_root: Path) -> list[str]: k for k in _sys.modules if k.startswith(submodule_prefix) ]: _sys.modules.pop(_mod_key, None) - except Exception: # noqa: BLE001 + except Exception: # noqa: BLE001, S112 # Silently skip broken step packages at load time continue diff --git a/src/specify_cli/workflows/step/catalog/_domain.py b/src/specify_cli/workflows/step/catalog/_domain.py index 08a1b22f57..bc166562bc 100644 --- a/src/specify_cli/workflows/step/catalog/_domain.py +++ b/src/specify_cli/workflows/step/catalog/_domain.py @@ -5,15 +5,17 @@ import hashlib import json import os +import stat +import tempfile import time from dataclasses import dataclass +from datetime import UTC from pathlib import Path from typing import Any import yaml from ...._download_security import ( - MAX_JSON_CATALOG_BYTES as MAX_JSON_CATALOG_BYTES, read_response_limited, ) @@ -111,48 +113,104 @@ def _load(self) -> dict[str, Any]: return default_registry def save(self) -> None: - """Persist registry to disk. - - Raises ``StepValidationError`` with a clear message on filesystem - errors (read-only fs, permission denied, ...) so callers can surface - a clean error to the user rather than an unhandled ``OSError``. - """ + """Persist registry atomically without truncating an existing file.""" if self._has_symlinked_parent() or self.registry_path.is_symlink(): raise StepValidationError( "Refusing to write step registry through a symlinked path." ) + fd = -1 + tmp: str | None = None try: self.steps_dir.mkdir(parents=True, exist_ok=True) - with open(self.registry_path, "w", encoding="utf-8") as f: + fd, tmp = tempfile.mkstemp( + dir=str(self.registry_path.parent), + prefix=f".{self.registry_path.name}.", + suffix=".tmp", + ) + # Keep the exclusive descriptor open while writing and checking the + # path so a replaced temporary file can never be committed. + with os.fdopen(os.dup(fd), "w", encoding="utf-8") as f: json.dump(self.data, f, indent=2) - except OSError as exc: + f.flush() + os.fsync(f.fileno()) + try: + if self.registry_path.exists(): + existing = self.registry_path.stat(follow_symlinks=False) + if stat.S_ISREG(existing.st_mode) and hasattr(os, "fchmod"): + os.fchmod(fd, stat.S_IMODE(existing.st_mode)) + if stat.S_ISREG(existing.st_mode) and hasattr(os, "fchown"): + try: + os.fchown(fd, existing.st_uid, existing.st_gid) + except PermissionError: + pass + except OSError: + # Persisting valid data is more important than preserving mode + # or ownership metadata when that best-effort operation fails. + pass + staged = os.stat(tmp, follow_symlinks=False) + opened = os.fstat(fd) + if ( + not stat.S_ISREG(staged.st_mode) + or staged.st_dev != opened.st_dev + or staged.st_ino != opened.st_ino + ): + raise OSError("Staged step registry changed before commit") + os.close(fd) + fd = -1 + os.replace(tmp, self.registry_path) + tmp = None + except (OSError, TypeError, ValueError) as exc: raise StepValidationError( f"Failed to write step registry at {self.registry_path}: {exc}" ) from exc + finally: + if fd >= 0: + try: + os.close(fd) + except OSError: + pass + if tmp is not None: + try: + os.unlink(tmp) + except OSError: + pass def add(self, step_id: str, metadata: dict[str, Any]) -> None: """Add or update an installed step entry.""" import copy - from datetime import datetime, timezone + from datetime import datetime raw_existing = self.data["steps"].get(step_id) + had_entry = step_id in self.data["steps"] # Corrupted-but-parseable registries may hold non-dict entries; treat # them as absent rather than crashing on existing.get() (mirrors # WorkflowRegistry.add). existing = raw_existing if isinstance(raw_existing, dict) else {} metadata_to_store = copy.deepcopy(metadata) metadata_to_store["installed_at"] = existing.get( - "installed_at", datetime.now(timezone.utc).isoformat() + "installed_at", datetime.now(UTC).isoformat() ) - metadata_to_store["updated_at"] = datetime.now(timezone.utc).isoformat() + metadata_to_store["updated_at"] = datetime.now(UTC).isoformat() self.data["steps"][step_id] = metadata_to_store - self.save() + try: + self.save() + except (StepValidationError, TypeError, ValueError): + if had_entry: + self.data["steps"][step_id] = raw_existing + else: + del self.data["steps"][step_id] + raise def remove(self, step_id: str) -> bool: """Remove an installed step entry. Returns True if found.""" if step_id in self.data["steps"]: + removed_entry = self.data["steps"][step_id] del self.data["steps"][step_id] - self.save() + try: + self.save() + except (StepValidationError, TypeError, ValueError): + self.data["steps"][step_id] = removed_entry + raise return True return False @@ -420,6 +478,7 @@ def _fetch_single_catalog( pass from urllib.parse import urlparse + from specify_cli.authentication.http import open_url as _open_url def _validate_url(url: str) -> None: diff --git a/src/specify_cli/workflows/step/command_add.py b/src/specify_cli/workflows/step/command_add.py index 3aba60bce3..5738533e1a 100644 --- a/src/specify_cli/workflows/step/command_add.py +++ b/src/specify_cli/workflows/step/command_add.py @@ -14,6 +14,8 @@ from . import _helpers as step_helpers from . import step_app +_MAX_STEP_CATALOG_RESPONSE_BYTES = 50 * 1024 * 1024 + def _cleanup_download_tmp_path(tmp_path: cli.Path | None) -> None: """Best-effort unlink of a partially-downloaded step archive temp file. @@ -35,7 +37,11 @@ def _cleanup_download_tmp_path(tmp_path: cli.Path | None) -> None: def _print_installed(step_id: str, entry: dict) -> None: step_name = entry.get("name") or step_id - cli.console.print(f"[green]✓[/green] Step type '{step_name}' ({step_id}) installed") + cli.console.print( + "[green]✓[/green] Step type " + f"'{cli._escape_markup(str(step_name))}' " + f"({cli._escape_markup(str(step_id))}) installed" + ) cli.console.print( " Use [cyan]specify workflow step list[/cyan] to verify the installation." ) @@ -126,6 +132,8 @@ def _install_from_url( download_url = from_url extra_headers = None tmp_path: cli.Path | None = None + extract_tmp: tempfile.TemporaryDirectory[str] | None = None + committed = False try: resolved_url = _resolve_gh_asset( from_url, @@ -154,16 +162,30 @@ def _install_from_url( if hasattr(resp, "getheader") else None ) - archive_format = ( - cli.archive_format_from_name(final_url) - or cli.archive_format_from_name(from_url) - or cli.archive_format_from_content_type(content_type) - ) + declarations = [ + ("requested URL", from_url, cli.archive_format_from_name(from_url)), + ("final URL", final_url, cli.archive_format_from_name(final_url)), + ( + "Content-Type", + content_type or "", + cli.archive_format_from_content_type(content_type), + ), + ] + recognized = [item for item in declarations if item[2] is not None] + archive_format = recognized[0][2] if recognized else None if archive_format is None: raise installer.StepInstallError( "URL does not reference a supported archive " "(.zip, .tar.gz, or .tgz)" ) + if any(item[2] != archive_format for item in recognized): + details = ", ".join( + f"{label} declares {declared}" + for label, _value, declared in recognized + ) + raise installer.StepInstallError( + f"Archive format mismatch: {cli._escape_markup(details)}" + ) downloaded = cli.read_response_limited( resp, error_type=ValueError, @@ -176,15 +198,16 @@ def _install_from_url( tmp_path = cli.Path(tmp.name) tmp.write(downloaded) - with tempfile.TemporaryDirectory( - prefix="speckit-step-archive-" - ) as extract_dir: - extracted_root = cli.Path(extract_dir) + extract_tmp = tempfile.TemporaryDirectory(prefix="speckit-step-archive-") + extracted_root = cli.Path(extract_tmp.name) + try: # safe_extract_archive re-detects and confirms the archive bytes. cli.safe_extract_archive( tmp_path, extracted_root, - source_name=final_url, + source_name=next( + value for _label, value, declared in recognized if declared is not None + ), content_type=content_type, ) package_root = installer.resolve_package_root(extracted_root) @@ -195,6 +218,22 @@ def _install_from_url( source="url", force=force, ) + committed = True + finally: + try: + extract_tmp.cleanup() + except OSError as cleanup_exc: + if committed: + cli.console.print( + "[yellow]Warning:[/yellow] Could not remove temporary step " + f"archive directory: {cli._escape_markup(str(cleanup_exc))} " + f"(path: {cli._escape_markup(extract_tmp.name)})" + ) + elif __import__("sys").exc_info()[0] is None: + raise installer.StepInstallError( + "Failed to remove temporary step archive directory: " + f"{cleanup_exc}" + ) from cleanup_exc except cli.typer.Exit: raise except installer.StepInstallError: @@ -322,10 +361,17 @@ def _safe_fetch(url: str) -> bytes: final_url = resp.geturl() if not cli.is_https_or_localhost_http(final_url): raise ValueError(f"Redirect to non-HTTPS URL: {final_url}") - return cli._read_response_within_limit(resp) + return cli.read_response_limited( + resp, + max_bytes=_MAX_STEP_CATALOG_RESPONSE_BYTES, + error_type=ValueError, + label="step package response", + ) - with tempfile.TemporaryDirectory(prefix="speckit-step-package-") as package_tmp: - package_dir = cli.Path(package_tmp) + package_tmp = tempfile.TemporaryDirectory(prefix="speckit-step-package-") + package_dir = cli.Path(package_tmp.name) + committed = False + try: try: step_yml_content = _safe_fetch(step_yml_url) init_py_content = _safe_fetch(init_url) @@ -408,6 +454,22 @@ def _safe_fetch(url: str) -> bytes: catalog_metadata=info, force=force, ) + committed = True + finally: + try: + package_tmp.cleanup() + except OSError as cleanup_exc: + if committed: + cli.console.print( + "[yellow]Warning:[/yellow] Could not remove temporary step " + f"package directory: {cli._escape_markup(str(cleanup_exc))} " + f"(path: {cli._escape_markup(package_tmp.name)})" + ) + elif __import__("sys").exc_info()[0] is None: + raise installer.StepInstallError( + "Failed to remove temporary step package directory: " + f"{cleanup_exc}" + ) from cleanup_exc _print_installed(step_id, entry) diff --git a/src/specify_cli/workflows/step/command_remove.py b/src/specify_cli/workflows/step/command_remove.py index 58ec7602f3..76421b2321 100644 --- a/src/specify_cli/workflows/step/command_remove.py +++ b/src/specify_cli/workflows/step/command_remove.py @@ -3,9 +3,8 @@ from __future__ import annotations from .. import _commands as cli -from . import step_app - from . import _helpers as step_helpers +from . import step_app @step_app.command("remove") @@ -13,85 +12,31 @@ def workflow_step_remove( step_id: str = cli.typer.Argument(..., help="Step type ID to uninstall"), ): """Uninstall a custom step type.""" - from .catalog import StepRegistry, StepValidationError + import shutil + + from . import installer project_root = cli._require_specify_project() step_helpers._validate_step_id_or_exit(step_id) - registry = StepRegistry(project_root) - in_registry = registry.is_installed(step_id) - - steps_base_dir = step_helpers._resolve_steps_base_dir_or_exit(project_root) - step_dir = (steps_base_dir / step_id).resolve() - # Defense-in-depth: even though step_helpers._validate_step_id_or_exit rejects path - # separators, ensure that the resolved directory is a single child of - # steps_base_dir and is not steps_base_dir itself. try: - rel_parts = step_dir.relative_to(steps_base_dir).parts - except ValueError: - cli.console.print(f"[red]Error:[/red] Invalid step id '{step_id}'") - raise cli.typer.Exit(1) - if rel_parts != (step_id,): - cli.console.print(f"[red]Error:[/red] Invalid step id '{step_id}'") - raise cli.typer.Exit(1) - - dir_exists = step_dir.exists() - - if not in_registry and not dir_exists: - cli.console.print(f"[red]Error:[/red] Step type '{step_id}' is not installed") - raise cli.typer.Exit(1) - - if not in_registry and dir_exists: - # The registry was likely reset due to corruption. Warn the user that the - # directory is being removed even though there is no registry entry, so - # the orphaned package can be cleaned up and a fresh install attempted. + staged_dir, removed_orphan = installer.remove_step_package(project_root, step_id) + except installer.StepInstallError as exc: + cli.console.print(f"[red]Error:[/red] {exc}") + raise cli.typer.Exit(1) from exc + if removed_orphan: cli.console.print( - f"[yellow]Warning:[/yellow] '{step_id}' has no registry entry " - "(registry may have been reset). Removing the orphaned directory." + f"[yellow]Warning:[/yellow] '{cli._escape_markup(step_id)}' had no registry " + "entry. Removing the orphaned directory." ) - - if dir_exists and not in_registry: - # No registry write needed; just delete the orphaned directory. - import shutil - + cli.console.print(f"[green]✓[/green] Step type '{cli._escape_markup(step_id)}' uninstalled") + if staged_dir is not None: try: - shutil.rmtree(step_dir) + shutil.rmtree(staged_dir) except OSError as exc: cli.console.print( - f"[red]Error:[/red] Failed to remove step directory {step_dir}: {exc}" + "[yellow]Warning:[/yellow] Step was uninstalled, but its staged " + f"directory could not be deleted: {cli._escape_markup(str(exc))}. " + f"Remove it manually: {cli._escape_markup(str(staged_dir))}" ) - raise cli.typer.Exit(1) - elif in_registry: - # Remove the registry entry, then the directory. If the directory - # delete fails, restore the registry entry so state stays consistent - # and a future `step add` isn't blocked by an orphaned directory - # with no registry entry. - registry_metadata = registry.get(step_id) - try: - registry.remove(step_id) - except StepValidationError as exc: - cli.console.print(f"[red]Error:[/red] {exc}") - raise cli.typer.Exit(1) - if dir_exists: - import shutil - - try: - shutil.rmtree(step_dir) - except OSError as exc: - # Restore the original registry entry verbatim (bypass add() - # which would overwrite timestamps). - try: - if registry_metadata is not None: - registry.data["steps"][step_id] = registry_metadata - registry.save() - except Exception as restore_exc: # noqa: BLE001 - cli.console.print( - f"[yellow]Warning:[/yellow] Failed to restore registry entry " - f"for '{step_id}' after directory removal failure: {restore_exc}" - ) - cli.console.print( - f"[red]Error:[/red] Failed to remove step directory {step_dir}: {exc}" - ) - raise cli.typer.Exit(1) - cli.console.print(f"[green]✓[/green] Step type '{step_id}' uninstalled") diff --git a/src/specify_cli/workflows/step/installer.py b/src/specify_cli/workflows/step/installer.py index 1e37e2b57c..5c1a2efbc1 100644 --- a/src/specify_cli/workflows/step/installer.py +++ b/src/specify_cli/workflows/step/installer.py @@ -11,6 +11,7 @@ from __future__ import annotations +import contextlib import os import shutil import stat @@ -66,12 +67,73 @@ ) _WINDOWS_INVALID_CHARS: frozenset[str] = frozenset('<>:"|?*') +_SOURCES: frozenset[str] = frozenset({"catalog", "local", "url"}) class StepInstallError(Exception): """User-facing step package install/validation failure.""" +@contextlib.contextmanager +def _step_install_transaction(project_root: Path): + """Serialize step directory swaps with their registry updates.""" + from ...shared_infra import _ensure_safe_shared_directory + + lock_dir = Path(project_root) / ".specify" + try: + _ensure_safe_shared_directory( + Path(project_root), lock_dir, context="step install lock directory" + ) + except ValueError as exc: + raise StepInstallError(str(exc)) from exc + lock_file = lock_dir / ".step-install.lock" + if lock_file.is_symlink(): + raise StepInstallError(f"Refusing to use symlinked step install lock: {lock_file}") + + flags = os.O_RDWR | os.O_CREAT + flags |= getattr(os, "O_NOFOLLOW", 0) + flags |= getattr(os, "O_CLOEXEC", 0) + try: + fd = os.open(lock_file, flags, 0o600) + except OSError as exc: + raise StepInstallError(f"Failed to open step install lock: {exc}") from exc + try: + if lock_file.is_symlink(): + raise StepInstallError( + f"Refusing to use symlinked step install lock: {lock_file}" + ) + if os.name == "nt": + import errno + import msvcrt + import time + + if os.fstat(fd).st_size == 0: + os.write(fd, b"\0") + while True: + os.lseek(fd, 0, os.SEEK_SET) + try: + msvcrt.locking(fd, msvcrt.LK_NBLCK, 1) + break + except OSError as exc: + if exc.errno not in (errno.EACCES, errno.EDEADLK): + raise + time.sleep(0.05) + else: + import fcntl + + fcntl.flock(fd, fcntl.LOCK_EX) + yield + except StepInstallError: + raise + except OSError as exc: + raise StepInstallError(f"Failed to lock step installation: {exc}") from exc + finally: + try: + os.close(fd) + except OSError: + pass + + # --------------------------------------------------------------------------- # Step id + base directory validation # --------------------------------------------------------------------------- @@ -396,19 +458,51 @@ def _build_entry( catalog_name: str, catalog_metadata: Mapping[str, Any] | None, ) -> dict[str, Any]: + if source not in _SOURCES: + raise StepInstallError( + "Step install source must be one of: catalog, local, url" + ) + if source == "catalog" and not isinstance(catalog_name, str): + raise StepInstallError("Catalog step install requires a string catalog name") + if catalog_metadata is not None and not isinstance(catalog_metadata, Mapping): + raise StepInstallError("Catalog step metadata must be a mapping") catalog_metadata = catalog_metadata or {} + + def _string_value(metadata: Mapping[str, Any], field: str) -> str | None: + value = metadata.get(field) + if value is None: + return None + if not isinstance(value, str): + raise StepInstallError( + f"step metadata '{field}' must be a string when present" + ) + return value + + type_key = _string_value(step_meta, "type_key") + if not type_key: + raise StepInstallError("step.yml missing 'step.type_key' field") + package_values = { + field: _string_value(step_meta, field) + for field in ("name", "version", "description", "author") + } + catalog_values = { + field: _string_value(catalog_metadata, field) + for field in ("name", "version", "description", "author") + } entry: dict[str, Any] = { - "name": catalog_metadata.get("name") - or step_meta.get("name") - or step_id, - "version": catalog_metadata.get("version") - or step_meta.get("version") - or "0.0.0", - "description": catalog_metadata.get( - "description", step_meta.get("description", "") + "name": catalog_values["name"] or package_values["name"] or step_id, + "version": catalog_values["version"] or package_values["version"] or "0.0.0", + "description": ( + catalog_values["description"] + if catalog_values["description"] is not None + else package_values["description"] or "" + ), + "author": ( + catalog_values["author"] + if catalog_values["author"] is not None + else package_values["author"] or "" ), - "author": catalog_metadata.get("author", step_meta.get("author", "")), - "type_key": step_meta["type_key"], + "type_key": type_key, "source": source, } if source == "catalog": @@ -451,12 +545,7 @@ def _copy(current: Path, destination: Path) -> None: if stat.S_ISDIR(mode): _copy(Path(entry.path), target) elif stat.S_ISREG(mode): - try: - shutil.copyfile(entry.path, target) - except OSError as exc: - raise StepInstallError( - f"Failed to stage step package: {exc}" - ) from exc + _copy_regular_file(entry.path, target, mode) else: raise StepInstallError( f"Step package contains unsupported file: {entry.path}" @@ -465,6 +554,37 @@ def _copy(current: Path, destination: Path) -> None: _copy(source_dir, target_dir) +def _copy_regular_file(source: str, target: Path, expected_mode: int) -> None: + """Copy an inspected regular file without following a late symlink swap.""" + flags = os.O_RDONLY | getattr(os, "O_NOFOLLOW", 0) + try: + fd = os.open(source, flags) + except OSError as exc: + raise StepInstallError(f"Failed to stage step package: {exc}") from exc + try: + opened = os.fstat(fd) + source_state = os.stat(source, follow_symlinks=False) + if ( + not stat.S_ISREG(opened.st_mode) + or not stat.S_ISREG(source_state.st_mode) + or opened.st_dev != source_state.st_dev + or opened.st_ino != source_state.st_ino + or stat.S_IFMT(source_state.st_mode) != stat.S_IFMT(expected_mode) + ): + raise StepInstallError( + f"Step package file changed while staging: {source}" + ) + with os.fdopen(fd, "rb", closefd=False) as source_file, target.open("xb") as target_file: + shutil.copyfileobj(source_file, target_file) + except OSError as exc: + raise StepInstallError(f"Failed to stage step package: {exc}") from exc + finally: + try: + os.close(fd) + except OSError: + pass + + def _replace_install( step_dir: Path, staged_dir: Path, @@ -478,6 +598,11 @@ def _replace_install( from .catalog import StepValidationError if step_dir.exists(): + if not force: + raise StepInstallError( + f"Step directory already exists at '{step_dir}'. Remove it manually " + f"or use: [cyan]specify workflow step remove {step_id}[/cyan]" + ) # --force replacement: the replacement is fully staged and validated, # so it is safe to remove the previous installation now. try: @@ -518,7 +643,14 @@ def _replace_install( except (StepValidationError, OSError, TypeError, ValueError) as exc: # Fresh install: roll back the just-published directory so the system # is not left with an unregistered step package on disk. - shutil.rmtree(step_dir, ignore_errors=True) + try: + shutil.rmtree(step_dir) + except OSError as cleanup_exc: + raise StepInstallError( + f"Failed to update the step registry for '{step_id}': {exc}. " + f"The unregistered package remains at '{step_dir}' because rollback " + f"failed: {cleanup_exc}. Remove it manually before reinstalling." + ) from exc raise StepInstallError(str(exc)) from exc @@ -572,10 +704,11 @@ def install_step_package( registry = StepRegistry(project_root) _check_duplicate(registry, step_id, step_dir, force=force) - step_meta = validate_step_package(package_dir, step_id) - entry = _build_entry( + # Validate source and all caller-controlled metadata before creating any + # project directories. The staged metadata is used for the final entry. + _build_entry( step_id, - step_meta, + validate_step_package(package_dir, step_id), source=source, catalog_name=catalog_name, catalog_metadata=catalog_metadata, @@ -590,23 +723,105 @@ def install_step_package( raise StepInstallError(f"Failed to create staging directory: {exc}") from exc staged_dir = work_dir / "staged" + committed = False try: _copy_package_tree(package_dir, staged_dir) # Re-validate the complete staged copy: the source may have changed # while it was copied. - validate_step_package(staged_dir, step_id) - # Recheck the destination immediately before commit (TOCTOU). - _reject_unsafe_destination(step_dir) - _check_duplicate(registry, step_id, step_dir, force=force) - _replace_install( - step_dir, - staged_dir, - registry, + staged_meta = validate_step_package(staged_dir, step_id) + entry = _build_entry( step_id, - entry, - force=force, + staged_meta, + source=source, + catalog_name=catalog_name, + catalog_metadata=catalog_metadata, ) + # Serialize destination and registry changes. Source downloads/copying + # stay outside the lock, but all state that can conflict is reloaded and + # checked again immediately before publication. + with _step_install_transaction(project_root): + locked_base_dir = resolve_steps_base_dir(project_root) + if locked_base_dir != steps_base_dir: + raise StepInstallError( + "Step directory changed while staging; reinstall from the original source" + ) + step_dir = _resolve_step_dir(locked_base_dir, step_id) + _reject_unsafe_destination(step_dir) + _reject_builtin_collision(step_id) + registry = StepRegistry(project_root) + _check_duplicate(registry, step_id, step_dir, force=force) + _replace_install( + step_dir, + staged_dir, + registry, + step_id, + entry, + force=force, + ) + committed = True finally: - shutil.rmtree(work_dir, ignore_errors=True) + try: + shutil.rmtree(work_dir) + except OSError as cleanup_exc: + # The staged directory is private and cannot be loaded as a step, + # but callers still need an actionable residual-path diagnostic. + if work_dir.exists() and not committed and os.sys.exc_info()[0] is None: + raise StepInstallError( + f"Failed to remove staging directory '{work_dir}': {cleanup_exc}" + ) from cleanup_exc return entry + + +def remove_step_package(project_root: Path, step_id: str) -> tuple[Path | None, bool]: + """Remove one custom step under the same transaction as installation. + + The returned directory is staged after the registry removal has committed; + the second value identifies a removed orphan. Callers can delete the staged + directory best-effort without changing the successful result. + """ + from .catalog import StepRegistry, StepValidationError + + validate_step_id(step_id) + with _step_install_transaction(project_root): + steps_base_dir = resolve_steps_base_dir(project_root) + step_dir = _resolve_step_dir(steps_base_dir, step_id) + _reject_unsafe_destination(step_dir) + registry = StepRegistry(project_root) + in_registry = registry.is_installed(step_id) + if not in_registry and not step_dir.exists(): + raise StepInstallError(f"Step type '{step_id}' is not installed") + + staged_dir: Path | None = None + if step_dir.exists(): + try: + staged_dir = Path( + tempfile.mkdtemp( + prefix=f".{step_id}.removing-", dir=steps_base_dir + ) + ) + staged_dir.rmdir() + os.replace(step_dir, staged_dir) + except OSError as exc: + raise StepInstallError( + f"Failed to stage step directory '{step_dir}' for removal: {exc}" + ) from exc + + if not in_registry: + return staged_dir, True + try: + registry.remove(step_id) + except (StepValidationError, OSError, TypeError, ValueError) as exc: + if staged_dir is not None: + try: + os.replace(staged_dir, step_dir) + except OSError as restore_exc: + raise StepInstallError( + f"Failed to update the step registry for '{step_id}': {exc}. " + f"The package remains staged at '{staged_dir}' because restore " + f"failed: {restore_exc}." + ) from exc + raise StepInstallError( + f"Failed to update the step registry for '{step_id}': {exc}" + ) from exc + return staged_dir, False diff --git a/tests/specify_cli/bundles/test_primitives.py b/tests/specify_cli/bundles/test_primitives.py index a3b4d83f45..8cbcbf85a2 100644 --- a/tests/specify_cli/bundles/test_primitives.py +++ b/tests/specify_cli/bundles/test_primitives.py @@ -12,8 +12,8 @@ import pytest from specify_cli.bundler import BundlerError -from specify_cli.bundles.manifest import ComponentRef from specify_cli.bundles.adapters import DefaultPrimitiveInstaller +from specify_cli.bundles.manifest import ComponentRef from specify_cli.bundles.primitives import ( _ExtensionKindManager, _PresetKindManager, @@ -64,6 +64,22 @@ def test_offline_step_refuses_without_network(tmp_path: Path): manager.install(_component("steps")) +def test_step_manager_delegates_catalog_install_from_bundle_root(tmp_path, monkeypatch): + import specify_cli + + calls: list[tuple[str, Path]] = [] + + def _add(step_id: str) -> None: + calls.append((step_id, Path.cwd())) + + monkeypatch.setattr(specify_cli, "workflow_step_add", _add) + manager = _StepKindManager(tmp_path, allow_network=True) + + manager.install(_component("steps", "catalog-step")) + + assert calls == [("catalog-step", tmp_path)] + + def test_default_installer_threads_allow_network(tmp_path: Path): installer = DefaultPrimitiveInstaller(allow_network=False) with pytest.raises(BundlerError, match="network access is disabled"): @@ -96,10 +112,14 @@ def test_offline_workflow_allows_bundled(tmp_path: Path, monkeypatch): assets, "_locate_bundled_workflow", lambda wid: bundled ) calls: list[tuple] = [] + + def _workflow_add(wid, dev=None, from_url=None): + calls.append((wid, dev, from_url)) + monkeypatch.setattr( specify_cli, "workflow_add", - lambda wid, dev=object(), from_url=object(): calls.append((wid, dev, from_url)), + _workflow_add, ) manager = primitive_manager("workflows", tmp_path, allow_network=False) @@ -483,9 +503,9 @@ def _fake_install(self, *a, **k): def test_refresh_succeeds_and_passes_force_true(tmp_path: Path, monkeypatch): """Regression: bundle update (refresh=True) of an already-installed extension must succeed and pass force=True to install_from_directory.""" + import specify_cli._assets as assets from specify_cli.bundles.installer import install_bundle from specify_cli.bundles.manifest import BundleManifest - import specify_cli._assets as assets from specify_cli.extensions import ExtensionManager bundled = _write_manifest(tmp_path / "ext", "extension", "1.0.0") diff --git a/tests/specify_cli/workflows/step/catalog/test_command_list.py b/tests/specify_cli/workflows/step/catalog/test_command_list.py index 6aff5d3569..34f7076030 100644 --- a/tests/specify_cli/workflows/step/catalog/test_command_list.py +++ b/tests/specify_cli/workflows/step/catalog/test_command_list.py @@ -3,15 +3,13 @@ from __future__ import annotations - - - class TestWorkflowCliAlignment: """CLI alignment with extension/preset commands (#2342).""" def test_step_catalog_list_escapes_rich_markup(self, project_dir, monkeypatch): """User-editable step-catalog name/url/description must not be parsed as Rich markup.""" from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepCatalog diff --git a/tests/specify_cli/workflows/step/catalog/test_registry.py b/tests/specify_cli/workflows/step/catalog/test_registry.py new file mode 100644 index 0000000000..6cce0676e7 --- /dev/null +++ b/tests/specify_cli/workflows/step/catalog/test_registry.py @@ -0,0 +1,33 @@ +"""Persistence tests for the custom step registry.""" + +from __future__ import annotations + +import json + +import pytest + +from specify_cli.workflows.step.catalog import StepRegistry, StepValidationError + + +def _entry(step_id: str) -> dict[str, str]: + return { + "name": "Example", + "version": "1.0.0", + "description": "", + "author": "", + "type_key": step_id, + "source": "local", + } + + +def test_save_keeps_existing_registry_when_json_serialization_fails(project_dir): + registry = StepRegistry(project_dir) + registry.add("first", _entry("first")) + before = registry.registry_path.read_bytes() + registry.data["steps"]["second"] = {"not_json": {1, 2}} + + with pytest.raises(StepValidationError): + registry.save() + + assert registry.registry_path.read_bytes() == before + assert json.loads(before)["steps"]["first"]["type_key"] == "first" diff --git a/tests/specify_cli/workflows/step/test_command_add.py b/tests/specify_cli/workflows/step/test_command_add.py index 3250057e74..b2f03b879f 100644 --- a/tests/specify_cli/workflows/step/test_command_add.py +++ b/tests/specify_cli/workflows/step/test_command_add.py @@ -7,11 +7,11 @@ import pytest - class TestWorkflowStepAddCLI: @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") def test_add_rejects_symlinked_steps_base_dir(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepCatalog @@ -40,13 +40,14 @@ def _fake_get_step_info(self, step_id): def test_add_rejects_oversized_step_response(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app - from specify_cli.workflows import _commands as wf_commands - from specify_cli.workflows.step.catalog import StepCatalog from specify_cli.authentication import http as auth_http + from specify_cli.workflows.step import command_add + from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) - monkeypatch.setattr(wf_commands, "_MAX_WORKFLOW_YAML_BYTES", 100) + monkeypatch.setattr(command_add, "_MAX_STEP_CATALOG_RESPONSE_BYTES", 100) monkeypatch.setattr( StepCatalog, "get_step_info", @@ -96,7 +97,7 @@ def read(self, size=-1): assert result.exit_code != 0 assert ( - "responseexceedsthe100-byteworkflowsizelimit" + "steppackageresponse'exceedsmaximumsizeof100bytes" in "".join(result.output.split()) ) assert not ( @@ -118,9 +119,10 @@ def test_add_rejects_falsy_non_mapping_step_yml( genuinely empty document, so it must be distinguished (via ``yaml.compose``) and rejected too, rather than defaulting to {}.""" from typer.testing import CliRunner + from specify_cli import app - from specify_cli.workflows.step.catalog import StepCatalog from specify_cli.authentication import http as auth_http + from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) monkeypatch.setattr( @@ -441,9 +443,10 @@ def read(self, size=-1): def test_add_rejects_non_string_extra_files_key(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app - from specify_cli.workflows.step.catalog import StepCatalog from specify_cli.authentication import http as auth_http + from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) @@ -505,9 +508,10 @@ def test_add_rejects_invalid_extra_files_path( self, project_dir, monkeypatch, rel_path, expected ): from typer.testing import CliRunner + from specify_cli import app - from specify_cli.workflows.step.catalog import StepCatalog from specify_cli.authentication import http as auth_http + from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) @@ -556,9 +560,10 @@ def _fake_open_url(url, timeout=30, redirect_validator=None): def test_add_rejects_non_string_extra_files_url(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app - from specify_cli.workflows.step.catalog import StepCatalog from specify_cli.authentication import http as auth_http + from specify_cli.workflows.step.catalog import StepCatalog monkeypatch.chdir(project_dir) @@ -692,6 +697,7 @@ def _valid_archive_files(type_key="my-step"): class TestWorkflowStepAddSources: def test_dev_installs_and_loads(self, project_dir, tmp_path, monkeypatch): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows import STEP_REGISTRY, load_custom_steps @@ -715,6 +721,7 @@ def test_dev_installs_and_loads(self, project_dir, tmp_path, monkeypatch): def test_dev_install_list_and_remove(self, project_dir, tmp_path, monkeypatch): from typer.testing import CliRunner + from specify_cli import app package = _write_package(tmp_path, type_key="dev-step") @@ -739,6 +746,7 @@ def test_dev_install_list_and_remove(self, project_dir, tmp_path, monkeypatch): def test_dev_rejects_missing_init(self, project_dir, tmp_path, monkeypatch): from typer.testing import CliRunner + from specify_cli import app package = _write_package(tmp_path, type_key="dev-step") @@ -755,6 +763,7 @@ def test_dev_rejects_symlinked_source_root(self, project_dir, tmp_path, monkeypa if not hasattr(os, "symlink"): pytest.skip("symlinks are unavailable") from typer.testing import CliRunner + from specify_cli import app package = _write_package(tmp_path, type_key="dev-step") @@ -770,6 +779,7 @@ def test_dev_rejects_symlinked_source_root(self, project_dir, tmp_path, monkeypa def test_dev_and_from_are_mutually_exclusive(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) @@ -792,6 +802,7 @@ def test_dev_and_from_are_mutually_exclusive(self, project_dir, monkeypatch): @pytest.mark.parametrize("option", ["--dev", "--from"]) def test_empty_source_value_rejected(self, project_dir, monkeypatch, option): from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) @@ -802,6 +813,7 @@ def test_empty_source_value_rejected(self, project_dir, monkeypatch, option): def test_force_replaces_installed_package(self, project_dir, tmp_path, monkeypatch): from typer.testing import CliRunner + from specify_cli import app package = _write_package(tmp_path, type_key="dev-step", init_body="# old\n") @@ -834,6 +846,7 @@ def test_force_replaces_installed_package(self, project_dir, tmp_path, monkeypat def test_force_replaces_orphaned_directory(self, project_dir, tmp_path, monkeypatch): from typer.testing import CliRunner + from specify_cli import app orphan = ( @@ -858,6 +871,7 @@ def test_from_denied_confirmation_issues_no_request( ): import typer from typer.testing import CliRunner + from specify_cli import app from specify_cli.authentication import http as auth_http @@ -900,6 +914,7 @@ def test_from_archive_installs( ): import typer from typer.testing import CliRunner + from specify_cli import app from specify_cli.authentication import http as auth_http @@ -924,6 +939,7 @@ def test_from_archive_installs( def test_from_rejects_non_https(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) @@ -943,6 +959,7 @@ def test_from_rejects_non_https(self, project_dir, monkeypatch): def test_from_rejects_malformed_url(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) @@ -963,6 +980,7 @@ def test_from_rejects_malformed_url(self, project_dir, monkeypatch): def test_from_rejects_redirect_to_non_https(self, project_dir, monkeypatch): import typer from typer.testing import CliRunner + from specify_cli import app from specify_cli.authentication import http as auth_http @@ -993,6 +1011,7 @@ def test_from_rejects_redirect_to_non_https(self, project_dir, monkeypatch): def test_from_rejects_non_archive_body(self, project_dir, monkeypatch): import typer from typer.testing import CliRunner + from specify_cli import app from specify_cli.authentication import http as auth_http @@ -1025,6 +1044,7 @@ def test_from_rejects_archive_with_unrelated_siblings( ): import typer from typer.testing import CliRunner + from specify_cli import app from specify_cli.authentication import http as auth_http @@ -1058,11 +1078,69 @@ def test_from_rejects_archive_with_unrelated_siblings( assert result.exit_code != 0 assert "exactly one top-level" in result.output + def test_from_rejects_original_url_format_mismatch_after_redirect( + self, project_dir, monkeypatch + ): + import typer + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: True) + body = _make_tar_gz(_valid_archive_files()) + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None, extra_headers=None: ( + _ArchiveResponse("https://example.com/download", body, None) + ), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step", "--from", "https://example.com/pkg.zip"] + ) + + assert result.exit_code != 0 + assert "Archive format mismatch" in result.output + + def test_from_escapes_installed_name(self, project_dir, monkeypatch): + import typer + from typer.testing import CliRunner + + from specify_cli import app + from specify_cli.authentication import http as auth_http + + monkeypatch.chdir(project_dir) + monkeypatch.setattr(typer, "confirm", lambda *a, **k: True) + body = _make_zip( + { + "step.yml": "step:\n type_key: my-step\n name: '[/]'\n", + "__init__.py": "# init\n", + } + ) + monkeypatch.setattr( + auth_http, + "open_url", + lambda url, timeout=30, redirect_validator=None, extra_headers=None: ( + _ArchiveResponse(url, body, "application/zip") + ), + ) + + result = CliRunner().invoke( + app, ["workflow", "step", "add", "my-step", "--from", "https://example.com/pkg.zip"] + ) + + assert result.exit_code == 0, result.output + assert "[/]" in result.output + def test_from_denied_when_already_installed_errors_before_prompt( self, project_dir, tmp_path, monkeypatch ): import typer from typer.testing import CliRunner + from specify_cli import app package = _write_package(tmp_path, type_key="my-step") @@ -1139,6 +1217,7 @@ def test_dev_install_loads_runs_and_removes( self, project_dir, tmp_path, monkeypatch ): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows import load_custom_steps diff --git a/tests/specify_cli/workflows/step/test_command_info.py b/tests/specify_cli/workflows/step/test_command_info.py index b63d5f8f78..62391f5f03 100644 --- a/tests/specify_cli/workflows/step/test_command_info.py +++ b/tests/specify_cli/workflows/step/test_command_info.py @@ -1,15 +1,12 @@ """Command-focused workflow tests.""" -from __future__ import annotations - - - +from typing import ClassVar class TestWorkflowStepRichMarkup: """Step discovery commands render metadata as literal text.""" - METADATA = { + METADATA: ClassVar[dict[str, str]] = { "id": "[magenta]step-id[/magenta]", "name": "[red]Step Name[/red]", "version": "[green]1.0.0[/green]", @@ -21,6 +18,7 @@ def test_info_escapes_catalog_metadata( self, project_dir, monkeypatch ): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepCatalog, StepRegistry @@ -43,6 +41,7 @@ def test_info_escapes_catalog_metadata( def test_info_escapes_missing_step_id(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepCatalog, StepRegistry @@ -64,6 +63,7 @@ def test_info_escapes_missing_step_id(self, project_dir, monkeypatch): def test_info_prints_local_source(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepRegistry @@ -86,6 +86,7 @@ def test_info_prints_local_source(self, project_dir, monkeypatch): def test_info_prints_catalog_source_with_name(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepRegistry diff --git a/tests/specify_cli/workflows/step/test_command_list.py b/tests/specify_cli/workflows/step/test_command_list.py index e225eea93d..ff9ba007a7 100644 --- a/tests/specify_cli/workflows/step/test_command_list.py +++ b/tests/specify_cli/workflows/step/test_command_list.py @@ -1,15 +1,12 @@ """Command-focused workflow tests.""" -from __future__ import annotations - - - +from typing import ClassVar class TestWorkflowStepRichMarkup: """Step discovery commands render metadata as literal text.""" - METADATA = { + METADATA: ClassVar[dict[str, str]] = { "id": "[magenta]step-id[/magenta]", "name": "[red]Step Name[/red]", "version": "[green]1.0.0[/green]", @@ -21,6 +18,7 @@ def test_list_escapes_installed_metadata( self, project_dir, monkeypatch ): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepRegistry diff --git a/tests/specify_cli/workflows/step/test_command_remove.py b/tests/specify_cli/workflows/step/test_command_remove.py index a6bd5781c0..5e902b733d 100644 --- a/tests/specify_cli/workflows/step/test_command_remove.py +++ b/tests/specify_cli/workflows/step/test_command_remove.py @@ -7,7 +7,6 @@ import pytest - class TestWorkflowStepRemoveCLI: """Test the 'specify workflow step remove' CLI command edge cases.""" @@ -17,6 +16,7 @@ def test_remove_orphaned_directory(self, project_dir, monkeypatch): This covers the case where the registry was reset due to corruption. """ from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) @@ -40,6 +40,7 @@ def test_remove_orphaned_directory(self, project_dir, monkeypatch): def test_remove_not_installed(self, project_dir, monkeypatch): """step remove fails cleanly when neither directory nor registry entry exist.""" from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) @@ -53,6 +54,7 @@ def test_remove_not_installed(self, project_dir, monkeypatch): def test_remove_registered_step(self, project_dir, monkeypatch): """step remove works normally when both directory and registry entry exist.""" from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepRegistry @@ -79,6 +81,7 @@ def test_remove_registered_step(self, project_dir, monkeypatch): @pytest.mark.skipif(not hasattr(os, "symlink"), reason="symlinks are unavailable") def test_remove_rejects_symlinked_steps_base_dir(self, project_dir, monkeypatch): from typer.testing import CliRunner + from specify_cli import app monkeypatch.chdir(project_dir) diff --git a/tests/specify_cli/workflows/step/test_command_search.py b/tests/specify_cli/workflows/step/test_command_search.py index bd9982da0c..60f7852c7c 100644 --- a/tests/specify_cli/workflows/step/test_command_search.py +++ b/tests/specify_cli/workflows/step/test_command_search.py @@ -1,15 +1,12 @@ """Command-focused workflow tests.""" -from __future__ import annotations - - - +from typing import ClassVar class TestWorkflowStepRichMarkup: """Step discovery commands render metadata as literal text.""" - METADATA = { + METADATA: ClassVar[dict[str, str]] = { "id": "[magenta]step-id[/magenta]", "name": "[red]Step Name[/red]", "version": "[green]1.0.0[/green]", @@ -21,6 +18,7 @@ def test_search_escapes_catalog_metadata( self, project_dir, monkeypatch ): from typer.testing import CliRunner + from specify_cli import app from specify_cli.workflows.step.catalog import StepCatalog diff --git a/tests/specify_cli/workflows/step/test_installer.py b/tests/specify_cli/workflows/step/test_installer.py index ecbacc4422..18fbfd417d 100644 --- a/tests/specify_cli/workflows/step/test_installer.py +++ b/tests/specify_cli/workflows/step/test_installer.py @@ -388,6 +388,38 @@ def test_catalog_provenance_shape(tmp_path, project_dir): assert entry["author"] == "author" +@pytest.mark.parametrize( + "field,value", + [ + ("name", "2026-09-24"), + ("version", "2026-09-24"), + ("description", "2026-09-24"), + ("author", "2026-09-24"), + ], +) +def test_rejects_non_string_persisted_metadata_before_publication( + tmp_path, project_dir, field, value +): + pkg = _write_package(tmp_path / "pkg") + (pkg / "step.yml").write_text( + f"step:\n type_key: my-step\n {field}: {value}\n", encoding="utf-8" + ) + + with pytest.raises(installer.StepInstallError, match="must be a string"): + installer.install_step_package(project_dir, "my-step", pkg, source="local") + + assert not (_steps_dir(project_dir) / "my-step").exists() + + +def test_rejects_unknown_source_before_creating_steps_dir(tmp_path, project_dir): + pkg = _write_package(tmp_path / "pkg") + + with pytest.raises(installer.StepInstallError, match="source"): + installer.install_step_package(project_dir, "my-step", pkg, source="unknown") + + assert not _steps_dir(project_dir).exists() + + # --------------------------------------------------------------------------- # Failure handling # --------------------------------------------------------------------------- @@ -449,6 +481,22 @@ def _validate(package_dir, step_id): ) == "# old\n" +def test_uses_metadata_from_staged_copy(tmp_path, project_dir, monkeypatch): + pkg = _write_package(tmp_path / "pkg") + original_copy = installer._copy_package_tree + + def _copy_then_change(source, target): + original_copy(source, target) + (target / "step.yml").write_text( + "step:\n type_key: my-step\n name: Staged Name\n", encoding="utf-8" + ) + + monkeypatch.setattr(installer, "_copy_package_tree", _copy_then_change) + entry = installer.install_step_package(project_dir, "my-step", pkg, source="local") + + assert entry["name"] == "Staged Name" + + def test_force_registry_failure_warns_reinstall( tmp_path, project_dir, monkeypatch ): diff --git a/tests/specify_cli/workflows/test_custom_steps.py b/tests/specify_cli/workflows/test_custom_steps.py new file mode 100644 index 0000000000..ed0d7e0650 --- /dev/null +++ b/tests/specify_cli/workflows/test_custom_steps.py @@ -0,0 +1,49 @@ +"""Runtime freshness tests for project-local custom workflow steps.""" + +from __future__ import annotations + +import shutil +from pathlib import Path + +from specify_cli.workflows import STEP_REGISTRY, load_custom_steps + + +def _write_step(project_root: Path, marker: str) -> None: + step_dir = project_root / ".specify" / "workflows" / "steps" / "custom" + step_dir.mkdir(parents=True) + (step_dir / "step.yml").write_text( + "step:\n type_key: custom\n", encoding="utf-8" + ) + (step_dir / "__init__.py").write_text( + "from specify_cli.workflows.base import StepBase, StepResult\n\n" + "class Custom(StepBase):\n" + " type_key = 'custom'\n" + " def execute(self, config, context):\n" + f" return StepResult(output={{'marker': {marker!r}}})\n", + encoding="utf-8", + ) + + +def test_custom_steps_refresh_for_active_project(tmp_path): + project_a = tmp_path / "a" + project_b = tmp_path / "b" + _write_step(project_a, "a") + _write_step(project_b, "b") + + assert load_custom_steps(project_a) == ["custom"] + assert STEP_REGISTRY["custom"].execute({}, None).output == {"marker": "a"} + + assert load_custom_steps(project_b) == ["custom"] + assert STEP_REGISTRY["custom"].execute({}, None).output == {"marker": "b"} + + +def test_removed_custom_step_is_not_retained(tmp_path): + project = tmp_path / "project" + _write_step(project, "old") + assert load_custom_steps(project) == ["custom"] + + step_dir = project / ".specify" / "workflows" / "steps" / "custom" + shutil.rmtree(step_dir) + + assert load_custom_steps(project) == [] + assert "custom" not in STEP_REGISTRY