mirror of
https://github.com/github/spec-kit.git
synced 2026-08-03 06:26:30 +08:00
Add --extension flag to specify init for opting into extensions at init time (#3914)
* Add --extension flag to specify init for installing extensions at init time Adds a repeatable --extension flag to `specify init` so users can opt into extensions (bundled name, local path, or HTTPS URL) during initialization, without a separate `specify extension add` step. - New `_install_extension_during_init` helper in commands/init.py that auto-detects source type (URL / local path / bundled name / catalog) and installs via ExtensionManager. Failures are non-fatal and recorded in the tracker without aborting init. - Extension tracker steps are pre-registered before the Live context and run after preset install, before finalize. - Five new tests in TestExtensionFlag covering bundled name, multiple extensions, local absolute path, unknown extension (graceful error), and combination with --preset. Rebased onto upstream/main and adapted to the refactored init command (moved to src/specify_cli/commands/init.py) from stale PR #2396. Assisted-by: GitHub Copilot (model: Claude Opus 4.8, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> * Address review: reuse hardened downloader, refresh events, escape labels, fix bundler call Responds to review feedback on #3914 and fixes CI (pytest bundler failure). - Extract shared `install_extension_from_url` helper in extensions/_commands.py that reuses the authenticated, redirect-guarded, bounded (50 MiB) download and TOCTOU-safe transient archive used by `extension add --from`. Both `extension add --from` and `specify init --extension <url>` now go through this single downloader instead of a second raw urlopen path. - Refresh native event configuration once after successful extension installs during init (mirrors `_refresh_events_and_warn` in the add path) so an extension declaring `events:` has its hooks activated. - Escape user-controlled extension specs and error text before interpolating them into StepTracker labels (Rich markup injection). - Pass `extensions=None` from bundler's `_run_init` so the init callback no longer receives the typer OptionInfo sentinel ('OptionInfo' object is not iterable), which broke `test_install_initializes_uninitialized_project`. - Add init URL coverage in TestExtensionFlag: non-HTTPS rejection and a successful HTTPS ZIP install with download-cache cleanup assertion. Assisted-by: GitHub Copilot (model: Claude Opus 4.8, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> * Add default-deny trust confirmation for URL extension installs at init URL-based --extension installs now require explicit trust, matching the `extension add --from` posture. Interactive sessions show an "Untrusted Source" panel and prompt (default no); non-interactive sessions deny by default unless --trust-extension-urls is passed. Trust is resolved before the Live display since the prompt can't be answered under the spinner. - Add --trust-extension-urls option and _ext_spec_is_url / _confirm_extension_url_trust helpers - Skip (not abort) unconfirmed URL extensions, consistent with other non-fatal extension failures - Pass trust_extension_urls=False from the bundler init callback - Add tests for deny-by-default, interactive confirm, and trusted install Assisted-by: GitHub Copilot (model: Claude Opus 4.8, autonomous) Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8bc6802d-81b8-48f4-8f60-cba3aebc3bb3 --------- Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: 8bc6802d-81b8-48f4-8f60-cba3aebc3bb3
This commit is contained in:
Manfred Riem
@@ -119,6 +119,8 @@ def _run_init(integration: str, *, script_type: str, offline: bool = False) -> N
|
||||
preset=None,
|
||||
integration=integration,
|
||||
integration_options=None,
|
||||
extensions=None,
|
||||
trust_extension_urls=False,
|
||||
)
|
||||
except typer.Exit as exc:
|
||||
if exc.exit_code:
|
||||
|
||||
@@ -30,6 +30,145 @@ def _stdin_is_interactive() -> bool:
|
||||
return sys.stdin.isatty()
|
||||
|
||||
|
||||
def _ext_spec_is_url(ext_spec: str) -> bool:
|
||||
"""Return True when *ext_spec* is an http(s) URL rather than a name/path."""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
try:
|
||||
return urlparse(ext_spec).scheme in ("http", "https")
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
|
||||
def _confirm_extension_url_trust(
|
||||
url_specs: list[str], *, trust_override: bool
|
||||
) -> dict[str, bool]:
|
||||
"""Resolve trust for each URL-based extension before the Live display.
|
||||
|
||||
URL installs pull an arbitrary external extension, so they get the same
|
||||
default-deny confirmation as ``extension add --from``. Returns a mapping of
|
||||
``url_spec -> approved``. With *trust_override* every URL is pre-approved.
|
||||
In a non-interactive session without the override, every URL is denied
|
||||
(the prompt cannot be answered), mirroring the default-deny posture.
|
||||
"""
|
||||
from rich.markup import escape as _escape_markup
|
||||
from rich.panel import Panel
|
||||
|
||||
approvals: dict[str, bool] = {}
|
||||
interactive = _stdin_is_interactive()
|
||||
for spec in url_specs:
|
||||
if trust_override:
|
||||
approvals[spec] = True
|
||||
continue
|
||||
if not interactive:
|
||||
approvals[spec] = False
|
||||
continue
|
||||
console.print()
|
||||
console.print(
|
||||
Panel(
|
||||
"[bold]You are installing an extension from an external URL that is not\n"
|
||||
"listed in any of your configured extension catalogs.[/bold]\n\n"
|
||||
f"URL: {_escape_markup(spec)}\n\n"
|
||||
"Only install extensions from sources you trust.",
|
||||
title="[bold yellow]⚠ Untrusted Source[/bold yellow]",
|
||||
border_style="yellow",
|
||||
padding=(1, 2),
|
||||
)
|
||||
)
|
||||
console.print()
|
||||
approvals[spec] = typer.confirm(
|
||||
f"Install extension from {spec}?", default=False
|
||||
)
|
||||
return approvals
|
||||
|
||||
|
||||
def _install_extension_during_init(project_path: Path, ext_spec: str, speckit_version: str) -> str:
|
||||
"""Install a single extension during ``specify init``.
|
||||
|
||||
Handles bundled extension names, local directory paths, and HTTPS URLs.
|
||||
Returns a short status message on success.
|
||||
Raises ``ValueError`` on failure so the caller can convert it to a
|
||||
tracker error without aborting the entire init.
|
||||
"""
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from .._assets import _locate_bundled_extension
|
||||
from ..extensions import ExtensionCatalog, ExtensionError, ExtensionManager
|
||||
from ..extensions._commands import (
|
||||
_resolve_catalog_extension,
|
||||
install_extension_from_url,
|
||||
)
|
||||
|
||||
manager = ExtensionManager(project_path)
|
||||
|
||||
# --- URL ---
|
||||
parsed = urlparse(ext_spec)
|
||||
if parsed.scheme in ("http", "https"):
|
||||
try:
|
||||
manifest = install_extension_from_url(
|
||||
manager, project_path, ext_spec, speckit_version
|
||||
)
|
||||
except ExtensionError as exc:
|
||||
raise ValueError(str(exc)) from exc
|
||||
return f"{manifest.name} v{manifest.version} installed"
|
||||
|
||||
# --- Local path ---
|
||||
if ext_spec.startswith(("./", "../", "/", "~/", ".\\", "..\\")) or Path(ext_spec).is_absolute():
|
||||
source_path = Path(ext_spec).expanduser().resolve()
|
||||
if not source_path.exists():
|
||||
raise ValueError(f"Directory not found: {source_path}")
|
||||
if not (source_path / "extension.yml").exists():
|
||||
raise ValueError(f"No extension.yml found in {source_path}")
|
||||
manifest = manager.install_from_directory(source_path, speckit_version)
|
||||
return f"{manifest.name} v{manifest.version} installed"
|
||||
|
||||
# --- Bundled extension name or catalog ID ---
|
||||
bundled_path = _locate_bundled_extension(ext_spec)
|
||||
if bundled_path is not None:
|
||||
if manager.registry.is_installed(ext_spec):
|
||||
return "already installed"
|
||||
manifest = manager.install_from_directory(bundled_path, speckit_version)
|
||||
return f"{manifest.name} v{manifest.version} installed"
|
||||
|
||||
# Fall back to catalog
|
||||
catalog = ExtensionCatalog(project_path)
|
||||
ext_info, catalog_error = _resolve_catalog_extension(ext_spec, catalog, "add")
|
||||
if catalog_error:
|
||||
raise ValueError(f"Could not query extension catalog: {catalog_error}")
|
||||
if not ext_info:
|
||||
raise ValueError(f"Extension '{ext_spec}' not found in bundled extensions or catalog")
|
||||
|
||||
resolved_id = ext_info["id"]
|
||||
if resolved_id != ext_spec:
|
||||
bundled_path = _locate_bundled_extension(resolved_id)
|
||||
if bundled_path is not None:
|
||||
if manager.registry.is_installed(resolved_id):
|
||||
return "already installed"
|
||||
manifest = manager.install_from_directory(bundled_path, speckit_version)
|
||||
return f"{manifest.name} v{manifest.version} installed"
|
||||
|
||||
if ext_info.get("bundled") and not ext_info.get("download_url"):
|
||||
from ..extensions import REINSTALL_COMMAND
|
||||
|
||||
raise ValueError(
|
||||
f"Extension '{resolved_id}' is bundled with spec-kit but not found in the installed package. "
|
||||
f"Try reinstalling spec-kit: {REINSTALL_COMMAND}"
|
||||
)
|
||||
|
||||
if not ext_info.get("_install_allowed", True):
|
||||
catalog_name = ext_info.get("_catalog_name", "community")
|
||||
raise ValueError(
|
||||
f"Extension '{ext_spec}' is in the '{catalog_name}' catalog but installation is not allowed from that catalog"
|
||||
)
|
||||
|
||||
zip_path = catalog.download_extension(resolved_id)
|
||||
try:
|
||||
manifest = manager.install_from_zip(zip_path, speckit_version)
|
||||
finally:
|
||||
zip_path.unlink(missing_ok=True)
|
||||
return f"{manifest.name} v{manifest.version} installed"
|
||||
|
||||
|
||||
def ensure_constitution_from_template(
|
||||
project_path: Path, tracker: StepTracker | None = None
|
||||
) -> None:
|
||||
@@ -142,6 +281,16 @@ def register(app: typer.Typer) -> None:
|
||||
"--integration-options",
|
||||
help='Options for the integration (e.g. --integration-options="--commands-dir .myagent/cmds")',
|
||||
),
|
||||
extensions: list[str] | None = typer.Option(
|
||||
None,
|
||||
"--extension",
|
||||
help="Install an extension during initialization (bundled name, local path, or HTTPS URL). Repeatable.",
|
||||
),
|
||||
trust_extension_urls: bool = typer.Option(
|
||||
False,
|
||||
"--trust-extension-urls",
|
||||
help="Pre-authorize installing extensions from external URLs without the interactive trust prompt (required for non-interactive URL installs).",
|
||||
),
|
||||
):
|
||||
"""
|
||||
Initialize a new Specify project.
|
||||
@@ -174,6 +323,10 @@ def register(app: typer.Typer) -> None:
|
||||
specify init --here --integration gemini
|
||||
specify init my-project --integration generic --integration-options="--commands-dir .myagent/commands/" # Bring your own agent; requires --commands-dir
|
||||
specify init my-project --integration claude --preset healthcare-compliance # With preset
|
||||
specify init my-project --integration copilot --extension git # With bundled extension
|
||||
specify init my-project --extension git --extension selftest # Multiple extensions
|
||||
specify init my-project --extension ./my-extensions/custom-ext # Local path extension
|
||||
specify init my-project --extension https://example.com/extensions/my-ext.zip --trust-extension-urls # URL extension (non-interactive)
|
||||
"""
|
||||
# Lazy imports to avoid circular dependency — __init__.py imports this module
|
||||
from .. import (
|
||||
@@ -413,10 +566,31 @@ def register(app: typer.Typer) -> None:
|
||||
("chmod", "Ensure scripts executable"),
|
||||
("constitution", "Constitution setup"),
|
||||
("workflow", "Install bundled workflow"),
|
||||
("final", "Finalize"),
|
||||
]:
|
||||
tracker.add(key, label)
|
||||
|
||||
if extensions:
|
||||
from rich.markup import escape as _escape_markup
|
||||
|
||||
for i, ext_spec in enumerate(extensions):
|
||||
tracker.add(
|
||||
f"extension-{i}", f"Install extension: {_escape_markup(ext_spec)}"
|
||||
)
|
||||
|
||||
tracker.add("final", "Finalize")
|
||||
|
||||
# Resolve trust for URL-based extensions BEFORE entering the Live
|
||||
# display: the confirmation prompt cannot be shown/answered underneath
|
||||
# the Rich Live spinner. URL installs are default-deny unless the user
|
||||
# confirms interactively or passes --trust-extension-urls.
|
||||
extension_url_approvals: dict[str, bool] = {}
|
||||
if extensions:
|
||||
url_specs = [e for e in extensions if _ext_spec_is_url(e)]
|
||||
if url_specs:
|
||||
extension_url_approvals = _confirm_extension_url_trust(
|
||||
url_specs, trust_override=trust_extension_urls
|
||||
)
|
||||
|
||||
# Disable transient mode on Windows: PowerShell 5.1's legacy console
|
||||
# hangs when Rich tries to restore cursor state via VT escape sequences.
|
||||
_transient = sys.platform != "win32"
|
||||
@@ -626,6 +800,46 @@ def register(app: typer.Typer) -> None:
|
||||
continuing="Continuing without the optional preset.",
|
||||
)
|
||||
|
||||
# Install extensions specified via --extension
|
||||
if extensions:
|
||||
from rich.markup import escape as _escape_markup
|
||||
|
||||
from ..extensions._commands import _refresh_events_and_warn
|
||||
|
||||
speckit_ver = get_speckit_version()
|
||||
any_extension_installed = False
|
||||
for i, ext_spec in enumerate(extensions):
|
||||
tracker.start(f"extension-{i}")
|
||||
# Skip URL extensions the user did not confirm as trusted
|
||||
# (default-deny; resolved before the Live display).
|
||||
if _ext_spec_is_url(ext_spec) and not extension_url_approvals.get(
|
||||
ext_spec, False
|
||||
):
|
||||
tracker.error(
|
||||
f"extension-{i}",
|
||||
"skipped: untrusted URL not confirmed "
|
||||
"(use --trust-extension-urls)",
|
||||
)
|
||||
continue
|
||||
try:
|
||||
status_msg = _install_extension_during_init(
|
||||
project_path, ext_spec, speckit_ver
|
||||
)
|
||||
tracker.complete(f"extension-{i}", status_msg)
|
||||
any_extension_installed = True
|
||||
except Exception as ext_err:
|
||||
sanitized_ext = str(ext_err).replace("\n", " ").strip()
|
||||
tracker.error(
|
||||
f"extension-{i}",
|
||||
f"failed: {_escape_markup(sanitized_ext[:120])}",
|
||||
)
|
||||
|
||||
# Refresh native event configuration once after the batch so
|
||||
# that an extension declaring ``events:`` has its hooks
|
||||
# activated, mirroring the ``extension add`` path.
|
||||
if any_extension_installed:
|
||||
_refresh_events_and_warn(project_path)
|
||||
|
||||
# Seed the constitution AFTER preset installation so that a
|
||||
# preset-provided constitution-template (resolved via the
|
||||
# priority stack) wins over the core template.
|
||||
|
||||
@@ -95,6 +95,141 @@ def _refresh_events_and_warn(project_root: Path) -> None:
|
||||
console.print(f" {key}: {_escape_markup(detail)}")
|
||||
|
||||
|
||||
def install_extension_from_url(
|
||||
manager,
|
||||
project_root: Path,
|
||||
url: str,
|
||||
speckit_version: str,
|
||||
*,
|
||||
priority: int = 10,
|
||||
force: bool = False,
|
||||
):
|
||||
"""Download an archive from *url* and install it, reusing the hardened path.
|
||||
|
||||
Shares the same download hardening as ``extension add --from``:
|
||||
HTTPS enforcement, the catalog's authenticated + redirect-guarded
|
||||
``_open_url`` fetch, a bounded (50 MiB) response read, archive-format
|
||||
detection (ZIP or tar.gz/tgz), and a TOCTOU-safe transient download file
|
||||
consumed directly by ``install_from_zip``.
|
||||
|
||||
Returns the installed manifest. Raises ``ExtensionError`` on any failure so
|
||||
callers can present a uniform message without a second downloader.
|
||||
"""
|
||||
import urllib.error
|
||||
|
||||
from . import ExtensionCatalog, ExtensionError
|
||||
|
||||
if not is_https_or_localhost_http(url):
|
||||
raise ExtensionError(
|
||||
"URL must use HTTPS (HTTP is only allowed for localhost)"
|
||||
)
|
||||
|
||||
download_dir = _validate_safe_cache_dir(project_root)
|
||||
archive_filename = f"extension-url-download-{uuid4().hex}.archive"
|
||||
# Only used for diagnostic messages: the real archive is a transient inode
|
||||
# (unlinked on POSIX, O_TEMPORARY on Windows) consumed via ``archive_file``
|
||||
# below, so this path is never opened again.
|
||||
archive_path = download_dir / archive_filename
|
||||
|
||||
try:
|
||||
dl_catalog = ExtensionCatalog(project_root)
|
||||
download_url = url
|
||||
extra_headers = None
|
||||
resolved_url = dl_catalog._resolve_github_release_asset_api_url(download_url)
|
||||
if resolved_url:
|
||||
download_url = resolved_url
|
||||
extra_headers = {"Accept": "application/octet-stream"}
|
||||
|
||||
with dl_catalog._open_url(
|
||||
download_url, timeout=60, extra_headers=extra_headers
|
||||
) as response:
|
||||
archive_data = read_response_limited(
|
||||
response,
|
||||
error_type=ExtensionError,
|
||||
label=f"extension {url}",
|
||||
)
|
||||
final_url = (
|
||||
response.geturl() if hasattr(response, "geturl") else download_url
|
||||
)
|
||||
content_type = (
|
||||
response.getheader("Content-Type")
|
||||
if hasattr(response, "getheader")
|
||||
else None
|
||||
)
|
||||
except urllib.error.URLError as exc:
|
||||
raise ExtensionError(f"Failed to download from {url}: {exc}") from exc
|
||||
|
||||
download_fd = -1
|
||||
download_file = None
|
||||
try:
|
||||
try:
|
||||
download_fd = _safe_open_download_zip(
|
||||
project_root, download_dir, archive_filename
|
||||
)
|
||||
except OSError as exc:
|
||||
raise ExtensionError(
|
||||
f"Could not safely create download file: {exc}"
|
||||
) from exc
|
||||
|
||||
try:
|
||||
download_file = os.fdopen(download_fd, "w+b")
|
||||
download_fd = -1
|
||||
download_file.write(archive_data)
|
||||
download_file.flush()
|
||||
download_file.seek(0)
|
||||
except OSError as exc:
|
||||
raise ExtensionError(
|
||||
f"Could not safely write download file: {exc}"
|
||||
) from exc
|
||||
|
||||
format_source = (
|
||||
final_url
|
||||
if archive_format_from_name(final_url) is not None
|
||||
else url
|
||||
)
|
||||
try:
|
||||
detect_archive_format(
|
||||
archive_path,
|
||||
archive_file=download_file,
|
||||
source_name=format_source,
|
||||
content_type=content_type,
|
||||
error_type=ExtensionError,
|
||||
)
|
||||
except ExtensionError as exc:
|
||||
raise ExtensionError(
|
||||
f"{url} did not return a ZIP archive or tar.gz/tgz archive "
|
||||
f"(got {len(archive_data)} bytes). This usually means the request "
|
||||
"was not authenticated and a login/HTML page was returned. "
|
||||
"Verify the URL and configured credentials."
|
||||
) from exc
|
||||
|
||||
# Consume the transient inode reserved above rather than reopening the
|
||||
# cache pathname during extraction.
|
||||
try:
|
||||
return manager.install_from_zip(
|
||||
archive_path,
|
||||
speckit_version,
|
||||
priority=priority,
|
||||
force=force,
|
||||
archive_file=download_file,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise ExtensionError(
|
||||
f"Could not install extension from downloaded archive: {exc}"
|
||||
) from exc
|
||||
finally:
|
||||
if download_file is not None:
|
||||
try:
|
||||
download_file.close()
|
||||
except OSError:
|
||||
pass
|
||||
elif download_fd >= 0:
|
||||
try:
|
||||
os.close(download_fd)
|
||||
except OSError:
|
||||
pass
|
||||
|
||||
|
||||
def _load_catalog_command_config(project_root: Path, config_path: Path) -> dict:
|
||||
"""Load extension catalog CLI config with user-facing shape errors."""
|
||||
try:
|
||||
@@ -810,134 +945,20 @@ def extension_add(
|
||||
)
|
||||
|
||||
elif from_url:
|
||||
# Install from an archive URL.
|
||||
import urllib.error
|
||||
|
||||
# Install from URL archive via the shared hardened downloader
|
||||
# (HTTPS enforcement, authenticated redirect-guarded fetch,
|
||||
# bounded read, archive-format detection, TOCTOU-safe transient
|
||||
# archive). Same path used by ``specify init --extension <url>``.
|
||||
console.print(f"Downloading from {safe_url}...")
|
||||
manifest = install_extension_from_url(
|
||||
manager,
|
||||
project_root,
|
||||
from_url,
|
||||
speckit_version,
|
||||
priority=priority,
|
||||
force=force,
|
||||
)
|
||||
|
||||
download_dir = _validate_safe_cache_dir(project_root)
|
||||
archive_filename = f"extension-url-download-{uuid4().hex}.archive"
|
||||
# Only used for diagnostic messages: the real archive is a
|
||||
# transient inode (unlinked on POSIX, O_TEMPORARY on Windows)
|
||||
# consumed via ``archive_file`` below, so this path is never
|
||||
# opened again.
|
||||
archive_path = download_dir / archive_filename
|
||||
|
||||
try:
|
||||
# Use the catalog's authenticated fetch so configured
|
||||
# credentials (incl. GitHub Enterprise Server) are applied
|
||||
# and GHES release-asset URLs resolve via /api/v3 — keeping
|
||||
# --from consistent with catalog-based installs.
|
||||
dl_catalog = ExtensionCatalog(project_root)
|
||||
download_url = from_url
|
||||
extra_headers = None
|
||||
resolved_url = dl_catalog._resolve_github_release_asset_api_url(download_url)
|
||||
if resolved_url:
|
||||
download_url = resolved_url
|
||||
extra_headers = {"Accept": "application/octet-stream"}
|
||||
|
||||
with dl_catalog._open_url(
|
||||
download_url, timeout=60, extra_headers=extra_headers
|
||||
) as response:
|
||||
archive_data = read_response_limited(
|
||||
response,
|
||||
error_type=ExtensionError,
|
||||
label=f"extension {from_url}",
|
||||
)
|
||||
final_url = (
|
||||
response.geturl()
|
||||
if hasattr(response, "geturl")
|
||||
else download_url
|
||||
)
|
||||
content_type = (
|
||||
response.getheader("Content-Type")
|
||||
if hasattr(response, "getheader")
|
||||
else None
|
||||
)
|
||||
|
||||
download_fd = -1
|
||||
download_file = None
|
||||
try:
|
||||
try:
|
||||
download_fd = _safe_open_download_zip(
|
||||
project_root, download_dir, archive_filename
|
||||
)
|
||||
except OSError as exc:
|
||||
console.print(
|
||||
"[red]Error:[/red] Could not safely create download file: "
|
||||
f"{_escape_markup(str(exc))}"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
try:
|
||||
download_file = os.fdopen(download_fd, "w+b")
|
||||
download_fd = -1
|
||||
download_file.write(archive_data)
|
||||
download_file.flush()
|
||||
download_file.seek(0)
|
||||
except OSError as exc:
|
||||
console.print(
|
||||
"[red]Error:[/red] Could not safely write download file: "
|
||||
f"{_escape_markup(str(exc))}"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
format_source = (
|
||||
final_url
|
||||
if archive_format_from_name(final_url) is not None
|
||||
else from_url
|
||||
)
|
||||
try:
|
||||
detect_archive_format(
|
||||
archive_path,
|
||||
archive_file=download_file,
|
||||
source_name=format_source,
|
||||
content_type=content_type,
|
||||
error_type=ExtensionError,
|
||||
)
|
||||
except ExtensionError:
|
||||
console.print(
|
||||
f"[red]Error:[/red] {safe_url} did not return a ZIP archive "
|
||||
"or tar.gz/tgz archive "
|
||||
f"(got {len(archive_data)} bytes). This usually means "
|
||||
"the request was not authenticated and a login/HTML page was "
|
||||
"returned. Verify the URL and configured credentials."
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
# Consume the transient inode reserved above rather
|
||||
# than reopening the cache pathname during extraction.
|
||||
try:
|
||||
manifest = manager.install_from_zip(
|
||||
archive_path,
|
||||
speckit_version,
|
||||
priority=priority,
|
||||
force=force,
|
||||
archive_file=download_file,
|
||||
)
|
||||
except OSError as exc:
|
||||
console.print(
|
||||
"[red]Error:[/red] Could not install extension from downloaded archive: "
|
||||
f"{_escape_markup(str(exc))}"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
finally:
|
||||
if download_file is not None:
|
||||
try:
|
||||
download_file.close()
|
||||
except OSError:
|
||||
pass
|
||||
elif download_fd >= 0:
|
||||
try:
|
||||
os.close(download_fd)
|
||||
except OSError:
|
||||
pass
|
||||
except urllib.error.URLError as e:
|
||||
console.print(
|
||||
f"[red]Error:[/red] Failed to download from {safe_url}: "
|
||||
f"{_escape_markup(str(e))}"
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
else:
|
||||
# Try bundled extensions first (shipped with spec-kit)
|
||||
bundled_path = _locate_bundled_extension(extension)
|
||||
|
||||
@@ -2372,3 +2372,279 @@ def test_refresh_shared_templates_preserves_recovered_user_file(tmp_path):
|
||||
|
||||
# Recovered user content must survive (fail-before: replaced by bundled body).
|
||||
assert user_file.read_text(encoding="utf-8") == "# USER CUSTOM CONTENT\n"
|
||||
|
||||
|
||||
class TestExtensionFlag:
|
||||
"""Tests for the --extension flag on specify init."""
|
||||
|
||||
def _run_init(self, tmp_path, args, project_name="ext-test"):
|
||||
from unittest.mock import patch
|
||||
from typer.testing import CliRunner
|
||||
from specify_cli import app
|
||||
|
||||
project = tmp_path / project_name
|
||||
project.mkdir(exist_ok=True)
|
||||
old_cwd = os.getcwd()
|
||||
try:
|
||||
os.chdir(project)
|
||||
runner = CliRunner()
|
||||
# Patch get_speckit_version to return a stable (non-dev) version so that
|
||||
# the extension compatibility check (SpecifierSet(">=0.2.0")) passes.
|
||||
with patch(
|
||||
"specify_cli.commands.init.get_speckit_version",
|
||||
return_value="0.8.2",
|
||||
):
|
||||
result = runner.invoke(app, [
|
||||
"init", "--here",
|
||||
"--integration", "copilot",
|
||||
"--script", "sh",
|
||||
"--ignore-agent-tools",
|
||||
] + args, catch_exceptions=False)
|
||||
finally:
|
||||
os.chdir(old_cwd)
|
||||
return project, result
|
||||
|
||||
def test_bundled_extension_installed(self, tmp_path):
|
||||
"""--extension git installs the bundled git extension."""
|
||||
project, result = self._run_init(tmp_path, ["--extension", "git"], project_name="ext-bundled")
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
|
||||
ext_dir = project / ".specify" / "extensions" / "git"
|
||||
assert ext_dir.exists(), "git extension directory not found"
|
||||
assert (ext_dir / "extension.yml").exists(), "extension.yml not found"
|
||||
|
||||
# Tracker should show extension step as done
|
||||
normalized = _normalize_cli_output(result.output)
|
||||
assert "Install extension: git" in normalized
|
||||
|
||||
def test_multiple_extensions_installed(self, tmp_path):
|
||||
"""--extension can be specified multiple times."""
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "git", "--extension", "selftest"],
|
||||
project_name="ext-multi",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
|
||||
ext_dir_git = project / ".specify" / "extensions" / "git"
|
||||
ext_dir_selftest = project / ".specify" / "extensions" / "selftest"
|
||||
assert ext_dir_git.exists(), "git extension not installed"
|
||||
assert ext_dir_selftest.exists(), "selftest extension not installed"
|
||||
|
||||
def test_local_path_extension_installed(self, tmp_path):
|
||||
"""--extension /abs/path installs from a local absolute directory path."""
|
||||
from specify_cli import _locate_bundled_extension
|
||||
|
||||
# Use the bundled git extension directory as our "local" extension source
|
||||
bundled_git = _locate_bundled_extension("git")
|
||||
assert bundled_git is not None, "bundled git extension not found; cannot run test"
|
||||
|
||||
# Pass the absolute path directly (starts with "/")
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", str(bundled_git)],
|
||||
project_name="ext-local",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
|
||||
ext_dir = project / ".specify" / "extensions" / "git"
|
||||
assert ext_dir.exists(), "extension from local path not installed"
|
||||
|
||||
def test_unknown_extension_shows_error_in_tracker(self, tmp_path):
|
||||
"""An unknown extension name records a tracker error but does not abort init."""
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "nonexistent-xyz-ext"],
|
||||
project_name="ext-unknown",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, "init should not abort on unknown extension"
|
||||
normalized = _normalize_cli_output(result.output)
|
||||
assert "failed" in normalized.lower(), "expected 'failed' for unknown extension"
|
||||
|
||||
def test_extension_flag_works_with_preset(self, tmp_path):
|
||||
"""--extension and --preset can be combined."""
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "git", "--preset", "lean"],
|
||||
project_name="ext-preset",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
|
||||
ext_dir = project / ".specify" / "extensions" / "git"
|
||||
assert ext_dir.exists(), "git extension not installed alongside preset"
|
||||
|
||||
@staticmethod
|
||||
def _zip_bytes_from_dir(source_dir):
|
||||
"""Build in-memory ZIP bytes from an extension directory (yml at root)."""
|
||||
import io
|
||||
import zipfile
|
||||
|
||||
buf = io.BytesIO()
|
||||
with zipfile.ZipFile(buf, "w", zipfile.ZIP_DEFLATED) as zf:
|
||||
for path in sorted(source_dir.rglob("*")):
|
||||
if path.is_file():
|
||||
zf.write(path, arcname=str(path.relative_to(source_dir)))
|
||||
return buf.getvalue()
|
||||
|
||||
def test_url_extension_rejects_non_https(self, tmp_path):
|
||||
"""A non-HTTPS URL is rejected before any download; init is not aborted."""
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "http://example.com/ext.zip", "--trust-extension-urls"],
|
||||
project_name="ext-http",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, "init should not abort on a rejected URL"
|
||||
normalized = _normalize_cli_output(result.output)
|
||||
assert "failed" in normalized.lower()
|
||||
# No extension directory should have been created for the bad URL.
|
||||
assert not (project / ".specify" / "extensions" / "ext").exists()
|
||||
|
||||
def test_url_extension_skipped_without_trust(self, tmp_path):
|
||||
"""Non-interactive URL install without --trust-extension-urls is denied."""
|
||||
from unittest.mock import patch
|
||||
|
||||
with patch(
|
||||
"specify_cli.commands.init._stdin_is_interactive", return_value=False
|
||||
), patch("specify_cli.authentication.http.open_url") as mock_open:
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "https://example.com/git.zip"],
|
||||
project_name="ext-url-denied",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
# Default-deny: no download attempted, nothing installed.
|
||||
mock_open.assert_not_called()
|
||||
normalized = _normalize_cli_output(result.output)
|
||||
assert "untrusted url" in normalized.lower()
|
||||
assert not (project / ".specify" / "extensions" / "git").exists()
|
||||
|
||||
def test_url_extension_interactive_confirm_installs(self, tmp_path):
|
||||
"""An interactive 'yes' to the trust prompt allows the URL install."""
|
||||
import io
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from specify_cli import _locate_bundled_extension
|
||||
|
||||
bundled_git = _locate_bundled_extension("git")
|
||||
assert bundled_git is not None, "bundled git extension not found"
|
||||
zip_bytes = self._zip_bytes_from_dir(bundled_git)
|
||||
|
||||
class FakeResponse(io.BytesIO):
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
def _cache_dir_stand_in(project_root):
|
||||
d = project_root / ".specify" / "extensions" / ".cache" / "downloads"
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
return d
|
||||
|
||||
def _open_download_zip(project_root, download_dir, zip_filename):
|
||||
target = download_dir / zip_filename
|
||||
o_temporary = getattr(os, "O_TEMPORARY", 0)
|
||||
if o_temporary:
|
||||
return os.open(
|
||||
target, os.O_RDWR | os.O_CREAT | os.O_EXCL | o_temporary, 0o600
|
||||
)
|
||||
fd = os.open(target, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
|
||||
try:
|
||||
os.unlink(target)
|
||||
except OSError:
|
||||
os.close(fd)
|
||||
raise
|
||||
return fd
|
||||
|
||||
with patch(
|
||||
"specify_cli.commands.init._stdin_is_interactive", return_value=True
|
||||
), patch("typer.confirm", return_value=True), patch(
|
||||
"specify_cli.authentication.http.open_url",
|
||||
return_value=FakeResponse(zip_bytes),
|
||||
), patch(
|
||||
"specify_cli.extensions._commands._validate_safe_cache_dir",
|
||||
side_effect=_cache_dir_stand_in,
|
||||
), patch(
|
||||
"specify_cli.extensions._commands._safe_open_download_zip",
|
||||
side_effect=_open_download_zip,
|
||||
):
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "https://example.com/git.zip"],
|
||||
project_name="ext-url-confirm",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
assert (project / ".specify" / "extensions" / "git").exists()
|
||||
|
||||
def test_url_extension_installs_zip(self, tmp_path):
|
||||
"""A successful HTTPS ZIP download installs via the shared hardened path."""
|
||||
import io
|
||||
|
||||
from unittest.mock import patch
|
||||
|
||||
from specify_cli import _locate_bundled_extension
|
||||
|
||||
bundled_git = _locate_bundled_extension("git")
|
||||
assert bundled_git is not None, "bundled git extension not found"
|
||||
zip_bytes = self._zip_bytes_from_dir(bundled_git)
|
||||
|
||||
class FakeResponse(io.BytesIO):
|
||||
def __enter__(self):
|
||||
return self
|
||||
|
||||
def __exit__(self, exc_type, exc, tb):
|
||||
return False
|
||||
|
||||
def _cache_dir_stand_in(project_root):
|
||||
d = project_root / ".specify" / "extensions" / ".cache" / "downloads"
|
||||
d.mkdir(parents=True, exist_ok=True)
|
||||
return d
|
||||
|
||||
def _open_download_zip(project_root, download_dir, zip_filename):
|
||||
target = download_dir / zip_filename
|
||||
o_temporary = getattr(os, "O_TEMPORARY", 0)
|
||||
if o_temporary:
|
||||
return os.open(
|
||||
target, os.O_RDWR | os.O_CREAT | os.O_EXCL | o_temporary, 0o600
|
||||
)
|
||||
fd = os.open(target, os.O_RDWR | os.O_CREAT | os.O_EXCL, 0o600)
|
||||
try:
|
||||
os.unlink(target)
|
||||
except OSError:
|
||||
os.close(fd)
|
||||
raise
|
||||
return fd
|
||||
|
||||
with patch(
|
||||
"specify_cli.authentication.http.open_url",
|
||||
return_value=FakeResponse(zip_bytes),
|
||||
), patch(
|
||||
"specify_cli.extensions._commands._validate_safe_cache_dir",
|
||||
side_effect=_cache_dir_stand_in,
|
||||
), patch(
|
||||
"specify_cli.extensions._commands._safe_open_download_zip",
|
||||
side_effect=_open_download_zip,
|
||||
):
|
||||
project, result = self._run_init(
|
||||
tmp_path,
|
||||
["--extension", "https://example.com/git.zip", "--trust-extension-urls"],
|
||||
project_name="ext-url",
|
||||
)
|
||||
|
||||
assert result.exit_code == 0, f"init failed:\n{result.output}"
|
||||
ext_dir = project / ".specify" / "extensions" / "git"
|
||||
assert ext_dir.exists(), "extension from URL not installed"
|
||||
assert (ext_dir / "extension.yml").exists()
|
||||
# Transient download archive must not linger in the cache.
|
||||
cache_dir = project / ".specify" / "extensions" / ".cache" / "downloads"
|
||||
leftover = list(cache_dir.glob("*.zip")) if cache_dir.exists() else []
|
||||
assert not leftover, f"download cache not cleaned: {leftover}"
|
||||
|
||||
Reference in New Issue
Block a user