github/github-spec-kit
1
0
Fork 0
mirror of https://github.com/github/spec-kit.git synced 2026-08-03 06:26:30 +08:00
Code Issues Packages Projects Releases Wiki Activity

Fix installed-workflow ownership/disabled bypass and resume enforcement

Browse Source 
Address 3 current Copilot review findings on the disabled-workflow guard
in `workflow run`/`workflow resume`:

- The lexical `.specify/workflows/<id>` ownership scan stopped at the
  first match scanning from the start of the path. A nested project
  living beneath an outer installed workflow's own directory tree (reusing
  the same segment names) was attributed to the wrong (outer) workflow
  and ID, gating the run on an unrelated workflow's disabled state.
  `_scan_for_workflow_owner` now scans from the end so the nearest
  (innermost) owner always wins.

- A path with no `.specify/workflows` segments of its own (e.g.
  `/tmp/alias.yml`) that is itself a symlink resolving *into* installed
  storage bypassed the disabled check entirely, since only the raw
  lexical path was inspected. `_resolve_installed_workflow_ownership` now
  additionally resolves the real path when the lexical scan finds no
  owner and re-runs the same scan against it, so an outward-pointing
  alias into a disabled workflow is caught too. Genuinely standalone
  external files (no symlink anywhere on the path) are unaffected.

- `workflow resume` bypassed the disabled check altogether: engine.resume()
  replays a persisted run directly from disk with no registry awareness.
  RunState now optionally persists `installed_workflow_id` and
  `installed_registry_root` at run start (set by workflow_run when the
  source resolved to an installed ID); `workflow_resume` pre-loads the
  run state and re-checks the registry's *current* disabled state before
  calling engine.resume(), mirroring workflow_run's own guard. Both new
  fields default to None via RunState.load()'s `.get()`, so runs from a
  direct/non-installed source, and any run persisted before this schema
  addition, resume exactly as before.

The ownership-mapping logic (previously inlined in workflow_run) is
extracted into `_resolve_installed_workflow_ownership` /
`_scan_for_workflow_owner` so both the lexical and resolved-path cases
share the same scan and the existing inward-symlink-component refusal.

Assisted-by: GitHub Copilot (model: Claude Sonnet 5, autonomous)
Co-authored-by: Copilot <223556219+Copilot@users.noreply.github.com>
This commit is contained in:
marcelsafin
2026-07-11 14:47:21 +02:00
parent 915caf8da9
commit 792450e259
3 changed files with 301 additions and 83 deletions
Download Patch File Download Diff File Expand all files Collapse all files

193
src/specify_cli/workflows/_commands.py
View File

@@ -123,6 +123,109 @@ def _reject_unsafe_workflow_storage(project_root: Path) -> None:
)
def _scan_for_workflow_owner(parts: tuple[str, ...]) -> tuple[int, str] | None:
"""Find the *nearest* (innermost) ``.specify/workflows/<id>`` owner in
*parts*, scanning from the end of the path.
Scanning from the end (rather than stopping at the first match from the
start) matters for a project nested beneath an unrelated outer path that
happens to reuse the same ``.specify``/``workflows`` segment names: the
first-from-start match would pick the outer directory and the wrong
workflow ID, silently missing the real (inner) owner's disabled check.
Returns ``(i, workflow_id)`` where ``i`` is the index of the owning
``.specify`` segment, or ``None`` if no owner segment is present.
"""
for i in range(len(parts) - 3, -1, -1):
if parts[i] == ".specify" and parts[i + 1] == "workflows":
return i, parts[i + 2]
return None
def _resolve_installed_workflow_ownership(
source_path: Path, err
) -> tuple[Path | None, str | None]:
"""Map a direct ``workflow.yml`` *source_path* back to the installed
workflow (``registry_root``, ``registered_id``) it belongs to, if any.
A path can point at installed storage two ways, both of which must
receive the same registry disabled-check:
1. Lexically: the path's own (symlink-preserving) segments literally
contain ``.specify/workflows/<id>`` -- collapsing ``..``/``.`` but
never resolving symlinks, so a symlinked ``workflow.yml`` leaf (or
symlinked ``<id>`` directory) inside the owned tree is caught by the
inward-symlink refusal below rather than silently followed.
2. Via an outward-pointing alias: *source_path* (or one of its parent
directories) is itself a symlink whose *resolved* target lands
inside some project's ``.specify/workflows/<id>/workflow.yml``, even
though the raw path used to invoke the command has no such segments
at all (e.g. ``/tmp/alias.yml`` -> a disabled installed workflow's
real file). Only the lexical case also runs the inward-symlink
refusal: the resolved case's segments are real by construction (an
already-fully-resolved path cannot itself contain a symlink
component), so that check would be a vacuous no-op there.
Returns ``(None, None)`` when neither applies -- a genuinely standalone
external workflow file, which is allowed to run unchecked.
"""
lexical = Path(os.path.normpath(str(source_path.absolute())))
parts = lexical.parts
match = _scan_for_workflow_owner(parts)
if match is not None:
i, registered_id = match
registry_root = Path(*parts[:i]) if i else Path(lexical.anchor or ".")
# The path-derived registry_root here may differ from the cwd's
# project_root already checked by _reject_unsafe_workflow_storage
# (e.g. this path points into another project entirely, or this
# project's own .specify is itself a symlink to an
# attacker-controlled tree) -- check it explicitly rather than
# trusting that cwd-scoped guard, and don't rely on
# WorkflowRegistry's own symlinked-parent handling as the safety
# signal here: it fails closed by raising OSError at construction
# time (see catalog.py's _load), but that surfaces as an opaque
# exception rather than this guard's clean, specific CLI error for
# the actual owning project root.
_reject_unsafe_dir(registry_root / ".specify", ".specify")
_reject_unsafe_dir(
registry_root / ".specify" / "workflows", ".specify/workflows"
)
# A legitimately installed workflow's own directory tree never
# contains a symlink (workflow add/remove both refuse one at
# install time); one appearing here means the file actually loaded
# below would not be the file this ownership match is based on, so
# refuse rather than silently mismatch.
for k in range(i + 2, len(parts) + 1):
if Path(*parts[:k]).is_symlink():
err.print(
"[red]Error:[/red] Refusing to run: "
f".specify/workflows/{_escape_markup(registered_id)} "
"contains a symlinked path component"
)
raise typer.Exit(1)
return registry_root, registered_id
# No lexical owner segments. source_path (or a parent directory) might
# still be a symlink whose resolved target lands inside some project's
# installed workflow storage under a completely unrelated-looking path
# -- check that too so an alias into a disabled workflow can't dodge
# the registry guard.
try:
resolved = source_path.resolve(strict=False)
except OSError:
return None, None
if resolved == lexical:
# Nothing on this path is a symlink; already covered above.
return None, None
resolved_parts = resolved.parts
resolved_match = _scan_for_workflow_owner(resolved_parts)
if resolved_match is None:
return None, None
i, registered_id = resolved_match
registry_root = Path(*resolved_parts[:i]) if i else Path(resolved.anchor or ".")
return registry_root, registered_id
_WORKFLOW_ID_PATTERN = re.compile(r"^[a-z0-9](?:[a-z0-9-]*[a-z0-9])?$")
_RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"runs", "steps"})
@@ -580,48 +683,13 @@ def workflow_run(
registered_id = source
else:
# A direct YAML path may still point at an installed workflow's own
# file; map it back to its owning project and ID from the *lexical*
# path (collapsing .. / . without resolving symlinks) rather than
# resolve(): resolving first would follow a symlinked workflow.yml
# out of .specify/workflows, fail to find an owner, and let
# engine.load_workflow below run the symlink target unchecked --
# silently bypassing a disabled workflow's guard.
lexical = Path(os.path.normpath(str(source_path.absolute())))
parts = lexical.parts
for i in range(len(parts) - 2):
if parts[i] == ".specify" and parts[i + 1] == "workflows":
registry_root = Path(*parts[:i]) if i else Path(lexical.anchor or ".")
registered_id = parts[i + 2]
# The path-derived registry_root here may differ from the
# cwd's project_root already checked by
# _reject_unsafe_workflow_storage above (e.g. this path
# points into another project entirely, or this project's
# own .specify is itself a symlink to an attacker-controlled
# tree) -- check it explicitly rather than trusting that
# cwd-scoped guard, and don't rely on WorkflowRegistry's own
# symlinked-parent handling below as the safety signal here:
# it now fails closed by raising OSError at construction
# time (see catalog.py's _load), but that surfaces as an
# opaque exception rather than this guard's clean, specific
# CLI error for the actual owning project root.
_reject_unsafe_dir(registry_root / ".specify", ".specify")
_reject_unsafe_dir(
registry_root / ".specify" / "workflows", ".specify/workflows"
)
# A legitimately installed workflow's own directory tree
# never contains a symlink (workflow add/remove both refuse
# one at install time); one appearing here means the file
# actually loaded below would not be the file this ownership
# match is based on, so refuse rather than silently mismatch.
for k in range(i + 2, len(parts) + 1):
if Path(*parts[:k]).is_symlink():
err.print(
"[red]Error:[/red] Refusing to run: "
f".specify/workflows/{_escape_markup(registered_id)} "
"contains a symlinked path component"
)
raise typer.Exit(1)
break
# file (lexically, or via a symlinked alias pointing into installed
# storage); map it back to its owning project and ID so the
# disabled check below can't be silently bypassed.
owner_root, owner_id = _resolve_installed_workflow_ownership(source_path, err)
if owner_id is not None:
registry_root = owner_root
registered_id = owner_id
if registered_id is not None:
installed_meta = _open_workflow_registry(registry_root, err).get(registered_id)
@@ -658,7 +726,12 @@ def workflow_run(
try:
with _stdout_to_stderr_when(json_output):
state = engine.execute(definition, inputs)
state = engine.execute(
definition,
inputs,
installed_workflow_id=registered_id,
installed_registry_root=registry_root if registered_id else None,
)
except ValueError as exc:
err.print(f"[red]Error:[/red] {exc}")
raise typer.Exit(1)
@@ -700,7 +773,7 @@ def workflow_resume(
):
"""Resume a paused or failed workflow run."""
from . import load_custom_steps
from .engine import WorkflowEngine
from .engine import RunState, WorkflowEngine
project_root = _require_specify_project()
load_custom_steps(project_root)
@@ -711,6 +784,38 @@ def workflow_resume(
inputs = _parse_input_values(input_values, json_output=json_output)
err = _error_console(json_output)
# Pre-load the persisted run state so a run started from an installed
# workflow that has since been disabled cannot resume unchecked --
# engine.resume() replays the run directly from disk with no registry
# awareness at all, which would otherwise bypass the same disabled
# guard `workflow run` enforces. Runs without installed_workflow_id
# (a direct/non-installed source, or a run persisted before this field
# existed) are unaffected and resume exactly as before.
try:
pre_state = RunState.load(run_id, project_root)
except FileNotFoundError:
err.print(f"[red]Error:[/red] Run not found: {run_id}")
raise typer.Exit(1)
except ValueError as exc:
err.print(f"[red]Error:[/red] {exc}")
raise typer.Exit(1)
if pre_state.installed_workflow_id is not None:
owner_root = (
Path(pre_state.installed_registry_root)
if pre_state.installed_registry_root
else project_root
)
installed_meta = _open_workflow_registry(owner_root, err).get(
pre_state.installed_workflow_id
)
if isinstance(installed_meta, dict) and not installed_meta.get("enabled", True):
err.print(
f"[red]Error:[/red] Workflow '{_escape_markup(pre_state.installed_workflow_id)}' is disabled. "
f"Enable with: specify workflow enable {_escape_markup(pre_state.installed_workflow_id)}"
)
raise typer.Exit(1)
try:
with _stdout_to_stderr_when(json_output):
state = engine.resume(run_id, inputs or None)

29
src/specify_cli/workflows/engine.py
View File

@@ -428,6 +428,8 @@ class RunState:
run_id: str | None = None,
workflow_id: str = "",
project_root: Path | None = None,
installed_workflow_id: str | None = None,
installed_registry_root: str | None = None,
) -> None:
# ``run_id is None`` (omitted) → auto-generate. An explicit empty
# string is *not* the same as "omitted" and must be validated like
@@ -441,6 +443,15 @@ class RunState:
self._validate_run_id(self.run_id)
self.workflow_id = workflow_id
self.project_root = project_root or Path(".")
# Identifies the installed workflow (if any) this run was started
# from, and the project root that owns its registry — set by
# execute() when the source was resolved to an installed ID (see
# workflow_run's ownership mapping). None for a direct/non-installed
# YAML source, and for any run persisted before this field existed
# (load() defaults it to None), preserving old runs' existing
# resume behavior unchanged.
self.installed_workflow_id = installed_workflow_id
self.installed_registry_root = installed_registry_root
self.status = RunStatus.CREATED
self.current_step_index = 0
self.current_step_id: str | None = None
@@ -503,6 +514,8 @@ class RunState:
state_data = {
"run_id": self.run_id,
"workflow_id": self.workflow_id,
"installed_workflow_id": self.installed_workflow_id,
"installed_registry_root": self.installed_registry_root,
"status": self.status.value,
"current_step_index": self.current_step_index,
"current_step_id": self.current_step_id,
@@ -559,6 +572,8 @@ class RunState:
run_id=state_data["run_id"],
workflow_id=state_data["workflow_id"],
project_root=project_root,
installed_workflow_id=state_data.get("installed_workflow_id"),
installed_registry_root=state_data.get("installed_registry_root"),
)
state.status = RunStatus(state_data["status"])
state.current_step_index = state_data.get("current_step_index", 0)
@@ -654,6 +669,8 @@ class WorkflowEngine:
definition: WorkflowDefinition,
inputs: dict[str, Any] | None = None,
run_id: str | None = None,
installed_workflow_id: str | None = None,
installed_registry_root: Path | None = None,
) -> RunState:
"""Execute a workflow definition.
@@ -665,6 +682,12 @@ class WorkflowEngine:
User-provided input values.
run_id:
Optional run ID (uses SPECKIT_WORKFLOW_RUN_ID when set, otherwise auto-generated).
installed_workflow_id, installed_registry_root:
When the run was started from an installed workflow (as opposed
to a direct/non-installed YAML source), identifies it and its
owning registry root so a later ``resume`` can re-check the
registry's current disabled state before continuing — see
``workflow_resume``.
Returns
-------
@@ -682,6 +705,12 @@ class WorkflowEngine:
run_id=effective_run_id,
workflow_id=definition.id,
project_root=self.project_root,
installed_workflow_id=installed_workflow_id,
installed_registry_root=(
str(installed_registry_root)
if installed_registry_root is not None
else None
),
)
# Persist a copy of the workflow definition so resume can

162
tests/test_workflows.py
View File

@@ -7805,6 +7805,46 @@ steps:
assert result.exit_code == 0, result.output
assert WorkflowRegistry(project_dir).is_installed("align-wf")
def test_add_from_url_temp_cleanup_failure_after_success_still_exits_zero(
self, project_dir, monkeypatch
):
"""An OSError while deleting the --from download's temp file after
_validate_and_install_local() has already committed the file and
registry entry must not surface as an unhandled failure for an
install that already succeeded -- it must be a warning, exit 0."""
from unittest.mock import patch
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry
monkeypatch.chdir(project_dir)
data = self.WORKFLOW_YAML.format(version="1.0.0").encode()
runner = CliRunner()
import tempfile
real_unlink = Path.unlink
def unlink_boom(self_path, *args, **kwargs):
if self_path.suffix == ".yml" and self_path.parent == Path(tempfile.gettempdir()):
raise OSError("permission denied")
return real_unlink(self_path, *args, **kwargs)
with patch(
"specify_cli.authentication.http.open_url",
side_effect=lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(data, url),
), pytest.MonkeyPatch.context() as mp:
mp.setattr(Path, "unlink", unlink_boom)
result = runner.invoke(
app,
["workflow", "add", "align-wf", "--from", "https://example.com/workflow.yml"],
)
assert result.exit_code == 0, result.output
assert "Warning" in result.output
assert "permissiondenied" in "".join(result.output.split())
assert WorkflowRegistry(project_dir).is_installed("align-wf")
def test_add_from_url_id_mismatch_errors(self, project_dir, monkeypatch):
from unittest.mock import patch
from typer.testing import CliRunner
@@ -8042,6 +8082,7 @@ steps:
WorkflowRegistry(project_dir)
assert not (outside / "workflows").exists()
@pytest.mark.skipif(sys.platform == "win32", reason="chmod mode bits not reliable on Windows")
def test_registry_save_preserves_existing_file_mode(self, project_dir):
"""A registry shared as 0640/0644 must keep that mode after a save,
not be silently replaced by mkstemp's 0600 default -- otherwise
@@ -8058,6 +8099,7 @@ steps:
mode = stat.S_IMODE(registry.registry_path.stat().st_mode)
assert mode == 0o644, f"expected 0644, got {oct(mode)}"
@pytest.mark.skipif(sys.platform == "win32", reason="chmod mode bits not reliable on Windows")
def test_registry_save_on_new_registry_uses_secure_default_mode(self, project_dir):
"""A brand-new registry file (no prior mode to preserve) should keep
mkstemp's secure 0600 default rather than something more permissive."""
@@ -8933,6 +8975,7 @@ steps:
assert seen["validator"] is _reject_insecure_download_redirect
@pytest.mark.skipif(sys.platform == "win32", reason="chmod mode bits not reliable on Windows")
def test_registry_save_failure_preserves_file_on_disk(self, project_dir, monkeypatch):
"""A failed dump must not truncate the persisted registry, and must
not alter its on-disk mode either -- the chmod-to-match-existing-mode
@@ -9463,55 +9506,96 @@ steps:
assert result.exit_code != 0
assert "Failed to update" in result.output
def test_update_survives_oserror_from_backup_read(self, project_dir, monkeypatch):
"""OSError while reading the backup for one workflow must not abort the whole update."""
import json
from pathlib import Path
def test_update_registry_save_failure_restores_prior_file_without_redundant_write(
self, project_dir, monkeypatch
):
"""A registry.add() save failure during `workflow update` must be
fully restored by _install_workflow_from_catalog's own atomic
rollback (rename-based, not a byte-level rewrite). The outer
workflow_update loop must not perform any redundant write of its
own onto the destination file -- that write happened only after
typer.Exit already unwound, could itself fail/truncate the safely
preserved file, and is provably unnecessary here since the inner
transaction already restored it via rename."""
from typer.testing import CliRunner
from specify_cli import app
from specify_cli.workflows.catalog import WorkflowRegistry, WorkflowCatalog
from specify_cli.workflows.catalog import WorkflowCatalog, WorkflowRegistry
monkeypatch.chdir(project_dir)
registry_path = WorkflowRegistry(project_dir).registry_path
registry_path.parent.mkdir(parents=True, exist_ok=True)
registry_path.write_text(
json.dumps(
{
"schema_version": "1.0",
"workflows": {
"wobbly": {
"name": "Wobbly",
"version": "0.0.1",
"source": "catalog",
"url": "https://example.com/wobbly.yml",
},
},
}
),
encoding="utf-8",
)
# An existing installed file whose read_bytes will raise OSError.
wf_file = project_dir / ".specify" / "workflows" / "wobbly" / "workflow.yml"
wf_file.parent.mkdir(parents=True, exist_ok=True)
wf_file.write_bytes(b"schema_version: '1.0'\nworkflow:\n id: wobbly\n name: Wobbly\n version: 0.0.1\nsteps: []\n")
monkeypatch.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {"version": "9.9.9", "url": "https://example.com/wobbly.yml", "_install_allowed": True},
lambda self, wid: {
"id": wid,
"name": "Align Workflow",
"version": "1.0.0",
"url": "https://example.com/workflow.yml",
"_install_allowed": True,
"_catalog_name": "test-catalog",
},
)
real_read = Path.read_bytes
def _boom(self, *args, **kwargs):
if self.name == "workflow.yml" and "wobbly" in str(self):
raise OSError("simulated permission denied")
return real_read(self, *args, **kwargs)
monkeypatch.setattr(Path, "read_bytes", _boom)
original_data = self.WORKFLOW_YAML.format(version="1.0.0").encode()
runner = CliRunner()
result = runner.invoke(app, ["workflow", "update"], input="y\n")
assert result.exit_code != 0, result.output
assert "Filesystem error" in result.output
with pytest.MonkeyPatch.context() as mp:
mp.setattr(
"specify_cli.authentication.http.open_url",
lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(
original_data, url
),
)
result = runner.invoke(app, ["workflow", "add", "align-wf"])
assert result.exit_code == 0, result.output
dest_file = project_dir / ".specify" / "workflows" / "align-wf" / "workflow.yml"
assert dest_file.read_bytes() == original_data
new_data = self.WORKFLOW_YAML.format(version="2.0.0").encode()
def boom_save(self):
raise OSError("disk full")
dest_writes: list[bytes] = []
real_write_bytes = Path.write_bytes
resolved_dest_file = dest_file.resolve()
def tracking_write_bytes(self_path, data, *args, **kwargs):
if self_path.resolve() == resolved_dest_file:
dest_writes.append(data)
return real_write_bytes(self_path, data, *args, **kwargs)
with pytest.MonkeyPatch.context() as mp:
mp.setattr(
WorkflowCatalog,
"get_workflow_info",
lambda self, wid: {
"id": wid,
"name": "Align Workflow",
"version": "2.0.0",
"url": "https://example.com/workflow.yml",
"_install_allowed": True,
"_catalog_name": "test-catalog",
},
)
mp.setattr(
"specify_cli.authentication.http.open_url",
lambda url, timeout=None, extra_headers=None, redirect_validator=None: self._FakeResponse(
new_data, url
),
)
mp.setattr(WorkflowRegistry, "save", boom_save)
mp.setattr(Path, "write_bytes", tracking_write_bytes)
result = runner.invoke(app, ["workflow", "update"], input="y\n")
assert result.exit_code != 0
assert "Failed to update" in result.output
# No redundant/second write of the destination file was attempted --
# the inner atomic commit/rollback (rename-based) is the only thing
# that ever touches it.
assert dest_writes == []
assert dest_file.read_bytes() == original_data
registry = WorkflowRegistry(project_dir)
assert registry.is_installed("align-wf")
assert registry.get("align-wf")["version"] == "1.0.0"
def test_enable_disable_corrupted_registry_entry_errors(self, project_dir, monkeypatch):
import json