mirror of
https://github.com/github/spec-kit.git
synced 2026-08-03 06:26:30 +08:00
Compare commits
21 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
d3a3888c3a | ||
|
|
7cd97e47f0 | ||
|
|
93dbf6d575 | ||
|
|
4fc0a5b06e | ||
|
|
0a7f288ae4 | ||
|
|
5e384bb9f5 | ||
|
|
e4cfa4c19c | ||
|
|
6e8623bbd7 | ||
|
|
370551ea89 | ||
|
|
38eb2fcc4b | ||
|
|
37041087dd | ||
|
|
3a7a8758f7 | ||
|
|
c0f4cee25a | ||
|
|
93fc533d79 | ||
|
|
3356161d88 | ||
|
|
9fb467f8de | ||
|
|
8c816fac40 | ||
|
|
fb7dc0c4d6 | ||
|
|
a5560fcf13 | ||
|
|
5601830ba3 | ||
|
|
0f6ea64a03 |
@@ -65,7 +65,8 @@
|
||||
},
|
||||
"chat.tools.terminal.autoApprove": {
|
||||
".specify/scripts/bash/": true,
|
||||
".specify/scripts/powershell/": true
|
||||
".specify/scripts/powershell/": true,
|
||||
".specify/scripts/python/": true
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
43
AGENTS.md
43
AGENTS.md
@@ -187,7 +187,7 @@ context_markers:
|
||||
end: "<!-- SPECKIT END -->"
|
||||
```
|
||||
|
||||
- The Specify CLI does **not** write this config. When `context_file` is empty, the extension's bundled scripts self-seed it by looking up the active integration's key in the extension's own `agent-context-defaults.json` map (`extensions/agent-context/scripts/bash/update-agent-context.sh` and `.ps1`). The CLI registry is never consulted — all agent→context-file knowledge lives inside the extension.
|
||||
- The Specify CLI does **not** write this config. When `context_file` is empty, the extension's bundled scripts self-seed it by looking up the active integration's key in the extension's own `agent-context-defaults.json` map (`extensions/agent-context/scripts/bash/update-agent-context.sh`, `.ps1`, and `extensions/agent-context/scripts/python/update_agent_context.py`). The CLI registry is never consulted — all agent→context-file knowledge lives inside the extension.
|
||||
- `context_markers.{start,end}` are read solely by the extension's scripts; they default to the Spec Kit markers shown above and can be customized by editing `agent-context-config.yml` directly.
|
||||
|
||||
Existing projects created by older Spec Kit versions keep working: any previously written managed section or extension config is left intact and is only ever updated by the extension when run.
|
||||
@@ -268,6 +268,25 @@ echo "✅ Done"
|
||||
|
||||
## Command File Formats
|
||||
|
||||
### Script References (`scripts:` frontmatter)
|
||||
|
||||
Core command templates (`templates/commands/*.md`) that invoke a helper script declare it in a `scripts:` frontmatter block with one line per supported script type. The `{SCRIPT}` placeholder in the command body is replaced at install time with the entry matching the project's selected script type (`--script sh|ps|py`):
|
||||
|
||||
```yaml
|
||||
scripts:
|
||||
sh: scripts/bash/setup-plan.sh --json
|
||||
ps: scripts/powershell/setup-plan.ps1 -Json
|
||||
py: scripts/python/setup_plan.py --json
|
||||
```
|
||||
|
||||
| Key | Script type | Location |
|
||||
| ---- | ---------------------- | -------------------------- |
|
||||
| `sh` | POSIX shell (bash/zsh) | `scripts/bash/*.sh` |
|
||||
| `ps` | PowerShell | `scripts/powershell/*.ps1` |
|
||||
| `py` | Python | `scripts/python/*.py` |
|
||||
|
||||
All three entries must be present and behaviorally equivalent — agents parse the same stdout contract (`FEATURE_DIR:…`, `AVAILABLE_DOCS:…`, `--json` shapes) regardless of which one runs. (The bundled `agent-context` and `git` extension command templates also invoke helpers but do not yet use `scripts:` frontmatter — see [Script Types and Migration](#script-types-and-migration).)
|
||||
|
||||
### Markdown Format
|
||||
|
||||
**Standard format:**
|
||||
@@ -328,9 +347,29 @@ Different agents use different argument placeholders. The placeholder used in co
|
||||
- **TOML-based**: `{{args}}` (e.g., Gemini)
|
||||
- **YAML-based**: `{{args}}` (e.g., Goose)
|
||||
- **Custom**: some agents override the default (e.g., Forge uses `{{parameters}}`)
|
||||
- **Script placeholders**: `{SCRIPT}` (replaced with actual script path)
|
||||
- **Script placeholders**: `{SCRIPT}` (replaced with the resolved command from the template's `scripts:` frontmatter, per the project's `--script sh|ps|py` selection)
|
||||
- **Agent placeholders**: `__AGENT__` (replaced with agent name)
|
||||
|
||||
## Script Types and Migration
|
||||
|
||||
Spec Kit ships every core workflow script in three interchangeable variants — POSIX shell (`sh`), PowerShell (`ps`), and Python (`py`) — selected per project with `specify init --script sh|ps|py`. Each core command template that invokes a helper script carries all three in its `scripts:` frontmatter (templates that don't call a script, e.g. `constitution`/`specify`, have no `scripts:` block); see [Script References](#script-references-scripts-frontmatter).
|
||||
|
||||
### Why Python is recommended
|
||||
|
||||
- **No extra runtime.** The `specify` CLI is already Python, so the interpreter is guaranteed present — `py` adds no new dependency.
|
||||
- **Path toward a single source of truth.** The shell variants require paired `.sh` + `.ps1` maintenance and diverge on JSON handling (`jq` vs manual parsing). The Python variant avoids `jq` and is intended to eventually replace that dual-maintenance — but that consolidation has not happened yet: all three variants are still maintained in parallel (see the parity rule below).
|
||||
- **Parity-tested.** The Python ports are covered by tests — output-parity tests against the shell scripts where the contract is stdout-based, and direct unit tests elsewhere — so the stdout contract agents rely on stays stable.
|
||||
|
||||
### Defaults and availability
|
||||
|
||||
- `py` is available today for the core command templates (via their `scripts:` frontmatter). The bundled extensions (`agent-context`, `git`) ship Python script variants on disk, but their command templates still hard-code the Bash/PowerShell invocations, so `--script py` does not yet route those extension commands to Python — wiring `py` into the extension command templates is tracked separately.
|
||||
- Selection is per project: interactive `specify init` prompts for the script type, while non-interactive runs default to a shell variant by OS (`sh` on Linux/macOS, `ps` on Windows). `py` is chosen at the prompt or via `--script py`.
|
||||
- `sh` and `ps` remain fully supported. Nothing is removed, and `py` is not yet the default.
|
||||
|
||||
### Parity rule for contributors
|
||||
|
||||
All three script types are first-class: any change to a workflow script must update `sh`, `ps`, and `py` together and keep their tests (parity and unit) green. Making `py` the default and eventually retiring `sh`/`ps` is future work gated on adoption, tracked under the script-unification epic ([#3277](https://github.com/github/spec-kit/issues/3277)) — not something to act on from this doc.
|
||||
|
||||
## Special Processing Requirements
|
||||
|
||||
Some agents require custom processing beyond the standard template transformations:
|
||||
|
||||
25
CHANGELOG.md
25
CHANGELOG.md
@@ -2,6 +2,31 @@
|
||||
|
||||
<!-- insert new changelog below this comment -->
|
||||
|
||||
## [0.14.0] - 2026-07-23
|
||||
|
||||
### Changed
|
||||
|
||||
- docs: add spec-kit-copilot to community friends (#3675)
|
||||
- fix(integrations): recompute invoke_separator from retained parsed_options (#3664)
|
||||
- fix(workflows): preserve intra-overlay order for multiple insert_after edits (#3662)
|
||||
- fix(bundler): reject falsy non-mapping requires/provides in manifest from_dict (#3661)
|
||||
- fix(bundler): dump_yaml writes literal UTF-8 (allow_unicode=True) (#3660)
|
||||
- fix(integrations): declare kiro-cli multi-install safe (#3477)
|
||||
- fix(git-extension): trim trailing whitespace before stripping commit-message quotes (#3673)
|
||||
- fix(bundler): order bundle members by canonical POSIX arcname (reproducible builds) (#3658)
|
||||
- fix(integrations): Cline overrides post_process_command_content (correct hook name) (#3657)
|
||||
- docs(workflows): gate step docstring lists the 'retry' on_reject behaviour (#3656)
|
||||
- fix: harden bounded reads and redirect validation (#3671)
|
||||
- fix(packaging): bundle scripts/python into the wheel core_pack (#3665) (#3670)
|
||||
- fix: bundle scripts/python in wheel so --script py works (#3665) (#3668)
|
||||
- docs(workflows): init step docstring lists the 'py' script type (#3655)
|
||||
- fix(integrations): declare LingmaIntegration multi_install_safe (#3654)
|
||||
- fix: guard constitution command against feature execution (#3646)
|
||||
- Fix duplicate step numbering in specify command (#3647)
|
||||
- docs(scripts): document the 'py' script type and sh/ps migration plan (#3284) (#3653)
|
||||
- harden: bound HTTP reads and enforce strict redirects (#3140)
|
||||
- chore: release 0.13.4, begin 0.13.5.dev0 development (#3649)
|
||||
|
||||
## [0.13.4] - 2026-07-22
|
||||
|
||||
### Changed
|
||||
|
||||
@@ -1,7 +1,7 @@
|
||||
# Community Friends
|
||||
|
||||
> [!NOTE]
|
||||
> Community projects listed here are independently created and maintained by their respective authors. They are **not reviewed, nor endorsed, nor supported by GitHub**. Review their source code before installation and use at your own discretion.
|
||||
> Community projects listed here are independently created and maintained by their respective authors. Unless explicitly marked as a **first-party GitHub project**, they are **not reviewed, nor endorsed, nor supported by GitHub**. Review their source code before installation and use at your own discretion.
|
||||
|
||||
Community projects that extend, visualize, or build on Spec Kit:
|
||||
|
||||
@@ -16,3 +16,5 @@ Community projects that extend, visualize, or build on Spec Kit:
|
||||
- **[cc-spec-kit](https://github.com/speckit-community/cc-spec-kit)** — Community-maintained plugin for Claude Code and GitHub Copilot CLI that installs Spec Kit skills via the plugin marketplace.
|
||||
|
||||
- **[spectatui](https://github.com/tinesoft/spectatui)** — A terminal UI (TUI) dashboard for Spec Kit that lets you track features, manage specifications, integrations, presets, workflows, and extensions, and monitor AI agent workflows. Attach to existing AI sessions or launch new ones from your terminal. Keyboard and mouse support. Light/dark theme support. Customizable and performance-oriented. Requires the `specify` CLI in your PATH.
|
||||
|
||||
- **[spec-kit-copilot](https://github.com/github/spec-kit-copilot)** — _First-party GitHub project._ A GitHub Copilot **skills plugin** that exposes the Spec Kit `specify` CLI to the Copilot agent in both the Copilot CLI and the GitHub Copilot app. It provides a focused skill per `specify` command group — setup, init, check, extensions, presets, bundles, workflows, workflow steps, and self-upgrade — so you can navigate and drive the entire Spec Kit ecosystem through natural language, letting Copilot decide when and how to run the right `specify` commands on your behalf.
|
||||
|
||||
@@ -2,7 +2,7 @@
|
||||
|
||||
This guide shows how to iterate on the `specify` CLI locally without publishing a release or committing to `main` first.
|
||||
|
||||
> Scripts now have both Bash (`.sh`) and PowerShell (`.ps1`) variants. The CLI auto-selects based on OS unless you pass `--script sh|ps`.
|
||||
> Scripts are available as Bash (`.sh`), PowerShell (`.ps1`), and Python (`.py`) variants. Interactive `specify init` prompts you to choose one; non-interactive runs default to a shell variant for your OS. Pass `--script sh|ps|py` to select explicitly.
|
||||
|
||||
## 1. Clone and Switch Branches
|
||||
|
||||
@@ -189,7 +189,7 @@ rm -rf .venv dist build *.egg-info
|
||||
| `ModuleNotFoundError: typer` | Run `uv pip install -e .` |
|
||||
| Scripts not executable (Linux) | Re-run init or `chmod +x scripts/*.sh` |
|
||||
| Git commands unavailable | Install the git extension with `specify extension add git` |
|
||||
| Wrong script type downloaded | Pass `--script sh` or `--script ps` explicitly |
|
||||
| Wrong script type downloaded | Pass `--script sh`, `--script ps`, or `--script py` explicitly |
|
||||
| TLS errors on corporate network | Configure your environment's certificate store or proxy. The `--skip-tls` flag is deprecated and has no effect. |
|
||||
|
||||
## 14. Next Steps
|
||||
|
||||
@@ -3,7 +3,7 @@
|
||||
This guide will help you get started with Spec-Driven Development using Spec Kit. Throughout, we illustrate each step with a running example: **Taskify**, a small team productivity platform.
|
||||
|
||||
> [!NOTE]
|
||||
> Automation scripts are provided as both Bash (`.sh`) and PowerShell (`.ps1`) variants. The `specify` CLI auto-selects based on your OS unless you pass `--script sh|ps`.
|
||||
> Automation scripts are provided as Bash (`.sh`), PowerShell (`.ps1`), and Python (`.py`) variants. Interactive `specify init` prompts you to choose one; non-interactive runs default to a shell variant for your OS. Pass `--script sh|ps|py` to select explicitly.
|
||||
|
||||
> [!NOTE]
|
||||
> Commands are shown here in `/speckit.*` form, but the exact invocation depends on your agent. Some skills-based agents use `$speckit-*` (e.g. Codex, ZCode) or `/skill:speckit-*` (e.g. Kimi). Use whichever form your agent exposes — the steps are otherwise identical.
|
||||
|
||||
@@ -86,7 +86,7 @@ specify integration install <key>
|
||||
|
||||
| Option | Description |
|
||||
| ------------------------ | ------------------------------------------------------------------------ |
|
||||
| `--script sh\|ps` | Script type: `sh` (bash/zsh) or `ps` (PowerShell) |
|
||||
| `--script sh\|ps\|py` | Script type: `sh` (bash/zsh), `ps` (PowerShell), or `py` (Python) |
|
||||
| `--force` | Opt in to installing alongside integrations that are not declared multi-install safe |
|
||||
| `--integration-options` | Integration-specific options (e.g. `--integration-options="--commands-dir .myagent/cmds"`) |
|
||||
|
||||
@@ -122,7 +122,7 @@ specify integration switch <key>
|
||||
|
||||
| Option | Description |
|
||||
| ------------------------ | ------------------------------------------------------------------------ |
|
||||
| `--script sh\|ps` | Script type: `sh` (bash/zsh) or `ps` (PowerShell) |
|
||||
| `--script sh\|ps\|py` | Script type: `sh` (bash/zsh), `ps` (PowerShell), or `py` (Python) |
|
||||
| `--force` | Force removal of modified files during uninstall; when the target is already installed, overwrite managed shared templates while changing the default |
|
||||
| `--refresh-shared-infra` | Also overwrite shared infrastructure files even if you customized them (otherwise customizations are preserved) |
|
||||
| `--integration-options` | Options for the target integration when it is not already installed |
|
||||
@@ -150,7 +150,7 @@ specify integration upgrade [<key>]
|
||||
| Option | Description |
|
||||
| ------------------------ | ------------------------------------------------------------------------ |
|
||||
| `--force` | Overwrite files even if they have been modified |
|
||||
| `--script sh\|ps` | Script type: `sh` (bash/zsh) or `ps` (PowerShell) |
|
||||
| `--script sh\|ps\|py` | Script type: `sh` (bash/zsh), `ps` (PowerShell), or `py` (Python) |
|
||||
| `--integration-options` | Options for the integration |
|
||||
|
||||
Reinstalls an installed integration with updated templates and commands (e.g., after upgrading Spec Kit). Defaults to the default integration; if a key is provided, it must be one of the installed integrations. Detects locally modified files and blocks the upgrade unless `--force` is used. Stale files from the previous install that are no longer needed are removed automatically. Shared templates stay aligned with the default integration even when upgrading a non-default integration.
|
||||
@@ -257,31 +257,30 @@ Spec Kit tracks one default integration in `.specify/integration.json` with `def
|
||||
|
||||
An integration is multi-install safe when it uses a static, unique agent root and command directory, stable command invocation settings, and a separate install manifest whose managed files do not overlap another safe integration. Registry tests enforce those path and manifest invariants. Shared Spec Kit templates remain aligned to the single default integration.
|
||||
|
||||
The Isolation column below lists paths Spec Kit manages for that integration (skills/commands roots and any integration-owned rule files). It is not a full inventory of every file an agent may read.
|
||||
|
||||
**Agent-context defaults are separate.** The optional agent-context extension maps each integration to a default context file in `extensions/agent-context/agent-context-defaults.json`. Those defaults are independent of multi-install safety: several agents may share a root file such as `AGENTS.md` when the extension is enabled. Multi-install safety does not require a unique context file per safe integration.
|
||||
The Command directory column below lists the directory each integration installs its commands or skills into. Context-file targeting is a separate concern from integration multi-install safety: `multi_install_safe` is an integration declaration about command/skill paths, whereas the optional agent-context extension manages a per-agent context file (for example `AGENTS.md` or `CLAUDE.md`) and can even synchronize several anchors at once via its `context_files` setting. Multiple agents mapping to the same context file is expected there and does not affect whether an integration is multi-install safe; see the agent-context extension for details.
|
||||
|
||||
The currently declared multi-install safe integrations are:
|
||||
|
||||
| Key | Isolation |
|
||||
| --- | --------- |
|
||||
| `auggie` | `.augment/commands`, `.augment/rules/specify-rules.md` |
|
||||
| `claude` | `.claude/skills`, `CLAUDE.md` |
|
||||
| `cline` | `.clinerules/workflows`, `.clinerules/specify-rules.md` |
|
||||
| `codebuddy` | `.codebuddy/commands`, `CODEBUDDY.md` |
|
||||
| `codex` | `.agents/skills`, `AGENTS.md` |
|
||||
| `cursor-agent` | `.cursor/skills`, `.cursor/rules/specify-rules.mdc` |
|
||||
| `firebender` | `.firebender/commands`, `.firebender/rules/specify-rules.mdc` |
|
||||
| `gemini` | `.gemini/commands`, `GEMINI.md` |
|
||||
| Key | Command directory |
|
||||
| --- | ----------------- |
|
||||
| `auggie` | `.augment/commands` |
|
||||
| `claude` | `.claude/skills` |
|
||||
| `cline` | `.clinerules/workflows` |
|
||||
| `codebuddy` | `.codebuddy/commands` |
|
||||
| `codex` | `.agents/skills` |
|
||||
| `cursor-agent` | `.cursor/skills` |
|
||||
| `firebender` | `.firebender/commands` |
|
||||
| `gemini` | `.gemini/commands` |
|
||||
| `grok` | `.grok/skills` |
|
||||
| `junie` | `.junie/commands`, `.junie/AGENTS.md` |
|
||||
| `kilocode` | `.kilocode/workflows`, `.kilocode/rules/specify-rules.md` |
|
||||
| `qodercli` | `.qoder/commands`, `QODER.md` |
|
||||
| `qwen` | `.qwen/commands`, `QWEN.md` |
|
||||
| `shai` | `.shai/commands`, `SHAI.md` |
|
||||
| `tabnine` | `.tabnine/agent/commands`, `TABNINE.md` |
|
||||
| `trae` | `.trae/skills`, `.trae/rules/project_rules.md` |
|
||||
| `zcode` | `.zcode/skills`, `ZCODE.md` |
|
||||
| `junie` | `.junie/commands` |
|
||||
| `kilocode` | `.kilocode/workflows` |
|
||||
| `kiro-cli` | `.kiro/prompts` |
|
||||
| `qodercli` | `.qoder/commands` |
|
||||
| `qwen` | `.qwen/commands` |
|
||||
| `shai` | `.shai/commands` |
|
||||
| `tabnine` | `.tabnine/agent/commands` |
|
||||
| `trae` | `.trae/skills` |
|
||||
| `zcode` | `.zcode/skills` |
|
||||
|
||||
Integrations that share a command directory with another integration, require dynamic install paths such as `--commands-dir`, or merge shared tool settings are not declared safe by default. They can still be installed alongside another integration with `--force`.
|
||||
|
||||
|
||||
@@ -94,7 +94,12 @@ if [ -f "$_config_file" ]; then
|
||||
[ "$_val" = "false" ] && _enabled=false
|
||||
fi
|
||||
if echo "$_line" | grep -Eq '[[:space:]]+message:'; then
|
||||
_commit_msg=$(echo "$_line" | sed 's/^[^:]*:[[:space:]]*//' | sed 's/^["'\'']//' | sed 's/["'\'']*$//')
|
||||
# Trim trailing whitespace before stripping the closing quote:
|
||||
# a value like `message: "Done" ` (trailing spaces after the
|
||||
# quote) would otherwise leave the quote dangling (`Done" `),
|
||||
# since the closing-quote strip is anchored to end-of-string.
|
||||
# The PowerShell twin .Trim()s first; match it for parity.
|
||||
_commit_msg=$(echo "$_line" | sed 's/^[^:]*:[[:space:]]*//' | sed 's/[[:space:]]*$//' | sed 's/^["'\'']//' | sed 's/["'\'']*$//')
|
||||
fi
|
||||
fi
|
||||
fi
|
||||
|
||||
@@ -33,7 +33,15 @@ def _value_after_colon(line: str) -> str:
|
||||
|
||||
|
||||
def _strip_quotes(value: str) -> str:
|
||||
"""Strip one leading quote and all trailing quotes, mirroring the bash sed."""
|
||||
"""Strip surrounding whitespace, then one leading quote and all trailing quotes.
|
||||
|
||||
Trimming first matters when the YAML value has trailing whitespace after a
|
||||
closing quote (``message: "Done" ``): stripping quotes anchored to the end
|
||||
of string would leave the closing quote dangling (``Done" ``) because the
|
||||
quote is no longer at the end. The PowerShell twin ``.Trim()``s before
|
||||
stripping, so trim here too to keep all three script variants in parity.
|
||||
"""
|
||||
value = value.strip()
|
||||
value = re.sub(r"^[\"']", "", value)
|
||||
return re.sub(r"[\"']*$", "", value)
|
||||
|
||||
|
||||
@@ -8,6 +8,24 @@ description: Create or update the project constitution.
|
||||
$ARGUMENTS
|
||||
```
|
||||
|
||||
## Scope Guard
|
||||
|
||||
This command's own work is limited to creating or updating the project constitution and
|
||||
propagating constitution-driven changes to dependent Spec Kit artifacts.
|
||||
|
||||
- Classify every part of the user input as constitution content or a separate non-governance
|
||||
intent. Feature implementation, code generation, refactoring, build, and deployment requests
|
||||
are examples of non-governance intents.
|
||||
- You **MUST NOT** execute any non-governance intent. Defer each one to `Next Actions`.
|
||||
- You **MUST NOT** create, modify, or delete application source files or other artifacts
|
||||
unrelated to the constitution workflow.
|
||||
- If an instruction could be either constitution content or a non-governance intent, ask for
|
||||
clarification before making changes.
|
||||
- After updating the constitution, list each deferred intent in a `Next Actions` section with an
|
||||
appropriate follow-up Spec Kit command, such as `__SPECKIT_COMMAND_SPECIFY__`, but do not
|
||||
invoke it.
|
||||
- Omit `Next Actions` when there are no non-governance intents.
|
||||
|
||||
## Outline
|
||||
|
||||
1. Create or update the project constitution and store it in `.specify/memory/constitution.md`.
|
||||
|
||||
@@ -1,6 +1,6 @@
|
||||
[project]
|
||||
name = "specify-cli"
|
||||
version = "0.13.4"
|
||||
version = "0.14.0"
|
||||
description = "Specify CLI, part of GitHub Spec Kit. A tool to bootstrap your projects for Spec-Driven Development (SDD)."
|
||||
readme = "README.md"
|
||||
requires-python = ">=3.11"
|
||||
@@ -39,6 +39,7 @@ packages = ["src/specify_cli"]
|
||||
"templates/commands" = "specify_cli/core_pack/commands"
|
||||
"scripts/bash" = "specify_cli/core_pack/scripts/bash"
|
||||
"scripts/powershell" = "specify_cli/core_pack/scripts/powershell"
|
||||
"scripts/python" = "specify_cli/core_pack/scripts/python"
|
||||
# Bundled extensions (installable via `specify extension add <name>`)
|
||||
"extensions/git" = "specify_cli/core_pack/extensions/git"
|
||||
"extensions/agent-context" = "specify_cli/core_pack/extensions/agent-context"
|
||||
|
||||
218
src/specify_cli/_download_security.py
Normal file
218
src/specify_cli/_download_security.py
Normal file
@@ -0,0 +1,218 @@
|
||||
"""Helpers for bounded HTTP downloads."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import io
|
||||
import socket
|
||||
from ipaddress import IPv4Address, IPv6Address, ip_address
|
||||
from typing import NoReturn, TypeVar
|
||||
from urllib.parse import ParseResult, urlparse
|
||||
|
||||
|
||||
ErrorT = TypeVar("ErrorT", bound=Exception)
|
||||
|
||||
MAX_DOWNLOAD_BYTES = 50 * 1024 * 1024
|
||||
READ_CHUNK_SIZE = 64 * 1024
|
||||
|
||||
# Tighter ceiling for responses that are read fully into memory and parsed as
|
||||
# JSON. The 50 MiB MAX_DOWNLOAD_BYTES default is sized for archive/payload
|
||||
# downloads; JSON metadata responses are far smaller, so capping them close to
|
||||
# their real size shrinks the memory-DoS surface and keeps the "too large"
|
||||
# error reachable (rather than only triggering on tens of MiB). Pass it
|
||||
# explicitly at each JSON call site so the intended bound is pinned there.
|
||||
# METADATA covers fixed-shape single-object responses (an OAuth token, one
|
||||
# release's metadata): a few KiB in practice, 1 MiB is already generous.
|
||||
MAX_JSON_METADATA_BYTES = 1 * 1024 * 1024
|
||||
|
||||
|
||||
def _ip_address_without_scope(
|
||||
hostname: str,
|
||||
) -> IPv4Address | IPv6Address | None:
|
||||
"""Parse a canonical IP literal, validating an optional IPv6 zone ID."""
|
||||
if "%" in hostname:
|
||||
# Accept only the RFC 6874 ``%25<zone>`` spelling. Other escapes can
|
||||
# alter the IPv6 address when urllib unquotes the authority.
|
||||
address_text, separator, zone = hostname.partition("%25")
|
||||
if (
|
||||
not separator
|
||||
or ":" not in address_text
|
||||
or "%" in address_text
|
||||
or "%" in zone
|
||||
):
|
||||
return None
|
||||
if not zone or any(
|
||||
not (character.isascii() and (character.isalnum() or character in "._~-"))
|
||||
for character in zone
|
||||
):
|
||||
return None
|
||||
else:
|
||||
address_text = hostname
|
||||
try:
|
||||
address = ip_address(address_text)
|
||||
except ValueError:
|
||||
return None
|
||||
if "%" in hostname and not isinstance(address, IPv6Address):
|
||||
return None
|
||||
return address
|
||||
|
||||
|
||||
def _is_ip_loopback(address: IPv4Address | IPv6Address | None) -> bool:
|
||||
if address is None:
|
||||
return False
|
||||
mapped = getattr(address, "ipv4_mapped", None)
|
||||
return address.is_loopback or bool(mapped and mapped.is_loopback)
|
||||
|
||||
|
||||
def _is_ip_local_redirect_target(
|
||||
address: IPv4Address | IPv6Address | None,
|
||||
) -> bool:
|
||||
"""Treat loopback and unspecified listener aliases as local targets."""
|
||||
if address is None:
|
||||
return False
|
||||
mapped = getattr(address, "ipv4_mapped", None)
|
||||
return _is_ip_loopback(address) or address.is_unspecified or bool(
|
||||
mapped and mapped.is_unspecified
|
||||
)
|
||||
|
||||
|
||||
def _parse_url(url: str) -> ParseResult | None:
|
||||
"""Parse *url*, rejecting missing hosts and malformed ports."""
|
||||
try:
|
||||
parsed = urlparse(url)
|
||||
hostname = parsed.hostname
|
||||
# Accessing ``port`` performs urllib's range and syntax validation.
|
||||
parsed.port
|
||||
except (TypeError, ValueError):
|
||||
return None
|
||||
if not hostname:
|
||||
return None
|
||||
|
||||
if "%" in hostname:
|
||||
# urllib unquotes reg-name/IPv4 authorities before connecting. Reject
|
||||
# them so encoded dots, characters, ports, or brackets cannot make the
|
||||
# validated hostname differ from the effective target. The only safe
|
||||
# percent form retained is a validated bracketed IPv6 zone ID.
|
||||
if _ip_address_without_scope(hostname) is None:
|
||||
return None
|
||||
elif ":" not in hostname:
|
||||
try:
|
||||
hostname.encode("idna")
|
||||
except UnicodeError:
|
||||
return None
|
||||
return parsed
|
||||
|
||||
|
||||
def _is_definite_loopback_host(hostname: str) -> bool:
|
||||
"""Recognize only unambiguous hosts that may safely authorize HTTP."""
|
||||
if not hostname.isascii():
|
||||
return False
|
||||
if hostname == "localhost":
|
||||
return True
|
||||
return _is_ip_loopback(_ip_address_without_scope(hostname))
|
||||
|
||||
|
||||
def _is_potential_local_target_host(hostname: str) -> bool:
|
||||
"""Conservatively classify aliases that could reach a local listener."""
|
||||
if ":" in hostname:
|
||||
return _is_ip_local_redirect_target(_ip_address_without_scope(hostname))
|
||||
try:
|
||||
host = hostname.encode("idna").decode("ascii").lower().removesuffix(".")
|
||||
except UnicodeError:
|
||||
return False
|
||||
if host == "localhost" or host.endswith(".localhost"):
|
||||
return True
|
||||
|
||||
address = _ip_address_without_scope(host)
|
||||
if address is None:
|
||||
# Historical IPv4 spellings are resolver-dependent. They are never
|
||||
# trusted to authorize HTTP, but treating them as potentially local
|
||||
# prevents them from bypassing a remote-to-loopback redirect check.
|
||||
try:
|
||||
address = ip_address(socket.inet_aton(host))
|
||||
except OSError:
|
||||
return False
|
||||
return _is_ip_local_redirect_target(address)
|
||||
|
||||
|
||||
def is_loopback_url(url: str) -> bool:
|
||||
"""Return whether *url* has an unambiguous loopback host."""
|
||||
parsed = _parse_url(url)
|
||||
return parsed is not None and _is_definite_loopback_host(parsed.hostname)
|
||||
|
||||
|
||||
def _is_potential_local_target_url(url: str) -> bool:
|
||||
parsed = _parse_url(url)
|
||||
return parsed is not None and _is_potential_local_target_host(parsed.hostname)
|
||||
|
||||
|
||||
def is_https_or_localhost_http(url: str) -> bool:
|
||||
"""Return True if *url* is HTTPS, or HTTP limited to loopback hosts.
|
||||
|
||||
Shared scheme-safety predicate used by the auth HTTP redirect handler and
|
||||
direct URL validations in CLI download flows.
|
||||
|
||||
A hostname is always required: a URL without one (e.g. ``https:///x``)
|
||||
has no real target and is rejected regardless of scheme.
|
||||
|
||||
The HTTP exception is deliberately limited to unambiguous ``localhost``
|
||||
and canonical IPv4/IPv6 loopback literals. Ambiguous numeric, Unicode, and
|
||||
unspecified-address aliases are classified defensively for redirects but
|
||||
never authorize HTTP. No DNS lookup is performed; DNS and hosts-file
|
||||
aliases require connection-level rebinding protection outside this helper.
|
||||
"""
|
||||
parsed = _parse_url(url)
|
||||
if parsed is None:
|
||||
return False
|
||||
return parsed.scheme == "https" or (
|
||||
parsed.scheme == "http" and _is_definite_loopback_host(parsed.hostname)
|
||||
)
|
||||
|
||||
|
||||
def is_safe_download_redirect(old_url: str, new_url: str) -> bool:
|
||||
"""Return whether a redirect preserves the shared download URL policy."""
|
||||
if not is_https_or_localhost_http(new_url):
|
||||
return False
|
||||
return not _is_potential_local_target_url(new_url) or is_loopback_url(old_url)
|
||||
|
||||
|
||||
def _raise(error_type: type[ErrorT], message: str) -> NoReturn:
|
||||
raise error_type(message)
|
||||
|
||||
|
||||
def read_response_limited(
|
||||
response,
|
||||
*,
|
||||
max_bytes: int = MAX_DOWNLOAD_BYTES,
|
||||
error_type: type[ErrorT] = ValueError,
|
||||
label: str = "download",
|
||||
) -> bytes:
|
||||
"""Read at most *max_bytes* from a response object.
|
||||
|
||||
``response.read(n)`` is only guaranteed to return *up to* ``n`` bytes and may
|
||||
return fewer even when more data is pending (e.g. chunked transfer encoding),
|
||||
so a single ``read(max_bytes + 1)`` cannot enforce the bound on its own. Read
|
||||
in a loop until EOF or until one byte past the limit has been accumulated.
|
||||
|
||||
*max_bytes* is keyword-only. It defaults to the module-wide
|
||||
``MAX_DOWNLOAD_BYTES`` (50 MiB) ceiling for archive/payload downloads;
|
||||
callers with a tighter budget (e.g. small JSON responses) should pass an
|
||||
explicit value so the intended bound is pinned at the call site rather than
|
||||
tracking changes to the shared default.
|
||||
"""
|
||||
if isinstance(max_bytes, bool) or not isinstance(max_bytes, int):
|
||||
raise TypeError("max_bytes must be an integer")
|
||||
if max_bytes < 0:
|
||||
raise ValueError("max_bytes must be non-negative")
|
||||
|
||||
output = io.BytesIO()
|
||||
total = 0
|
||||
limit = max_bytes + 1
|
||||
while total < limit:
|
||||
chunk = response.read(min(READ_CHUNK_SIZE, limit - total))
|
||||
if not chunk:
|
||||
break
|
||||
total += len(chunk)
|
||||
if total > max_bytes:
|
||||
_raise(error_type, f"{label} exceeds maximum size of {max_bytes} bytes")
|
||||
output.write(chunk)
|
||||
return output.getvalue()
|
||||
@@ -100,6 +100,8 @@ def resolve_github_release_asset_api_url(
|
||||
import json
|
||||
import urllib.error
|
||||
|
||||
from specify_cli._download_security import read_response_limited
|
||||
|
||||
parsed = urlparse(download_url)
|
||||
hostname = (parsed.hostname or "").lower()
|
||||
parts = [unquote(part) for part in parsed.path.strip("/").split("/")]
|
||||
@@ -158,10 +160,13 @@ def resolve_github_release_asset_api_url(
|
||||
if redirect_validator is not None:
|
||||
open_kwargs["redirect_validator"] = redirect_validator
|
||||
with open_url_fn(release_url, **open_kwargs) as response:
|
||||
raw_release_data = response.read(max_metadata_bytes + 1)
|
||||
if len(raw_release_data) > max_metadata_bytes:
|
||||
raise ValueError("GitHub release metadata exceeds size limit")
|
||||
release_data = json.loads(raw_release_data)
|
||||
release_data = json.loads(
|
||||
read_response_limited(
|
||||
response,
|
||||
max_bytes=max_metadata_bytes,
|
||||
label=f"GitHub release metadata {release_url}",
|
||||
)
|
||||
)
|
||||
except (
|
||||
urllib.error.URLError,
|
||||
json.JSONDecodeError,
|
||||
|
||||
@@ -4,8 +4,8 @@ Pure helpers for comparing PEP 440 versions and fetching the latest GitHub
|
||||
release tag. The ``self_app`` Typer sub-command group is co-located here so
|
||||
all version-related logic lives in one place.
|
||||
|
||||
Dependencies: stdlib + packaging + ._console only (no other internal imports
|
||||
at module level, keeping this layer thin and circular-import-safe).
|
||||
Dependencies: stdlib + packaging + ._console + ._download_security only
|
||||
(keeping this layer thin and circular-import-safe).
|
||||
"""
|
||||
from __future__ import annotations
|
||||
|
||||
@@ -28,6 +28,7 @@ from pathlib import Path
|
||||
import typer
|
||||
from packaging.version import InvalidVersion, Version
|
||||
|
||||
from ._download_security import MAX_JSON_METADATA_BYTES, read_response_limited
|
||||
from ._console import console
|
||||
|
||||
GITHUB_API_LATEST = "https://api.github.com/repos/github/spec-kit/releases/latest"
|
||||
@@ -119,7 +120,13 @@ def _fetch_latest_release_tag() -> tuple[str | None, str | None]:
|
||||
timeout=5,
|
||||
extra_headers={"Accept": "application/vnd.github+json"},
|
||||
) as resp:
|
||||
payload = json.loads(resp.read().decode("utf-8"))
|
||||
payload = json.loads(
|
||||
read_response_limited(
|
||||
resp,
|
||||
max_bytes=MAX_JSON_METADATA_BYTES,
|
||||
label="GitHub latest release",
|
||||
).decode("utf-8")
|
||||
)
|
||||
tag = payload.get("tag_name")
|
||||
if not isinstance(tag, str) or not tag:
|
||||
raise ValueError("GitHub API response missing valid tag_name")
|
||||
|
||||
@@ -8,6 +8,7 @@ import os
|
||||
import subprocess
|
||||
from typing import TYPE_CHECKING
|
||||
|
||||
from .._download_security import MAX_JSON_METADATA_BYTES, read_response_limited
|
||||
from .base import AuthProvider
|
||||
|
||||
if TYPE_CHECKING:
|
||||
@@ -17,6 +18,20 @@ if TYPE_CHECKING:
|
||||
_ADO_RESOURCE_ID = "499b84ac-1321-427f-aa17-267ca6975798"
|
||||
|
||||
|
||||
class _TokenResponseTooLarge(Exception):
|
||||
"""Raised when an Azure AD token response exceeds the bounded read limit."""
|
||||
|
||||
|
||||
def _extract_token(payload: object, key: str) -> str | None:
|
||||
"""Return a normalized token from a JSON object, or None for other shapes."""
|
||||
if not isinstance(payload, dict):
|
||||
return None
|
||||
token = payload.get(key)
|
||||
if not isinstance(token, str):
|
||||
return None
|
||||
return token.strip() or None
|
||||
|
||||
|
||||
class AzureDevOpsAuth(AuthProvider):
|
||||
"""Azure DevOps authentication provider.
|
||||
|
||||
@@ -74,8 +89,7 @@ class AzureDevOpsAuth(AuthProvider):
|
||||
if result.returncode != 0:
|
||||
return None
|
||||
payload = _json.loads(result.stdout)
|
||||
token = payload.get("accessToken", "").strip()
|
||||
return token or None
|
||||
return _extract_token(payload, "accessToken")
|
||||
except (
|
||||
OSError,
|
||||
subprocess.TimeoutExpired,
|
||||
@@ -119,9 +133,37 @@ class AzureDevOpsAuth(AuthProvider):
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
try:
|
||||
with urllib.request.urlopen(req, timeout=30) as resp: # noqa: S310
|
||||
payload = _json.loads(resp.read().decode("utf-8"))
|
||||
token = payload.get("access_token", "").strip()
|
||||
return token or None
|
||||
except (urllib.error.URLError, OSError, _json.JSONDecodeError, KeyError):
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
|
||||
def reject_token_redirect(_old_url: str, new_url: str) -> None:
|
||||
# A 307/308 redirect preserves this POST body, including the
|
||||
# client_secret. Refuse every redirect so credentials cannot
|
||||
# leave the fixed Microsoft token endpoint.
|
||||
raise urllib.error.URLError(
|
||||
f"Azure AD token request must not be redirected to {new_url}"
|
||||
)
|
||||
|
||||
opener = urllib.request.build_opener(
|
||||
_StripAuthOnRedirect((), reject_token_redirect)
|
||||
)
|
||||
with opener.open(req, timeout=30) as resp: # noqa: S310
|
||||
payload = _json.loads(
|
||||
read_response_limited(
|
||||
resp,
|
||||
max_bytes=MAX_JSON_METADATA_BYTES,
|
||||
error_type=_TokenResponseTooLarge,
|
||||
label="Azure DevOps token response",
|
||||
).decode("utf-8")
|
||||
)
|
||||
return _extract_token(payload, "access_token")
|
||||
except (
|
||||
urllib.error.URLError,
|
||||
OSError,
|
||||
_json.JSONDecodeError,
|
||||
UnicodeDecodeError,
|
||||
_TokenResponseTooLarge,
|
||||
):
|
||||
# Network failure, malformed JSON, or an oversized response — fall
|
||||
# through to the next strategy. Unrelated programming errors (other
|
||||
# ValueErrors, KeyErrors) intentionally propagate so they surface.
|
||||
return None
|
||||
|
||||
@@ -17,6 +17,7 @@ from fnmatch import fnmatch
|
||||
from typing import Callable
|
||||
from urllib.parse import urlparse
|
||||
|
||||
from .._download_security import is_safe_download_redirect
|
||||
from . import get_provider
|
||||
from .config import AuthConfigEntry, _default_config_path, find_entries_for_url, load_auth_config
|
||||
|
||||
@@ -60,8 +61,23 @@ def _hostname_in_hosts(hostname: str, hosts: tuple[str, ...]) -> bool:
|
||||
RedirectValidator = Callable[[str, str], None]
|
||||
|
||||
|
||||
def _validate_strict_redirect(old_url: str, new_url: str) -> None:
|
||||
if not is_safe_download_redirect(old_url, new_url):
|
||||
raise urllib.error.URLError(
|
||||
f"unsafe redirect to {new_url}: target must use HTTPS with a hostname, "
|
||||
"must not enter a local target from a remote host, and may use HTTP only "
|
||||
"within loopback (for example localhost, 127.0.0.1, ::1)"
|
||||
)
|
||||
|
||||
|
||||
class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
|
||||
"""Drop ``Authorization`` when a redirect leaves trusted hosts or downgrades."""
|
||||
"""Redirect handler that guards every redirect it is installed for.
|
||||
|
||||
1. Run any caller-provided redirect validator.
|
||||
2. Reject redirects that are not HTTPS with a hostname. HTTP loopback is
|
||||
allowed only when the previous hop is also loopback.
|
||||
3. Drop ``Authorization`` when a redirect leaves trusted hosts or downgrades.
|
||||
"""
|
||||
|
||||
def __init__(
|
||||
self,
|
||||
@@ -75,6 +91,8 @@ class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
|
||||
def redirect_request(self, req, fp, code, msg, headers, newurl):
|
||||
try:
|
||||
new_parsed = urlparse(newurl)
|
||||
# Force urllib's syntax and range validation before following.
|
||||
new_parsed.port
|
||||
except ValueError as exc:
|
||||
# Malformed redirect target (e.g. unterminated IPv6 bracket).
|
||||
# Surface as URLError so callers' download error handling applies.
|
||||
@@ -82,6 +100,7 @@ class _StripAuthOnRedirect(urllib.request.HTTPRedirectHandler):
|
||||
|
||||
if self._redirect_validator is not None:
|
||||
self._redirect_validator(req.full_url, newurl)
|
||||
_validate_strict_redirect(req.full_url, newurl)
|
||||
|
||||
original_auth = (
|
||||
req.get_header("Authorization")
|
||||
@@ -155,6 +174,12 @@ def open_url(
|
||||
*extra_headers* (e.g. ``Accept``) are merged into every attempt.
|
||||
*redirect_validator*, when provided, is called with ``(old_url, new_url)``
|
||||
before following each redirect and may raise to reject the redirect.
|
||||
|
||||
Every attempt uses an isolated opener so a process-wide opener installed
|
||||
with ``urllib.request.install_opener`` cannot replace the redirect guard.
|
||||
Redirect scheme safety: every attempt goes through
|
||||
``_StripAuthOnRedirect``, which rejects redirects to non-HTTPS URLs except
|
||||
HTTP between loopback URLs, and rejects remote-to-local redirects.
|
||||
"""
|
||||
entries = find_entries_for_url(url, _load_config())
|
||||
|
||||
@@ -188,7 +213,7 @@ def open_url(
|
||||
|
||||
# No entry worked (or none matched) — unauthenticated fallback
|
||||
req = _make_req({})
|
||||
if redirect_validator is not None:
|
||||
opener = urllib.request.build_opener(_StripAuthOnRedirect((), redirect_validator))
|
||||
return opener.open(req, timeout=timeout)
|
||||
return urllib.request.urlopen(req, timeout=timeout) # noqa: S310
|
||||
# No auth is attached on this path, so the handler's host list is empty:
|
||||
# here it runs redirect validation only, not auth stripping.
|
||||
opener = urllib.request.build_opener(_StripAuthOnRedirect((), redirect_validator))
|
||||
return opener.open(req, timeout=timeout)
|
||||
|
||||
@@ -60,7 +60,13 @@ def dump_yaml(path: Path, data: Any, *, within: Path | None = None) -> Path:
|
||||
try:
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
with path.open("w", encoding="utf-8") as handle:
|
||||
yaml.safe_dump(data, handle, sort_keys=False, default_flow_style=False)
|
||||
yaml.safe_dump(
|
||||
data,
|
||||
handle,
|
||||
sort_keys=False,
|
||||
default_flow_style=False,
|
||||
allow_unicode=True,
|
||||
)
|
||||
except OSError as exc:
|
||||
raise BundlerError(f"Could not write {path}: {exc}") from exc
|
||||
return path
|
||||
|
||||
@@ -111,8 +111,10 @@ class BundleManifest:
|
||||
license=str(bundle_raw.get("license", "")).strip(),
|
||||
)
|
||||
|
||||
requires_raw = data.get("requires") or {}
|
||||
if not isinstance(requires_raw, dict):
|
||||
requires_raw = data.get("requires")
|
||||
if requires_raw is None:
|
||||
requires_raw = {}
|
||||
elif not isinstance(requires_raw, dict):
|
||||
raise BundlerError("'requires' must be a mapping when present.")
|
||||
requires = Requires(
|
||||
speckit_version=str(requires_raw.get("speckit_version", "")).strip(),
|
||||
@@ -130,8 +132,10 @@ class BundleManifest:
|
||||
if isinstance(integration_raw, dict) and integration_raw.get("id"):
|
||||
integration = IntegrationRef(id=str(integration_raw["id"]).strip())
|
||||
|
||||
provides = data.get("provides") or {}
|
||||
if not isinstance(provides, dict):
|
||||
provides = data.get("provides")
|
||||
if provides is None:
|
||||
provides = {}
|
||||
elif not isinstance(provides, dict):
|
||||
raise BundlerError("'provides' must be a mapping when present.")
|
||||
|
||||
tags_raw = data.get("tags")
|
||||
|
||||
@@ -142,4 +142,10 @@ def _collect_files(
|
||||
# Skip symlinked files to avoid escaping the bundle directory.
|
||||
continue
|
||||
collected.append(path)
|
||||
return sorted(collected)
|
||||
# Order by the canonical POSIX arcname (the same key build_bundle uses to
|
||||
# NAME each member), not by pathlib.Path comparison. Path ordering is
|
||||
# platform-dependent (Windows folds case and uses backslash separators),
|
||||
# which would lay out zip members differently across build hosts and break
|
||||
# the byte-for-byte reproducible-build guarantee even though the member
|
||||
# names are identical.
|
||||
return sorted(collected, key=lambda p: p.relative_to(bundle_dir).as_posix())
|
||||
|
||||
@@ -64,8 +64,15 @@ def with_integration_setting(
|
||||
elif raw_options is not None:
|
||||
current.pop("parsed_options", None)
|
||||
|
||||
# Recompute the separator from the options actually STORED on ``current``
|
||||
# after the update, not the raw ``parsed_options`` argument. When only
|
||||
# ``script_type`` changes (``parsed_options`` and ``raw_options`` both
|
||||
# None), the previously-stored ``parsed_options`` are retained above, so
|
||||
# deriving the separator from the argument (None) would drop an
|
||||
# options-dependent separator (e.g. Copilot ``--skills`` -> "-") back to
|
||||
# the default ".".
|
||||
current["invoke_separator"] = integration.effective_invoke_separator(
|
||||
parsed_options, project_root
|
||||
current.get("parsed_options"), project_root
|
||||
)
|
||||
settings[key] = current
|
||||
return settings
|
||||
|
||||
@@ -138,8 +138,14 @@ class ClineIntegration(MarkdownIntegration):
|
||||
content,
|
||||
)
|
||||
|
||||
def post_process_content(self, content: str) -> str:
|
||||
"""Apply Cline-specific transformations to command content."""
|
||||
def post_process_command_content(self, content: str) -> str:
|
||||
"""Apply Cline-specific transformations to command content.
|
||||
|
||||
Overrides the ``IntegrationBase`` hook of the same name so that
|
||||
``CommandRegistrar.register_commands()`` (which dispatches to
|
||||
``post_process_command_content``) applies these transforms to
|
||||
extension/preset command files too, not just core commands.
|
||||
"""
|
||||
updated = self._inject_hook_command_note(content)
|
||||
updated = self._rewrite_handoff_references(updated)
|
||||
return updated
|
||||
@@ -169,7 +175,7 @@ class ClineIntegration(MarkdownIntegration):
|
||||
content_bytes = path.read_bytes()
|
||||
content = content_bytes.decode("utf-8")
|
||||
|
||||
updated = self.post_process_content(content)
|
||||
updated = self.post_process_command_content(content)
|
||||
|
||||
if updated != content:
|
||||
path.write_bytes(updated.encode("utf-8"))
|
||||
|
||||
@@ -13,6 +13,13 @@ _KIRO_ARG_FALLBACK = "(the user will provide the argument in this conversation)"
|
||||
|
||||
class KiroCliIntegration(MarkdownIntegration):
|
||||
key = "kiro-cli"
|
||||
# Kiro CLI keeps everything under a static, isolated agent root
|
||||
# (``.kiro/`` with commands in ``.kiro/prompts``) that no other
|
||||
# integration writes to, so it is safe to install alongside others
|
||||
# (issue #3471). IntegrationBase defaults this to False; declaring it
|
||||
# True here is the actual behavior change this integration opts into.
|
||||
# The registry's multi-install-safe contract tests enforce that
|
||||
# isolation for every integration setting this flag.
|
||||
multi_install_safe = True
|
||||
config = {
|
||||
"name": "Kiro CLI",
|
||||
@@ -27,10 +34,3 @@ class KiroCliIntegration(MarkdownIntegration):
|
||||
"args": _KIRO_ARG_FALLBACK,
|
||||
"extension": ".md",
|
||||
}
|
||||
|
||||
# Kiro CLI keeps everything under a static, isolated agent root
|
||||
# (``.kiro/`` with commands in ``.kiro/prompts``) that no other
|
||||
# integration writes to, so it is safe to install alongside others
|
||||
# (issue #3471). The registry's multi-install-safe contract tests
|
||||
# enforce that isolation for every integration setting this flag.
|
||||
multi_install_safe = True
|
||||
|
||||
@@ -27,6 +27,7 @@ class LingmaIntegration(SkillsIntegration):
|
||||
"args": "$ARGUMENTS",
|
||||
"extension": "/SKILL.md",
|
||||
}
|
||||
multi_install_safe = True
|
||||
|
||||
@classmethod
|
||||
def options(cls) -> list[IntegrationOption]:
|
||||
|
||||
@@ -16,6 +16,10 @@ import yaml
|
||||
from rich.markup import escape as _escape_markup
|
||||
|
||||
from .._console import console
|
||||
from .._download_security import (
|
||||
is_https_or_localhost_http,
|
||||
is_safe_download_redirect,
|
||||
)
|
||||
|
||||
preset_app = typer.Typer(
|
||||
name="preset",
|
||||
@@ -102,38 +106,25 @@ def preset_add(
|
||||
|
||||
elif from_url:
|
||||
# Validate URL scheme before downloading
|
||||
from ipaddress import ip_address
|
||||
from urllib.parse import urlparse as _urlparse
|
||||
|
||||
try:
|
||||
_parsed = _urlparse(from_url)
|
||||
_parsed.port
|
||||
except ValueError:
|
||||
console.print(f"[red]Error:[/red] Invalid URL: {_escape_markup(from_url)}")
|
||||
raise typer.Exit(1)
|
||||
|
||||
def _is_allowed_download_url(parsed_url):
|
||||
host = parsed_url.hostname
|
||||
if not host:
|
||||
return False
|
||||
is_loopback = host == "localhost"
|
||||
if not is_loopback:
|
||||
try:
|
||||
is_loopback = ip_address(host).is_loopback
|
||||
except ValueError:
|
||||
# Host is not an IP literal (e.g., a regular hostname); treat as non-loopback.
|
||||
pass
|
||||
return parsed_url.scheme == "https" or (parsed_url.scheme == "http" and is_loopback)
|
||||
|
||||
def _validate_download_redirect(old_url, new_url):
|
||||
if not _is_allowed_download_url(_urlparse(new_url)):
|
||||
if not is_safe_download_redirect(old_url, new_url):
|
||||
import urllib.error
|
||||
|
||||
raise urllib.error.URLError(
|
||||
"redirect target must use HTTPS with a hostname, "
|
||||
"or HTTP for localhost/loopback"
|
||||
"redirect target must use HTTPS without entering a local "
|
||||
"target, or stay within loopback over HTTP"
|
||||
)
|
||||
|
||||
if not _is_allowed_download_url(_parsed):
|
||||
if not is_https_or_localhost_http(from_url):
|
||||
console.print(
|
||||
"[red]Error:[/red] URL must use HTTPS with a hostname, "
|
||||
"or HTTP for localhost/loopback."
|
||||
@@ -167,7 +158,7 @@ def preset_add(
|
||||
redirect_validator=_validate_download_redirect,
|
||||
) as response:
|
||||
final_url = response.geturl() if hasattr(response, "geturl") else from_url
|
||||
if not _is_allowed_download_url(_urlparse(final_url)):
|
||||
if not is_https_or_localhost_http(final_url):
|
||||
console.print(
|
||||
"[red]Error:[/red] Preset URL redirected to a disallowed URL: "
|
||||
f"{final_url}. Redirect targets must use HTTPS with a hostname, "
|
||||
|
||||
@@ -20,6 +20,10 @@ import yaml
|
||||
from rich.markup import escape as _escape_markup
|
||||
|
||||
from .._console import console, err_console
|
||||
from .._download_security import (
|
||||
is_https_or_localhost_http,
|
||||
is_safe_download_redirect,
|
||||
)
|
||||
from .._project import _resolve_init_dir_override
|
||||
|
||||
workflow_app = typer.Typer(
|
||||
@@ -383,27 +387,12 @@ _RESERVED_WORKFLOW_IDS: frozenset[str] = frozenset({"overlays", "runs", "steps"}
|
||||
def _reject_insecure_download_redirect(old_url: str, new_url: str) -> None:
|
||||
"""Reject insecure redirects before they are followed."""
|
||||
import urllib.error
|
||||
from ipaddress import ip_address
|
||||
from urllib.parse import urlparse
|
||||
|
||||
def _is_loopback_http(url: str) -> bool:
|
||||
parsed = urlparse(url)
|
||||
if parsed.scheme != "http":
|
||||
return False
|
||||
host = parsed.hostname or ""
|
||||
if host == "localhost":
|
||||
return True
|
||||
try:
|
||||
return ip_address(host).is_loopback
|
||||
except ValueError:
|
||||
return False
|
||||
|
||||
if urlparse(new_url).scheme == "https":
|
||||
return
|
||||
if _is_loopback_http(old_url) and _is_loopback_http(new_url):
|
||||
if is_safe_download_redirect(old_url, new_url):
|
||||
return
|
||||
raise urllib.error.URLError(
|
||||
"redirect target must use HTTPS; loopback HTTP may only redirect from loopback HTTP"
|
||||
"redirect target must use HTTPS without entering a local target; "
|
||||
"loopback HTTP may only redirect from another loopback URL"
|
||||
)
|
||||
|
||||
|
||||
@@ -1579,24 +1568,15 @@ def workflow_add(
|
||||
else (source if source.startswith(("http://", "https://")) else None)
|
||||
)
|
||||
if download_url is not None:
|
||||
from ipaddress import ip_address
|
||||
from urllib.parse import urlparse
|
||||
from specify_cli.authentication.http import open_url as _open_url
|
||||
|
||||
try:
|
||||
parsed_src = urlparse(download_url)
|
||||
urlparse(download_url).port
|
||||
except ValueError:
|
||||
console.print(f"[red]Error:[/red] Invalid URL: {_escape_markup(download_url)}")
|
||||
raise typer.Exit(1)
|
||||
src_host = parsed_src.hostname or ""
|
||||
src_loopback = src_host == "localhost"
|
||||
if not src_loopback:
|
||||
try:
|
||||
src_loopback = ip_address(src_host).is_loopback
|
||||
except ValueError:
|
||||
# Host is not an IP literal (e.g., a DNS name); keep default non-loopback.
|
||||
pass
|
||||
if parsed_src.scheme != "https" and not (parsed_src.scheme == "http" and src_loopback):
|
||||
if not is_https_or_localhost_http(download_url):
|
||||
console.print("[red]Error:[/red] Only HTTPS URLs are allowed, except HTTP for localhost.")
|
||||
raise typer.Exit(1)
|
||||
|
||||
@@ -1647,16 +1627,7 @@ def workflow_add(
|
||||
redirect_validator=_reject_insecure_download_redirect,
|
||||
) as resp:
|
||||
final_url = resp.geturl()
|
||||
final_parsed = urlparse(final_url)
|
||||
final_host = final_parsed.hostname or ""
|
||||
final_lb = final_host == "localhost"
|
||||
if not final_lb:
|
||||
try:
|
||||
final_lb = ip_address(final_host).is_loopback
|
||||
except ValueError:
|
||||
# Redirect host is not an IP literal; keep loopback as determined above.
|
||||
pass
|
||||
if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_lb):
|
||||
if not is_https_or_localhost_http(final_url):
|
||||
console.print(
|
||||
f"[red]Error:[/red] URL redirected to non-HTTPS: {_escape_markup(final_url)}"
|
||||
)
|
||||
@@ -1788,27 +1759,17 @@ def _install_workflow_from_catalog(
|
||||
raise typer.Exit(1)
|
||||
|
||||
# Validate URL scheme (HTTPS required, HTTP allowed for localhost only)
|
||||
from ipaddress import ip_address
|
||||
from urllib.parse import urlparse
|
||||
|
||||
try:
|
||||
parsed_url = urlparse(workflow_url)
|
||||
url_host = parsed_url.hostname or ""
|
||||
parsed_url.port
|
||||
except ValueError:
|
||||
console.print(
|
||||
f"[red]Error:[/red] Workflow '{safe_wf_id}' has a malformed install URL."
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
is_loopback = False
|
||||
if url_host == "localhost":
|
||||
is_loopback = True
|
||||
else:
|
||||
try:
|
||||
is_loopback = ip_address(url_host).is_loopback
|
||||
except ValueError:
|
||||
# Host is not an IP literal (e.g., a regular hostname); treat as non-loopback.
|
||||
pass
|
||||
if parsed_url.scheme != "https" and not (parsed_url.scheme == "http" and is_loopback):
|
||||
if not is_https_or_localhost_http(workflow_url):
|
||||
console.print(
|
||||
f"[red]Error:[/red] Workflow '{safe_wf_id}' has an invalid install URL. "
|
||||
"Only HTTPS URLs are allowed, except HTTP for localhost/loopback."
|
||||
@@ -1862,16 +1823,7 @@ def _install_workflow_from_catalog(
|
||||
) as response:
|
||||
# Validate final URL after redirects
|
||||
final_url = response.geturl()
|
||||
final_parsed = urlparse(final_url)
|
||||
final_host = final_parsed.hostname or ""
|
||||
final_loopback = final_host == "localhost"
|
||||
if not final_loopback:
|
||||
try:
|
||||
final_loopback = ip_address(final_host).is_loopback
|
||||
except ValueError:
|
||||
# Host is not an IP literal (e.g., a regular hostname); treat as non-loopback.
|
||||
pass
|
||||
if final_parsed.scheme != "https" and not (final_parsed.scheme == "http" and final_loopback):
|
||||
if not is_https_or_localhost_http(final_url):
|
||||
_safe_discard_staged_workflow_file(staged_file, workflow_dir, existed_before)
|
||||
console.print(
|
||||
f"[red]Error:[/red] Workflow '{safe_wf_id}' redirected to non-HTTPS URL: {_escape_markup(final_url)}"
|
||||
@@ -2694,28 +2646,17 @@ def workflow_step_add(
|
||||
)
|
||||
raise typer.Exit(1)
|
||||
|
||||
from urllib.parse import urlparse
|
||||
from specify_cli.authentication.http import open_url as _open_url
|
||||
|
||||
def _safe_fetch(url: str) -> bytes:
|
||||
parsed = urlparse(url)
|
||||
is_localhost = parsed.hostname in ("localhost", "127.0.0.1", "::1")
|
||||
if parsed.scheme != "https" and not (parsed.scheme == "http" and is_localhost):
|
||||
if not is_https_or_localhost_http(url):
|
||||
raise ValueError(f"Refusing to fetch from non-HTTPS URL: {url}")
|
||||
if not parsed.hostname:
|
||||
raise ValueError(f"Refusing to fetch from URL with no hostname: {url}")
|
||||
with _open_url(
|
||||
url, timeout=30, redirect_validator=_reject_insecure_download_redirect
|
||||
) as resp:
|
||||
final_url = resp.geturl()
|
||||
final_parsed = urlparse(final_url)
|
||||
final_is_localhost = final_parsed.hostname in ("localhost", "127.0.0.1", "::1")
|
||||
if final_parsed.scheme != "https" and not (
|
||||
final_parsed.scheme == "http" and final_is_localhost
|
||||
):
|
||||
if not is_https_or_localhost_http(final_url):
|
||||
raise ValueError(f"Redirect to non-HTTPS URL: {final_url}")
|
||||
if not final_parsed.hostname:
|
||||
raise ValueError(f"Redirect to URL with no hostname: {final_url}")
|
||||
return _read_response_within_limit(resp)
|
||||
|
||||
_validate_step_id_or_exit(step_id)
|
||||
|
||||
@@ -292,9 +292,21 @@ def _traverse_and_apply(
|
||||
cases[case_key] = _traverse_and_apply(case_steps, edits_by_anchor, sources)
|
||||
result.append(step)
|
||||
|
||||
# Insert after (highest priority closest to anchor — reversed merge order).
|
||||
for layer, edit in reversed(edits):
|
||||
if edit.operation == "insert_after":
|
||||
# Insert after: higher-priority overlays land closer to the anchor
|
||||
# (reversed merge order), but a single overlay's own inserts must keep
|
||||
# their declared order — mirroring the forward insert_before loop above.
|
||||
# Reversing the whole flat list would also flip an overlay's own edits,
|
||||
# so group contiguous same-layer edits and reverse the GROUP order only.
|
||||
after_groups: list[list[tuple[OverlayLayer, OverlayEdit]]] = []
|
||||
for layer, edit in edits:
|
||||
if edit.operation != "insert_after":
|
||||
continue
|
||||
if after_groups and after_groups[-1][0][0] is layer:
|
||||
after_groups[-1].append((layer, edit))
|
||||
else:
|
||||
after_groups.append([(layer, edit)])
|
||||
for group in reversed(after_groups):
|
||||
for layer, edit in group:
|
||||
new_step = copy.deepcopy(edit.step)
|
||||
_record_sources_recursively(new_step, layer.source, sources)
|
||||
result.append(new_step)
|
||||
|
||||
@@ -26,7 +26,7 @@ class GateStep(StepBase):
|
||||
later with ``specify workflow resume``.
|
||||
|
||||
The user's choice is stored in ``output.choice``. ``on_reject``
|
||||
controls abort / skip behaviour.
|
||||
controls abort / skip / retry behaviour.
|
||||
"""
|
||||
|
||||
type_key = "gate"
|
||||
|
||||
@@ -59,7 +59,7 @@ class InitStep(StepBase):
|
||||
Extra options for the integration (e.g. ``"--skills"`` or
|
||||
``"--commands-dir .myagent/cmds"``).
|
||||
``script``
|
||||
Script type, ``sh`` or ``ps``.
|
||||
Script type, ``sh``, ``ps``, or ``py``.
|
||||
``force``
|
||||
Merge/overwrite without confirmation when the directory is not
|
||||
empty.
|
||||
|
||||
@@ -14,6 +14,25 @@ $ARGUMENTS
|
||||
|
||||
You **MUST** consider the user input before proceeding (if not empty).
|
||||
|
||||
## Scope Guard
|
||||
|
||||
This command's own work is limited to updating the project constitution and propagating
|
||||
constitution-driven changes to the dependent artifacts identified in this command.
|
||||
|
||||
- Classify every part of the user input as either constitution content or a separate,
|
||||
non-governance intent.
|
||||
- If the input includes feature implementation, code generation, refactoring, building, or
|
||||
deployment requests, you **MUST NOT** execute them. Extract them as deferred intents instead.
|
||||
- You **MUST NOT** create, modify, or delete application source files, feature routes,
|
||||
components, tests, deployment files, or other artifacts unrelated to the constitution
|
||||
workflow and its required propagation.
|
||||
- If it is unclear whether an instruction is constitution content, ask for clarification before
|
||||
making changes.
|
||||
- After completing the constitution update, include a `Next Actions` section for each deferred
|
||||
intent. List the original intent and suggest the appropriate follow-up Spec Kit command, such
|
||||
as `__SPECKIT_COMMAND_SPECIFY__`, without invoking it.
|
||||
- If there are no non-governance intents, omit the `Next Actions` section.
|
||||
|
||||
## Pre-Execution Checks
|
||||
|
||||
**Check for extension hooks (before constitution update)**:
|
||||
@@ -104,6 +123,7 @@ Follow this execution flow:
|
||||
- New version and bump rationale.
|
||||
- Any files flagged for manual follow-up.
|
||||
- Suggested commit message (e.g., `docs: amend constitution to vX.Y.Z (principle additions + governance update)`).
|
||||
- A `Next Actions` section for any deferred non-governance intents.
|
||||
|
||||
Formatting & Style Requirements:
|
||||
|
||||
|
||||
@@ -139,9 +139,9 @@ Given that feature description, do this:
|
||||
7. Identify Key Entities (if data involved)
|
||||
8. Return: SUCCESS (spec ready for planning)
|
||||
|
||||
6. Write the specification to SPEC_FILE using the template structure, replacing placeholders with concrete details derived from the feature description (arguments) while preserving section order and headings.
|
||||
7. Write the specification to SPEC_FILE using the template structure, replacing placeholders with concrete details derived from the feature description (arguments) while preserving section order and headings.
|
||||
|
||||
7. **Specification Quality Validation**: After writing the initial spec, validate it against quality criteria:
|
||||
8. **Specification Quality Validation**: After writing the initial spec, validate it against quality criteria:
|
||||
|
||||
a. **Create Spec Quality Checklist**: Generate a checklist file at `SPECIFY_FEATURE_DIRECTORY/checklists/requirements.md` using the checklist template structure with these validation items:
|
||||
|
||||
|
||||
@@ -134,3 +134,34 @@ def test_string_integration_rejected_not_silently_dropped():
|
||||
data["integration"] = "copilot"
|
||||
with pytest.raises(BundlerError, match="'integration' must be a mapping when present"):
|
||||
BundleManifest.from_dict(data)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", [[], "", 0, False, "extensions"])
|
||||
def test_non_mapping_provides_rejected_including_falsy(bad):
|
||||
# `data.get("provides") or {}` coerced a FALSY non-mapping ([], '', 0, False)
|
||||
# to {} before the type check, so a malformed manifest passed validation as
|
||||
# a bundle that provides nothing. Only an absent/None value means "empty".
|
||||
data = valid_manifest_dict()
|
||||
data["provides"] = bad
|
||||
with pytest.raises(BundlerError, match="'provides' must be a mapping when present"):
|
||||
BundleManifest.from_dict(data)
|
||||
|
||||
|
||||
@pytest.mark.parametrize("bad", [[], "", 0, False, "speckit>=0.1"])
|
||||
def test_non_mapping_requires_rejected_including_falsy(bad):
|
||||
# Same falsy-coercion hole for `requires`.
|
||||
data = valid_manifest_dict()
|
||||
data["requires"] = bad
|
||||
with pytest.raises(BundlerError, match="'requires' must be a mapping when present"):
|
||||
BundleManifest.from_dict(data)
|
||||
|
||||
|
||||
def test_absent_provides_and_requires_do_not_raise_mapping_error():
|
||||
# Absent (None) optional mappings default to empty and must NOT trigger the
|
||||
# "must be a mapping when present" guard — that is reserved for present
|
||||
# non-mappings. (Structural completeness, e.g. requires.speckit_version, is
|
||||
# a separate concern checked by structural_errors().)
|
||||
data = valid_manifest_dict()
|
||||
data.pop("provides", None)
|
||||
data.pop("requires", None)
|
||||
BundleManifest.from_dict(data) # does not raise BundlerError
|
||||
|
||||
40
tests/contract/test_wheel_core_pack_scripts.py
Normal file
40
tests/contract/test_wheel_core_pack_scripts.py
Normal file
@@ -0,0 +1,40 @@
|
||||
"""Contract tests for the script variants bundled into the wheel's core_pack.
|
||||
|
||||
``specify init --script <type>`` installs from ``specify_cli/core_pack/scripts/``
|
||||
when the CLI runs from a wheel. Any script variant that lives in the repository
|
||||
must therefore be force-included at build time, otherwise the generated
|
||||
commands reference scripts the released package never ships (#3665).
|
||||
"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import tomllib
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).parents[2]
|
||||
|
||||
|
||||
def _force_include() -> dict[str, str]:
|
||||
with (REPO_ROOT / "pyproject.toml").open("rb") as pyproject_file:
|
||||
pyproject = tomllib.load(pyproject_file)
|
||||
return pyproject["tool"]["hatch"]["build"]["targets"]["wheel"]["force-include"]
|
||||
|
||||
|
||||
def test_every_script_variant_is_bundled_into_core_pack():
|
||||
force_include = _force_include()
|
||||
variants = sorted(
|
||||
path.name for path in (REPO_ROOT / "scripts").iterdir() if path.is_dir()
|
||||
)
|
||||
|
||||
assert variants, "expected at least one script variant under scripts/"
|
||||
for variant in variants:
|
||||
assert force_include.get(f"scripts/{variant}") == (
|
||||
f"specify_cli/core_pack/scripts/{variant}"
|
||||
), f"scripts/{variant} is missing from the wheel force-include list"
|
||||
|
||||
|
||||
def test_python_script_variant_is_bundled():
|
||||
# Explicit regression guard for #3665: `--script py` shipped skills that
|
||||
# invoked python3 .specify/scripts/python/*.py while the wheel bundled
|
||||
# only the bash and PowerShell variants.
|
||||
assert _force_include()["scripts/python"] == "specify_cli/core_pack/scripts/python"
|
||||
@@ -515,6 +515,32 @@ class TestAutoCommitParity:
|
||||
assert p.stderr.strip() == b.stderr.strip()
|
||||
assert self._last_message(bash_proj) == self._last_message(py_proj) == "spec done"
|
||||
|
||||
def test_custom_message_with_trailing_whitespace_after_quote(self, tmp_path: Path):
|
||||
"""Trailing whitespace after a closing quote must not leave the quote
|
||||
dangling in the commit message. A raw close-quote strip anchored to
|
||||
end-of-string skips the quote when spaces follow it (``spec done" ``);
|
||||
trimming first (matching the PowerShell twin) yields a clean message and
|
||||
keeps bash/python in parity."""
|
||||
bash_proj, py_proj = _twin_projects(tmp_path)
|
||||
config = (
|
||||
"auto_commit:\n"
|
||||
" default: false\n"
|
||||
" after_specify:\n"
|
||||
" enabled: true\n"
|
||||
' message: "spec done" \n' # trailing spaces after the closing quote
|
||||
)
|
||||
for proj in (bash_proj, py_proj):
|
||||
_write_config(proj, config)
|
||||
self._dirty(proj)
|
||||
b = _run_bash("auto-commit.sh", bash_proj, "after_specify")
|
||||
p = _run_py("auto-commit", py_proj, "after_specify")
|
||||
_assert_parity(b, p)
|
||||
assert (
|
||||
self._last_message(bash_proj)
|
||||
== self._last_message(py_proj)
|
||||
== "spec done"
|
||||
)
|
||||
|
||||
def test_default_true_applies_to_unlisted_event(self, tmp_path: Path):
|
||||
bash_proj, py_proj = _twin_projects(tmp_path)
|
||||
for proj in (bash_proj, py_proj):
|
||||
|
||||
@@ -1,15 +1,46 @@
|
||||
"""HTTP test helpers shared by version-related CLI tests."""
|
||||
"""HTTP test helpers shared by CLI tests."""
|
||||
|
||||
import io
|
||||
import json
|
||||
import urllib.request
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
import pytest
|
||||
|
||||
|
||||
def mock_urlopen_response(payload: dict) -> MagicMock:
|
||||
"""Build a urlopen context-manager mock whose read returns JSON."""
|
||||
body = json.dumps(payload).encode("utf-8")
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = body
|
||||
resp.read.side_effect = io.BytesIO(body).read
|
||||
cm = MagicMock()
|
||||
cm.__enter__.return_value = resp
|
||||
cm.__exit__.return_value = False
|
||||
return cm
|
||||
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def route_opener_open_through_urlopen(monkeypatch):
|
||||
"""Route build_opener().open through urllib.request.urlopen.
|
||||
|
||||
``open_url(...)`` fetches via ``build_opener(...).open()``, which bypasses
|
||||
``urllib.request.urlopen`` — and with it the urlopen patches these test
|
||||
modules are built on.
|
||||
Delegating ``open()`` to urlopen at call time keeps those patches
|
||||
effective; the redirect handler's own behavior is covered by
|
||||
``TestRedirectStripping`` in test_authentication.py.
|
||||
|
||||
Import this fixture into a test module to activate it there.
|
||||
"""
|
||||
|
||||
class _UrlopenDelegatingOpener:
|
||||
def open(self, req, data=None, timeout=None):
|
||||
if data is None:
|
||||
return urllib.request.urlopen(req, timeout=timeout)
|
||||
return urllib.request.urlopen(req, data=data, timeout=timeout)
|
||||
|
||||
monkeypatch.setattr(
|
||||
urllib.request,
|
||||
"build_opener",
|
||||
lambda *handlers: _UrlopenDelegatingOpener(),
|
||||
)
|
||||
|
||||
@@ -6,6 +6,8 @@ import os
|
||||
import pytest
|
||||
import yaml
|
||||
|
||||
from tests.http_helpers import route_opener_open_through_urlopen # noqa: F401
|
||||
|
||||
from specify_cli.integrations.catalog import (
|
||||
IntegrationCatalog,
|
||||
IntegrationCatalogEntry,
|
||||
|
||||
@@ -1,5 +1,7 @@
|
||||
"""Tests for LingmaIntegration."""
|
||||
|
||||
from specify_cli.integrations import get_integration
|
||||
|
||||
from .test_integration_base_skills import SkillsIntegrationTests
|
||||
|
||||
|
||||
@@ -8,3 +10,9 @@ class TestLingmaIntegration(SkillsIntegrationTests):
|
||||
FOLDER = ".lingma/"
|
||||
COMMANDS_SUBDIR = "skills"
|
||||
REGISTRAR_DIR = ".lingma/skills"
|
||||
|
||||
def test_multi_install_safe(self):
|
||||
# Lingma writes only to its isolated, static root .lingma/skills,
|
||||
# disjoint from every other integration, so it must be co-install safe
|
||||
# (mirrors trae/zcode and the kiro-cli #3471 precedent).
|
||||
assert get_integration(self.KEY).multi_install_safe is True
|
||||
|
||||
@@ -84,3 +84,31 @@ def test_write_integration_json_strips_integration_key(tmp_path):
|
||||
assert state["integration"] == "claude"
|
||||
assert state["default_integration"] == "claude"
|
||||
assert state["installed_integrations"] == ["claude"]
|
||||
|
||||
|
||||
def test_with_integration_setting_recomputes_separator_from_retained_options():
|
||||
"""Updating only script_type must not drop an options-dependent separator.
|
||||
|
||||
Copilot resolves the command-ref separator to '-' when '--skills' options
|
||||
are stored and '.' otherwise. A second call that changes only script_type
|
||||
(parsed_options=None, raw_options=None) retains the stored parsed_options,
|
||||
so invoke_separator must stay '-', not be recomputed from the None argument.
|
||||
"""
|
||||
from specify_cli.integrations import get_integration
|
||||
from specify_cli.integration_runtime import with_integration_setting
|
||||
|
||||
copilot = get_integration("copilot")
|
||||
|
||||
settings = with_integration_setting(
|
||||
{}, "copilot", copilot, parsed_options={"skills": True}
|
||||
)
|
||||
assert settings["copilot"]["invoke_separator"] == "-"
|
||||
|
||||
settings2 = with_integration_setting(
|
||||
{"integration_settings": settings}, "copilot", copilot, script_type="ps"
|
||||
)
|
||||
# parsed_options are retained (only script_type changed) ...
|
||||
assert settings2["copilot"]["parsed_options"] == {"skills": True}
|
||||
assert settings2["copilot"]["script"] == "ps"
|
||||
# ... so the separator must reflect them, not the (None) argument.
|
||||
assert settings2["copilot"]["invoke_separator"] == "-"
|
||||
|
||||
@@ -18,7 +18,7 @@ from specify_cli._version import (
|
||||
_verify_upgrade,
|
||||
)
|
||||
from tests.conftest import strip_ansi
|
||||
from tests.http_helpers import mock_urlopen_response
|
||||
from tests.http_helpers import mock_urlopen_response, route_opener_open_through_urlopen
|
||||
|
||||
__all__ = (
|
||||
"SENTINEL_GH_TOKEN",
|
||||
@@ -31,6 +31,7 @@ __all__ = (
|
||||
"_verify_upgrade",
|
||||
"mock_urlopen_response",
|
||||
"requires_posix",
|
||||
"route_opener_open_through_urlopen",
|
||||
"runner",
|
||||
"strip_ansi",
|
||||
)
|
||||
|
||||
@@ -14,6 +14,7 @@ Covers:
|
||||
from __future__ import annotations
|
||||
|
||||
import base64
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
|
||||
@@ -515,6 +516,23 @@ class TestAzureDevOpsAuth:
|
||||
with patch("specify_cli.authentication.azure_devops.subprocess.run", side_effect=boom):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
@pytest.mark.parametrize("payload", [[], {"accessToken": None}, {"accessToken": 123}])
|
||||
def test_resolve_token_azure_cli_unexpected_json_shape_returns_none(
|
||||
self, payload
|
||||
):
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
entry = AuthConfigEntry(
|
||||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-cli",
|
||||
)
|
||||
result = MagicMock(returncode=0, stdout=json.dumps(payload))
|
||||
|
||||
with patch(
|
||||
"specify_cli.authentication.azure_devops.subprocess.run",
|
||||
return_value=result,
|
||||
):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
def test_resolve_token_azure_ad_success(self, monkeypatch):
|
||||
"""azure-ad acquires token via OAuth2 client credentials."""
|
||||
from unittest.mock import patch, MagicMock
|
||||
@@ -524,10 +542,15 @@ class TestAzureDevOpsAuth:
|
||||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||||
)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.read.return_value = b'{"access_token": "ad-acquired-token"}'
|
||||
mock_resp.read.side_effect = io.BytesIO(b'{"access_token": "ad-acquired-token"}').read
|
||||
mock_resp.__enter__ = lambda s: s
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
with patch("urllib.request.urlopen", return_value=mock_resp):
|
||||
# The token request goes through a strict-redirect opener (so a 307/308
|
||||
# cannot forward the client_secret body to a non-HTTPS host), not bare
|
||||
# urlopen; patch the opener it builds.
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.return_value = mock_resp
|
||||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) == "ad-acquired-token"
|
||||
|
||||
def test_resolve_token_azure_ad_missing_secret_returns_none(self, monkeypatch):
|
||||
@@ -542,14 +565,123 @@ class TestAzureDevOpsAuth:
|
||||
def test_resolve_token_azure_ad_network_error_returns_none(self, monkeypatch):
|
||||
"""azure-ad returns None on network errors."""
|
||||
import urllib.error
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||||
entry = AuthConfigEntry(
|
||||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||||
)
|
||||
with patch("urllib.request.urlopen",
|
||||
side_effect=urllib.error.URLError("connection refused")):
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = urllib.error.URLError("connection refused")
|
||||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("status", "reason"),
|
||||
[(307, "Temporary Redirect"), (308, "Permanent Redirect")],
|
||||
)
|
||||
def test_resolve_token_azure_ad_rejects_https_redirect(
|
||||
self, monkeypatch, status, reason
|
||||
):
|
||||
"""The client-secret POST must never be redirected to another host."""
|
||||
import urllib.error
|
||||
from unittest.mock import MagicMock, patch
|
||||
from urllib.request import Request
|
||||
|
||||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||||
entry = AuthConfigEntry(
|
||||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||||
)
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = urllib.error.URLError("stop after setup")
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=mock_opener) as build_opener:
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
redirect_handler = build_opener.call_args.args[0]
|
||||
request = Request(
|
||||
"https://login.microsoftonline.com/tid/oauth2/v2.0/token",
|
||||
data=b"grant_type=client_credentials&client_secret=secret-value",
|
||||
headers={"Content-Type": "application/x-www-form-urlencoded"},
|
||||
)
|
||||
assert request.get_method() == "POST"
|
||||
assert b"client_secret=secret-value" in request.data
|
||||
with pytest.raises(urllib.error.URLError, match="must not be redirected"):
|
||||
redirect_handler.redirect_request(
|
||||
request,
|
||||
io.BytesIO(b""),
|
||||
status,
|
||||
reason,
|
||||
{},
|
||||
"https://evil.example/token",
|
||||
)
|
||||
|
||||
def test_resolve_token_azure_ad_oversized_response_returns_none(
|
||||
self, monkeypatch
|
||||
):
|
||||
"""Oversized token metadata is rejected before JSON parsing."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
from specify_cli._download_security import MAX_JSON_METADATA_BYTES
|
||||
|
||||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||||
entry = AuthConfigEntry(
|
||||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||||
)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.read.side_effect = io.BytesIO(
|
||||
b"x" * (MAX_JSON_METADATA_BYTES + 1)
|
||||
).read
|
||||
mock_resp.__enter__ = lambda s: s
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.return_value = mock_resp
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=mock_opener), patch(
|
||||
"specify_cli.authentication.azure_devops._json.loads",
|
||||
side_effect=AssertionError("oversized body must not be parsed"),
|
||||
):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
@pytest.mark.parametrize("payload", [[], {"access_token": None}, {"access_token": 123}])
|
||||
def test_resolve_token_azure_ad_unexpected_json_shape_returns_none(
|
||||
self, monkeypatch, payload
|
||||
):
|
||||
from unittest.mock import MagicMock, patch
|
||||
|
||||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||||
entry = AuthConfigEntry(
|
||||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||||
)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.read.side_effect = io.BytesIO(json.dumps(payload).encode()).read
|
||||
mock_resp.__enter__ = lambda s: s
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.return_value = mock_resp
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
def test_resolve_token_azure_ad_invalid_utf8_returns_none(self, monkeypatch):
|
||||
"""azure-ad returns None when the token response is not valid UTF-8."""
|
||||
from unittest.mock import MagicMock, patch
|
||||
monkeypatch.setenv("MY_SECRET", "secret-value")
|
||||
entry = AuthConfigEntry(
|
||||
hosts=("dev.azure.com",), provider="azure-devops", auth="azure-ad",
|
||||
tenant_id="tid", client_id="cid", client_secret_env="MY_SECRET",
|
||||
)
|
||||
mock_resp = MagicMock()
|
||||
mock_resp.read.side_effect = io.BytesIO(b"\xff").read
|
||||
mock_resp.__enter__ = lambda s: s
|
||||
mock_resp.__exit__ = MagicMock(return_value=False)
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.return_value = mock_resp
|
||||
|
||||
with patch("urllib.request.build_opener", return_value=mock_opener):
|
||||
assert AzureDevOpsAuth().resolve_token(entry) is None
|
||||
|
||||
|
||||
@@ -615,13 +747,15 @@ class TestAuthenticatedHttp:
|
||||
monkeypatch.setenv("GH_TOKEN", "my-token")
|
||||
self._set_config(monkeypatch, [_github_entry()])
|
||||
captured = {}
|
||||
def fake_urlopen(req, timeout=None):
|
||||
def fake_open(req, timeout=None):
|
||||
captured["req"] = req
|
||||
resp = MagicMock()
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = fake_open
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
open_url("https://example.com/file.json")
|
||||
assert captured["req"].get_header("Authorization") is None
|
||||
|
||||
@@ -630,13 +764,15 @@ class TestAuthenticatedHttp:
|
||||
from specify_cli.authentication.http import open_url
|
||||
self._set_config(monkeypatch, [])
|
||||
captured = {}
|
||||
def fake_urlopen(req, timeout=None):
|
||||
def fake_open(req, timeout=None):
|
||||
captured["req"] = req
|
||||
resp = MagicMock()
|
||||
resp.__enter__ = lambda s: s
|
||||
resp.__exit__ = MagicMock(return_value=False)
|
||||
return resp
|
||||
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=fake_urlopen):
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = fake_open
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
open_url("https://github.com/org/repo")
|
||||
assert captured["req"].get_header("Authorization") is None
|
||||
|
||||
@@ -658,8 +794,7 @@ class TestAuthenticatedHttp:
|
||||
return resp
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = fake_side_effect
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener), \
|
||||
patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=fake_side_effect):
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
open_url("https://github.com/org/repo")
|
||||
assert call_count == 2
|
||||
|
||||
@@ -700,21 +835,23 @@ class TestAuthenticatedHttpNegative:
|
||||
|
||||
def test_urlerror_propagates(self, monkeypatch):
|
||||
import urllib.error
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
from specify_cli.authentication.http import open_url
|
||||
self._set_config(monkeypatch, [])
|
||||
with patch("specify_cli.authentication.http.urllib.request.urlopen",
|
||||
side_effect=urllib.error.URLError("refused")):
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = urllib.error.URLError("refused")
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
with pytest.raises(urllib.error.URLError):
|
||||
open_url("https://example.com/file")
|
||||
|
||||
def test_timeout_propagates(self, monkeypatch):
|
||||
import socket
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
from specify_cli.authentication.http import open_url
|
||||
self._set_config(monkeypatch, [])
|
||||
with patch("specify_cli.authentication.http.urllib.request.urlopen",
|
||||
side_effect=socket.timeout("timed out")):
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = socket.timeout("timed out")
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
with pytest.raises(socket.timeout):
|
||||
open_url("https://example.com/file")
|
||||
|
||||
@@ -820,17 +957,18 @@ class TestRedirectStripping:
|
||||
assert new_req.headers.get("Authorization") is None
|
||||
assert new_req.unredirected_hdrs.get("Authorization") is None
|
||||
|
||||
def test_https_to_http_same_host_redirect_strips_auth(self):
|
||||
def test_https_to_http_same_host_redirect_rejected(self):
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
from urllib.request import Request
|
||||
import io
|
||||
import urllib.error
|
||||
|
||||
handler = _StripAuthOnRedirect(("github.com",))
|
||||
req = Request("https://github.com/org/repo", headers={"Authorization": "Bearer tok"})
|
||||
new_req = handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||||
"http://github.com/org/repo")
|
||||
assert new_req is not None
|
||||
assert new_req.headers.get("Authorization") is None
|
||||
assert new_req.unredirected_hdrs.get("Authorization") is None
|
||||
|
||||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||||
"http://github.com/org/repo")
|
||||
|
||||
def test_redirect_validator_can_reject_before_following_redirect(self):
|
||||
import urllib.error
|
||||
@@ -888,6 +1026,177 @@ class TestRedirectStripping:
|
||||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||||
"https://[::1/asset")
|
||||
|
||||
def test_redirect_rejects_https_downgrade(self):
|
||||
"""HTTPS downloads must not follow redirects to non-local HTTP URLs."""
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
from urllib.request import Request
|
||||
import io
|
||||
import urllib.error
|
||||
handler = _StripAuthOnRedirect(("example.com",))
|
||||
req = Request("https://example.com/archive.zip")
|
||||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||||
"http://evil.example.com/archive.zip")
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target",
|
||||
[
|
||||
"http://127.0.0.1/internal",
|
||||
"https://localhost/internal",
|
||||
"https://localhost./internal",
|
||||
"https://service.localhost/internal",
|
||||
"https://service.localhost./internal",
|
||||
"https://127.0.0.2/internal",
|
||||
"https://127.1/internal",
|
||||
"https://2130706433/internal",
|
||||
"https://0x7f000001/internal",
|
||||
"https://017700000001/internal",
|
||||
"https://0177.0.0.1/internal",
|
||||
"https://[::1]/internal",
|
||||
"https://[::1%25lo0]/internal",
|
||||
"https://[::ffff:127.0.0.1]/internal",
|
||||
"https://127%2e0%2e0%2e1/internal",
|
||||
"https://%31%32%37.0.0.1/internal",
|
||||
"https://127%2E1/internal",
|
||||
"https://local%68ost/internal",
|
||||
"https://[::ffff:127%2e0.0.1]/internal",
|
||||
"https://[::ffff:7f00%3a1]/internal",
|
||||
"https://[::ffff%3a127.0.0.1]/internal",
|
||||
"https://ℓocalhost/internal",
|
||||
"https://localhost/internal",
|
||||
"https://127。0。0。1/internal",
|
||||
"https://0.0.0.0/internal",
|
||||
"https://0/internal",
|
||||
"https://00.00.00.00/internal",
|
||||
"https://[::]/internal",
|
||||
"https://[::ffff:0.0.0.0]/internal",
|
||||
],
|
||||
)
|
||||
def test_redirect_rejects_remote_to_loopback(self, target):
|
||||
"""A remote response must not redirect a download into loopback."""
|
||||
import io
|
||||
import urllib.error
|
||||
from urllib.request import Request
|
||||
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
|
||||
handler = _StripAuthOnRedirect(())
|
||||
request = Request("https://example.com/archive.zip")
|
||||
|
||||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||||
handler.redirect_request(
|
||||
request,
|
||||
io.BytesIO(b""),
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
target,
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
("source", "target"),
|
||||
[
|
||||
(
|
||||
"http://localhost:8000/archive.zip",
|
||||
"http://127.0.0.1:8001/archive.zip",
|
||||
),
|
||||
(
|
||||
"http://127.0.0.2:8000/archive.zip",
|
||||
"http://127.255.255.254:8001/archive.zip",
|
||||
),
|
||||
(
|
||||
"https://[0:0:0:0:0:0:0:1]/archive.zip",
|
||||
"http://[::1]:8001/archive.zip",
|
||||
),
|
||||
],
|
||||
)
|
||||
def test_redirect_allows_loopback_to_http_loopback(self, source, target):
|
||||
"""Local development may continue redirecting between loopback URLs."""
|
||||
import io
|
||||
from urllib.request import Request
|
||||
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
|
||||
handler = _StripAuthOnRedirect(())
|
||||
request = Request(source)
|
||||
redirected = handler.redirect_request(
|
||||
request,
|
||||
io.BytesIO(b""),
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
target,
|
||||
)
|
||||
|
||||
assert redirected is not None
|
||||
|
||||
def test_multi_hop_remote_to_loopback_chain_is_rejected_at_first_hop(self):
|
||||
import io
|
||||
import urllib.error
|
||||
from urllib.request import Request
|
||||
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
|
||||
handler = _StripAuthOnRedirect(())
|
||||
request = Request("https://example.com/archive.zip")
|
||||
|
||||
with pytest.raises(urllib.error.URLError, match="unsafe redirect"):
|
||||
handler.redirect_request(
|
||||
request,
|
||||
io.BytesIO(b""),
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
"https://localhost:4443/hop",
|
||||
)
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"target",
|
||||
[
|
||||
"https://example.com:notaport/archive.zip",
|
||||
"https://example.com:+443/archive.zip",
|
||||
"https://example.com:65536/archive.zip",
|
||||
],
|
||||
)
|
||||
def test_malformed_redirect_port_raises_urlerror(self, target):
|
||||
import io
|
||||
import urllib.error
|
||||
from urllib.request import Request
|
||||
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
|
||||
handler = _StripAuthOnRedirect(())
|
||||
request = Request("https://example.com/archive.zip")
|
||||
|
||||
with pytest.raises(urllib.error.URLError, match="malformed redirect URL"):
|
||||
handler.redirect_request(
|
||||
request,
|
||||
io.BytesIO(b""),
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
target,
|
||||
)
|
||||
|
||||
def test_strict_redirect_error_describes_target_and_allowed_localhost(self):
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
from urllib.request import Request
|
||||
import io
|
||||
import urllib.error
|
||||
|
||||
handler = _StripAuthOnRedirect(("example.com",))
|
||||
req = Request("https://example.com/archive.zip")
|
||||
|
||||
with pytest.raises(urllib.error.URLError) as exc_info:
|
||||
handler.redirect_request(req, io.BytesIO(b""), 302, "Found", {},
|
||||
"http://evil.example.com/archive.zip")
|
||||
|
||||
error_message = str(exc_info.value)
|
||||
assert "http://evil.example.com/archive.zip" in error_message
|
||||
assert "localhost" in error_message
|
||||
assert "127.0.0.1" in error_message
|
||||
assert "::1" in error_message
|
||||
|
||||
|
||||
# ---------------------------------------------------------------------------
|
||||
# _fetch_latest_release_tag delegation
|
||||
@@ -907,7 +1216,7 @@ class TestFetchLatestReleaseTagDelegation:
|
||||
captured["request"] = req
|
||||
body = _json.dumps({"tag_name": "v9.9.9"}).encode()
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = body
|
||||
resp.read.side_effect = io.BytesIO(body).read
|
||||
cm = MagicMock()
|
||||
cm.__enter__.return_value = resp
|
||||
cm.__exit__.return_value = False
|
||||
@@ -927,20 +1236,25 @@ class TestFetchLatestReleaseTagDelegation:
|
||||
assert captured["request"].get_header("Authorization") == "Bearer forwarded-sentinel"
|
||||
|
||||
def test_no_config_means_no_auth(self, monkeypatch):
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
from specify_cli._version import _fetch_latest_release_tag
|
||||
self._set_config(monkeypatch, [])
|
||||
captured, side_effect = self._capture_request()
|
||||
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect):
|
||||
# The unauthenticated path uses the strict redirect opener too.
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = side_effect
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
_fetch_latest_release_tag()
|
||||
assert captured["request"].get_header("Authorization") is None
|
||||
|
||||
def test_accept_header_present(self, monkeypatch):
|
||||
from unittest.mock import patch
|
||||
from unittest.mock import MagicMock, patch
|
||||
from specify_cli._version import _fetch_latest_release_tag
|
||||
self._set_config(monkeypatch, [])
|
||||
captured, side_effect = self._capture_request()
|
||||
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect):
|
||||
mock_opener = MagicMock()
|
||||
mock_opener.open.side_effect = side_effect
|
||||
with patch("specify_cli.authentication.http.urllib.request.build_opener", return_value=mock_opener):
|
||||
_fetch_latest_release_tag()
|
||||
assert captured["request"].get_header("Accept") == "application/vnd.github+json"
|
||||
|
||||
|
||||
227
tests/test_download_security.py
Normal file
227
tests/test_download_security.py
Normal file
@@ -0,0 +1,227 @@
|
||||
"""Tests for bounded HTTP download helpers."""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import weakref
|
||||
|
||||
import pytest
|
||||
|
||||
from specify_cli._download_security import (
|
||||
is_https_or_localhost_http,
|
||||
is_loopback_url,
|
||||
read_response_limited,
|
||||
)
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url, allowed",
|
||||
[
|
||||
("https://example.com/preset.zip", True),
|
||||
("http://localhost:8000/preset.zip", True),
|
||||
("http://127.0.0.1/preset.zip", True),
|
||||
("http://127.0.0.2/preset.zip", True),
|
||||
("http://127.255.255.254/preset.zip", True),
|
||||
("http://[::1]/preset.zip", True),
|
||||
("http://[0:0:0:0:0:0:0:1]/preset.zip", True),
|
||||
("http://[::ffff:127.0.0.2]/preset.zip", True),
|
||||
("http://[::1%25lo0]/preset.zip", True),
|
||||
# Non-loopback HTTP is rejected.
|
||||
("http://example.com/preset.zip", False),
|
||||
("http://192.0.2.1/preset.zip", False),
|
||||
("http://[fe80::1]/preset.zip", False),
|
||||
("http://[fe80::1%25lo0]/preset.zip", False),
|
||||
("http://0.0.0.0/preset.zip", False),
|
||||
("http://0/preset.zip", False),
|
||||
("http://[::]/preset.zip", False),
|
||||
("http://[::ffff:0.0.0.0]/preset.zip", False),
|
||||
# Ambiguous/platform-dependent spellings may never authorize HTTP.
|
||||
("http://127.1/preset.zip", False),
|
||||
("http://2130706433/preset.zip", False),
|
||||
("http://0x7f000001/preset.zip", False),
|
||||
("http://017700000001/preset.zip", False),
|
||||
("http://0177.0.0.1/preset.zip", False),
|
||||
("http://00177.0.0.1/preset.zip", False),
|
||||
("http://localhost./preset.zip", False),
|
||||
("http://ℓocalhost/preset.zip", False),
|
||||
("http://127。0。0。1/preset.zip", False),
|
||||
# A hostname is always required, even for HTTPS.
|
||||
("https:///preset.zip", False),
|
||||
("https://", False),
|
||||
# Invalid ports must be rejected before urllib opens the URL.
|
||||
("https://example.com:notaport/preset.zip", False),
|
||||
("https://example.com:+443/preset.zip", False),
|
||||
("https://example.com:65536/preset.zip", False),
|
||||
# urllib decodes escapes in the authority before connecting; reject
|
||||
# encoded reg-names so validation and connection cannot disagree.
|
||||
("https://127%2e0%2e0%2e1/preset.zip", False),
|
||||
("https://%31%32%37.0.0.1/preset.zip", False),
|
||||
("https://local%68ost/preset.zip", False),
|
||||
("https://example.com%3a443/preset.zip", False),
|
||||
("https://[::1%lo0]/preset.zip", False),
|
||||
("https://[::ffff:127%2e0.0.1]/preset.zip", False),
|
||||
("https://[::ffff:7f00%3a1]/preset.zip", False),
|
||||
("https://[::ffff%3a127.0.0.1]/preset.zip", False),
|
||||
],
|
||||
)
|
||||
def test_is_https_or_localhost_http(url, allowed):
|
||||
assert is_https_or_localhost_http(url) is allowed
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"https://localhost/internal",
|
||||
"https://127.0.0.2/internal",
|
||||
"https://[::1]/internal",
|
||||
"https://[::1%25lo0]/internal",
|
||||
"https://[::ffff:127.0.0.2]/internal",
|
||||
],
|
||||
)
|
||||
def test_is_loopback_url_recognizes_effective_loopback_literals(url):
|
||||
assert is_loopback_url(url) is True
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"url",
|
||||
[
|
||||
"https://localhost./internal",
|
||||
"https://service.localhost/internal",
|
||||
"https://service.localhost./internal",
|
||||
"https://127.1/internal",
|
||||
"https://2130706433/internal",
|
||||
"https://0x7f000001/internal",
|
||||
"https://017700000001/internal",
|
||||
"https://0177.0.0.1/internal",
|
||||
"https://ℓocalhost/internal",
|
||||
"https://127。0。0。1/internal",
|
||||
"https://127%2e0%2e0%2e1/internal",
|
||||
"https://0.0.0.0/internal",
|
||||
"https://0/internal",
|
||||
"https://00.00.00.00/internal",
|
||||
"https://[::]/internal",
|
||||
"https://[::ffff:0.0.0.0]/internal",
|
||||
],
|
||||
)
|
||||
def test_is_loopback_url_does_not_authorize_ambiguous_spellings(url):
|
||||
assert is_loopback_url(url) is False
|
||||
|
||||
|
||||
class _Response:
|
||||
"""Faithful stream stand-in: read() advances a cursor and returns b"" at EOF."""
|
||||
|
||||
def __init__(self, data: bytes, *, chunk: int | None = None):
|
||||
self.data = data
|
||||
self.pos = 0
|
||||
# When set, never return more than *chunk* bytes per call even if more is
|
||||
# requested - simulates short reads (e.g. chunked transfer encoding).
|
||||
self.chunk = chunk
|
||||
|
||||
def read(self, size: int = -1) -> bytes:
|
||||
if size < 0:
|
||||
size = len(self.data) - self.pos
|
||||
if self.chunk is not None:
|
||||
size = min(size, self.chunk)
|
||||
out = self.data[self.pos : self.pos + size]
|
||||
self.pos += len(out)
|
||||
return out
|
||||
|
||||
|
||||
class _RecordingResponse(_Response):
|
||||
def __init__(self, data: bytes, *, chunk: int | None = None):
|
||||
super().__init__(data, chunk=chunk)
|
||||
self.requested_sizes: list[int] = []
|
||||
|
||||
def read(self, size: int = -1) -> bytes:
|
||||
self.requested_sizes.append(size)
|
||||
return super().read(size)
|
||||
|
||||
|
||||
class _TrackedChunk(bytearray):
|
||||
pass
|
||||
|
||||
|
||||
class _OneByteResponse:
|
||||
"""Return distinct weak-referenceable chunks to detect retained fragments."""
|
||||
|
||||
def __init__(self, count: int):
|
||||
self.remaining = count
|
||||
self.refs: list[weakref.ReferenceType[_TrackedChunk]] = []
|
||||
self.peak_live = 0
|
||||
|
||||
def read(self, _size: int = -1) -> bytes | _TrackedChunk:
|
||||
if self.remaining == 0:
|
||||
return b""
|
||||
self.remaining -= 1
|
||||
chunk = _TrackedChunk(b"x")
|
||||
self.refs.append(weakref.ref(chunk))
|
||||
self.peak_live = max(
|
||||
self.peak_live,
|
||||
sum(ref() is not None for ref in self.refs),
|
||||
)
|
||||
return chunk
|
||||
|
||||
|
||||
def test_read_response_limited_rejects_oversized_download():
|
||||
with pytest.raises(ValueError, match="exceeds maximum size"):
|
||||
read_response_limited(_Response(b"abcde"), max_bytes=4)
|
||||
|
||||
|
||||
def test_read_response_limited_returns_full_body_within_limit():
|
||||
assert read_response_limited(_Response(b"abcde"), max_bytes=10) == b"abcde"
|
||||
|
||||
|
||||
def test_read_response_limited_enforces_bound_under_short_reads():
|
||||
# A server that streams more than max_bytes total while every read() returns
|
||||
# fewer bytes than requested (chunked encoding) must still be rejected - a
|
||||
# single read(max_bytes + 1) could be fooled, the accumulating loop cannot.
|
||||
response = _Response(b"x" * 100, chunk=8)
|
||||
with pytest.raises(ValueError, match="exceeds maximum size"):
|
||||
read_response_limited(response, max_bytes=16)
|
||||
|
||||
|
||||
def test_read_response_limited_does_not_retain_short_read_fragments():
|
||||
response = _OneByteResponse(64)
|
||||
|
||||
assert read_response_limited(response, max_bytes=64) == b"x" * 64
|
||||
assert response.peak_live <= 2
|
||||
|
||||
|
||||
def test_read_response_limited_caps_underlying_reads_at_64_kib():
|
||||
response = _RecordingResponse(b"x" * (64 * 1024 + 1))
|
||||
|
||||
with pytest.raises(ValueError, match="exceeds maximum size"):
|
||||
read_response_limited(response, max_bytes=64 * 1024)
|
||||
|
||||
assert max(response.requested_sizes) <= 64 * 1024
|
||||
|
||||
|
||||
@pytest.mark.parametrize("value", [None, "1", 1.5, True])
|
||||
def test_read_response_limited_rejects_non_integer_limits(value):
|
||||
with pytest.raises(TypeError, match="integer"):
|
||||
read_response_limited(_Response(b""), max_bytes=value)
|
||||
|
||||
|
||||
def test_read_response_limited_rejects_negative_limit_without_reading():
|
||||
response = _RecordingResponse(b"")
|
||||
|
||||
with pytest.raises(ValueError, match="non-negative"):
|
||||
read_response_limited(response, max_bytes=-1)
|
||||
|
||||
assert response.requested_sizes == []
|
||||
|
||||
|
||||
def test_read_response_limited_allows_empty_response_at_zero_limit():
|
||||
assert read_response_limited(_Response(b""), max_bytes=0) == b""
|
||||
|
||||
|
||||
class _CustomLimitError(Exception):
|
||||
pass
|
||||
|
||||
|
||||
def test_read_response_limited_rejects_first_byte_at_zero_limit():
|
||||
with pytest.raises(_CustomLimitError, match="exceeds maximum size"):
|
||||
read_response_limited(
|
||||
_Response(b"x"),
|
||||
max_bytes=0,
|
||||
error_type=_CustomLimitError,
|
||||
)
|
||||
@@ -9,6 +9,7 @@ Tests cover:
|
||||
- Catalog stack (multi-catalog support)
|
||||
"""
|
||||
|
||||
import io
|
||||
import pytest
|
||||
import json
|
||||
import os
|
||||
@@ -22,6 +23,7 @@ from datetime import datetime, timezone
|
||||
from unittest.mock import MagicMock
|
||||
|
||||
from tests.conftest import strip_ansi
|
||||
from tests.http_helpers import route_opener_open_through_urlopen # noqa: F401
|
||||
from specify_cli import extensions as _ext_module
|
||||
from specify_cli.extensions import (
|
||||
CatalogEntry,
|
||||
@@ -4978,7 +4980,7 @@ class TestExtensionCatalog:
|
||||
zip_bytes = zip_buf.getvalue()
|
||||
|
||||
release_response = MagicMock()
|
||||
release_response.read.return_value = json.dumps(
|
||||
release_response.read.side_effect = io.BytesIO(json.dumps(
|
||||
{
|
||||
"assets": [
|
||||
{
|
||||
@@ -4987,12 +4989,12 @@ class TestExtensionCatalog:
|
||||
}
|
||||
]
|
||||
}
|
||||
).encode()
|
||||
).encode()).read
|
||||
release_response.__enter__ = lambda s: s
|
||||
release_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
asset_response = MagicMock()
|
||||
asset_response.read.return_value = zip_bytes
|
||||
asset_response.read.side_effect = io.BytesIO(zip_bytes).read
|
||||
asset_response.__enter__ = lambda s: s
|
||||
asset_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
@@ -8761,10 +8763,10 @@ def test_extension_wrapper_resolves_ghes_asset_when_host_configured(tmp_path, mo
|
||||
def fake_open(url, timeout=None, extra_headers=None):
|
||||
captured.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps({
|
||||
resp.read.side_effect = io.BytesIO(json.dumps({
|
||||
"assets": [{"name": "ext.zip",
|
||||
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/7"}]
|
||||
}).encode()
|
||||
}).encode()).read
|
||||
yield resp
|
||||
|
||||
monkeypatch.setattr(catalog, "_open_url", fake_open)
|
||||
|
||||
@@ -1,16 +1,20 @@
|
||||
"""Tests for GitHub-authenticated HTTP request helpers."""
|
||||
|
||||
import io
|
||||
import json
|
||||
import os
|
||||
from contextlib import contextmanager
|
||||
from unittest.mock import MagicMock, patch
|
||||
from urllib.request import Request
|
||||
|
||||
import pytest
|
||||
|
||||
from specify_cli._github_http import (
|
||||
GITHUB_HOSTS,
|
||||
build_github_request,
|
||||
resolve_github_release_asset_api_url,
|
||||
)
|
||||
from specify_cli.authentication.http import _StripAuthOnRedirect
|
||||
|
||||
|
||||
class TestBuildGitHubRequest:
|
||||
@@ -90,7 +94,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
@contextmanager
|
||||
def fake_open(url, timeout=None, extra_headers=None):
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps(release_json).encode()
|
||||
resp.read.side_effect = io.BytesIO(json.dumps(release_json).encode()).read
|
||||
yield resp
|
||||
return fake_open
|
||||
|
||||
@@ -198,7 +202,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
def capturing_open(url, timeout=None, extra_headers=None):
|
||||
captured_urls.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps({"assets": []}).encode()
|
||||
resp.read.side_effect = io.BytesIO(json.dumps({"assets": []}).encode()).read
|
||||
yield resp
|
||||
|
||||
resolve_github_release_asset_api_url(
|
||||
@@ -217,7 +221,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
def capturing_open(url, timeout=None, extra_headers=None):
|
||||
captured_urls.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps({"assets": []}).encode()
|
||||
resp.read.side_effect = io.BytesIO(json.dumps({"assets": []}).encode()).read
|
||||
yield resp
|
||||
|
||||
resolve_github_release_asset_api_url(
|
||||
@@ -260,7 +264,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
def recording_open(url, timeout=None, extra_headers=None):
|
||||
called.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = b"{}"
|
||||
resp.read.side_effect = io.BytesIO(b"{}").read
|
||||
yield resp
|
||||
|
||||
result = resolve_github_release_asset_api_url(
|
||||
@@ -299,7 +303,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
def recording_open(url, timeout=None, extra_headers=None):
|
||||
called.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = b"{}"
|
||||
resp.read.side_effect = io.BytesIO(b"{}").read
|
||||
yield resp
|
||||
|
||||
url = "https://ghes.example/api/v3/repos/o/r/releases/assets/7"
|
||||
@@ -317,7 +321,7 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
def capturing_open(url, timeout=None, extra_headers=None):
|
||||
captured.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps({"assets": []}).encode()
|
||||
resp.read.side_effect = io.BytesIO(json.dumps({"assets": []}).encode()).read
|
||||
yield resp
|
||||
|
||||
resolve_github_release_asset_api_url(
|
||||
@@ -344,10 +348,10 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
def capturing_open(url, timeout=None, extra_headers=None):
|
||||
captured.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps({
|
||||
resp.read.side_effect = io.BytesIO(json.dumps({
|
||||
"assets": [{"name": "pack.zip",
|
||||
"url": "https://api.github.com/repos/org/repo/releases/assets/99"}]
|
||||
}).encode()
|
||||
}).encode()).read
|
||||
yield resp
|
||||
|
||||
result = resolve_github_release_asset_api_url(
|
||||
@@ -357,3 +361,43 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
||||
)
|
||||
assert result == "https://api.github.com/repos/org/repo/releases/assets/99"
|
||||
assert captured == ["https://api.github.com/repos/org/repo/releases/tags/v1.0"]
|
||||
|
||||
|
||||
class TestGitHubRedirectAuth:
|
||||
"""Tests for GitHub-owned redirect auth handling."""
|
||||
|
||||
def test_multi_hop_github_redirect_preserves_unredirected_auth(self):
|
||||
"""Auth survives a multi-hop redirect chain within GitHub hosts."""
|
||||
handler = _StripAuthOnRedirect(tuple(GITHUB_HOSTS))
|
||||
req1 = Request(
|
||||
"https://github.com/org/repo",
|
||||
headers={"Authorization": "Bearer tok"},
|
||||
)
|
||||
|
||||
req2 = handler.redirect_request(
|
||||
req1,
|
||||
io.BytesIO(b""),
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
"https://codeload.github.com/org/repo/zip",
|
||||
)
|
||||
assert req2 is not None
|
||||
auth2 = req2.get_header("Authorization") or req2.unredirected_hdrs.get(
|
||||
"Authorization"
|
||||
)
|
||||
assert auth2 == "Bearer tok"
|
||||
|
||||
req3 = handler.redirect_request(
|
||||
req2,
|
||||
io.BytesIO(b""),
|
||||
302,
|
||||
"Found",
|
||||
{},
|
||||
"https://raw.githubusercontent.com/org/repo/main/file",
|
||||
)
|
||||
assert req3 is not None
|
||||
auth3 = req3.get_header("Authorization") or req3.unredirected_hdrs.get(
|
||||
"Authorization"
|
||||
)
|
||||
assert auth3 == "Bearer tok"
|
||||
|
||||
@@ -243,3 +243,34 @@ class TestRegressionPlainTemplate:
|
||||
assert output_file.exists(), f"Output file missing for {agent}"
|
||||
content = output_file.read_text(encoding="utf-8")
|
||||
assert body_text.strip() in content, f"Body text missing in {agent} output"
|
||||
|
||||
|
||||
class TestClineRealPostProcess:
|
||||
"""Cline's real command-content transforms (hook-command note + handoff
|
||||
dot->hyphen rewrite) must run for extension/preset commands registered via
|
||||
CommandRegistrar. This exercises the REAL method (not a monkeypatched
|
||||
marker), so it fails if Cline's override does not match the base hook name
|
||||
the registrar dispatches to (post_process_command_content)."""
|
||||
|
||||
def test_cline_transforms_applied_via_registrar(
|
||||
self, tmp_path, registrar, ext_dir
|
||||
):
|
||||
ext, cmd_dir = ext_dir
|
||||
body = (
|
||||
"- For each executable hook, output the following:\n"
|
||||
"agent: speckit.foo\n"
|
||||
)
|
||||
_write_cmd(cmd_dir, body=body)
|
||||
|
||||
commands = [{"name": "speckit.test.review", "file": "commands/review.md"}]
|
||||
registrar.register_commands("cline", commands, "test-ext", ext, tmp_path)
|
||||
|
||||
outputs = list((tmp_path / ".clinerules" / "workflows").rglob("*.md"))
|
||||
assert outputs, "no cline command file was written"
|
||||
content = outputs[0].read_text(encoding="utf-8")
|
||||
|
||||
# _inject_hook_command_note fired (its note text contains "replace dots")
|
||||
assert "replace dots" in content
|
||||
# _rewrite_handoff_references rewrote the dotted agent handoff
|
||||
assert "agent: speckit-foo" in content
|
||||
assert "agent: speckit.foo" not in content
|
||||
|
||||
@@ -2303,7 +2303,7 @@ class TestPresetCatalog:
|
||||
zip_bytes = zip_buf.getvalue()
|
||||
|
||||
release_response = MagicMock()
|
||||
release_response.read.return_value = json.dumps(
|
||||
release_response.read.side_effect = io.BytesIO(json.dumps(
|
||||
{
|
||||
"assets": [
|
||||
{
|
||||
@@ -2312,12 +2312,12 @@ class TestPresetCatalog:
|
||||
}
|
||||
]
|
||||
}
|
||||
).encode()
|
||||
).encode()).read
|
||||
release_response.__enter__ = lambda s: s
|
||||
release_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
asset_response = MagicMock()
|
||||
asset_response.read.return_value = zip_bytes
|
||||
asset_response.read.side_effect = io.BytesIO(zip_bytes).read
|
||||
asset_response.__enter__ = lambda s: s
|
||||
asset_response.__exit__ = MagicMock(return_value=False)
|
||||
|
||||
@@ -5381,6 +5381,9 @@ class TestPresetEnableDisable:
|
||||
|
||||
|
||||
LEAN_PRESET_DIR = Path(__file__).parent.parent / "presets" / "lean"
|
||||
CORE_CONSTITUTION_COMMAND = (
|
||||
Path(__file__).parent.parent / "templates" / "commands" / "constitution.md"
|
||||
)
|
||||
|
||||
LEAN_COMMAND_NAMES = [
|
||||
"speckit.specify",
|
||||
@@ -5391,6 +5394,31 @@ LEAN_COMMAND_NAMES = [
|
||||
]
|
||||
|
||||
|
||||
@pytest.mark.parametrize(
|
||||
"command_path",
|
||||
[
|
||||
CORE_CONSTITUTION_COMMAND,
|
||||
LEAN_PRESET_DIR / "commands" / "speckit.constitution.md",
|
||||
],
|
||||
ids=["core", "lean"],
|
||||
)
|
||||
def test_constitution_commands_guard_against_non_governance_work(command_path):
|
||||
"""Constitution commands defer non-governance work instead of executing it."""
|
||||
content = command_path.read_text()
|
||||
lower_content = content.lower()
|
||||
normalized_content = " ".join(lower_content.split())
|
||||
|
||||
assert "## Scope Guard" in content
|
||||
assert "**MUST NOT**" in content
|
||||
assert "Classify every part" in content
|
||||
assert "application source files" in content
|
||||
assert "non-governance intent" in content
|
||||
assert "`Next Actions`" in content
|
||||
assert "__SPECKIT_COMMAND_SPECIFY__" in content
|
||||
assert "omit" in lower_content
|
||||
assert "do not invoke it" in normalized_content or "without invoking it" in normalized_content
|
||||
|
||||
|
||||
class TestLeanPreset:
|
||||
"""Tests for the lean preset that ships with the repo."""
|
||||
|
||||
@@ -7458,10 +7486,10 @@ def test_preset_wrapper_resolves_ghes_asset_when_host_configured(tmp_path, monke
|
||||
def fake_open(url, timeout=None, extra_headers=None):
|
||||
captured.append(url)
|
||||
resp = MagicMock()
|
||||
resp.read.return_value = json.dumps({
|
||||
resp.read.side_effect = io.BytesIO(json.dumps({
|
||||
"assets": [{"name": "pack.zip",
|
||||
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/9"}]
|
||||
}).encode()
|
||||
}).encode()).read
|
||||
yield resp
|
||||
|
||||
monkeypatch.setattr(catalog, "_open_url", fake_open)
|
||||
|
||||
@@ -13,6 +13,7 @@ import specify_cli
|
||||
from specify_cli import app
|
||||
|
||||
from tests.self_upgrade_helpers import (
|
||||
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
|
||||
_InstallMethod,
|
||||
_assemble_installer_argv,
|
||||
_completed_process,
|
||||
|
||||
@@ -7,6 +7,7 @@ from unittest.mock import patch
|
||||
from specify_cli import app
|
||||
|
||||
from tests.self_upgrade_helpers import (
|
||||
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
|
||||
_completed_process,
|
||||
mock_urlopen_response,
|
||||
requires_posix,
|
||||
|
||||
@@ -6,6 +6,7 @@ from specify_cli import app
|
||||
|
||||
from tests.self_upgrade_helpers import (
|
||||
mock_urlopen_response,
|
||||
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
|
||||
runner,
|
||||
strip_ansi,
|
||||
)
|
||||
|
||||
@@ -8,6 +8,7 @@ import specify_cli
|
||||
from specify_cli import app
|
||||
|
||||
from tests.self_upgrade_helpers import (
|
||||
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
|
||||
SENTINEL_GH_TOKEN,
|
||||
SENTINEL_GITHUB_TOKEN,
|
||||
_InstallMethod,
|
||||
|
||||
39
tests/test_specify_template_numbering.py
Normal file
39
tests/test_specify_template_numbering.py
Normal file
@@ -0,0 +1,39 @@
|
||||
"""Regression tests for top-level step numbering in specify.md."""
|
||||
|
||||
import re
|
||||
from pathlib import Path
|
||||
|
||||
REPO_ROOT = Path(__file__).parent.parent
|
||||
SPECIFY_TEMPLATE = REPO_ROOT / "templates" / "commands" / "specify.md"
|
||||
MAIN_LIST_START = "Given that feature description, do this:"
|
||||
MAIN_LIST_END = "## Mandatory Post-Execution Hooks"
|
||||
|
||||
|
||||
def _main_execution_ordinals(text: str) -> list[int]:
|
||||
"""Extract top-level ordinals from the main execution flow."""
|
||||
_, start, execution_flow = text.partition(MAIN_LIST_START)
|
||||
execution_flow, end, _ = execution_flow.partition(MAIN_LIST_END)
|
||||
if not start or not end:
|
||||
return []
|
||||
|
||||
return [
|
||||
int(match.group(1))
|
||||
for line in execution_flow.splitlines()
|
||||
if (match := re.match(r"^(\d+)\. ", line))
|
||||
]
|
||||
|
||||
|
||||
def test_main_execution_list_has_no_duplicate_ordinals():
|
||||
"""The main execution list must not reuse a step number."""
|
||||
ordinals = _main_execution_ordinals(SPECIFY_TEMPLATE.read_text(encoding="utf-8"))
|
||||
duplicates = {ordinal for ordinal in ordinals if ordinals.count(ordinal) > 1}
|
||||
|
||||
assert not duplicates, f"Duplicate top-level ordinals found: {sorted(duplicates)}"
|
||||
|
||||
|
||||
def test_main_execution_list_is_sequential():
|
||||
"""The main execution list must run from 1 through N without gaps."""
|
||||
ordinals = _main_execution_ordinals(SPECIFY_TEMPLATE.read_text(encoding="utf-8"))
|
||||
|
||||
assert ordinals, "Could not find the main execution list in specify.md"
|
||||
assert ordinals == list(range(1, 9))
|
||||
@@ -2,11 +2,12 @@
|
||||
|
||||
Network isolation contract (SC-004 / FR-014): every test that exercises
|
||||
`specify self check` or `_fetch_latest_release_tag()` MUST mock the outbound
|
||||
urllib path it expects (`urlopen` for unauthenticated requests, `build_opener`
|
||||
for authenticated requests) so no real outbound call ever reaches api.github.com.
|
||||
Tests for non-network `self upgrade` behavior should keep that contract explicit
|
||||
with local mocks. Run this module under `pytest-socket` (if installed) with
|
||||
`--disable-socket` as an extra safety net.
|
||||
urllib path so no real call reaches api.github.com. Production always uses an
|
||||
isolated `build_opener`; this module's autouse fixture routes its `open()` back
|
||||
through the locally mocked `urlopen`. Tests for non-network `self upgrade`
|
||||
behavior should keep that contract explicit with local mocks. Run this module
|
||||
under `pytest-socket` (if installed) with `--disable-socket` as an extra safety
|
||||
net.
|
||||
"""
|
||||
|
||||
import urllib.error
|
||||
@@ -17,6 +18,7 @@ import pytest
|
||||
from typer.testing import CliRunner
|
||||
|
||||
from specify_cli import app
|
||||
from specify_cli._download_security import read_response_limited as _real_read_response_limited
|
||||
from specify_cli._version import (
|
||||
_fetch_latest_release_tag,
|
||||
_get_installed_version,
|
||||
@@ -24,7 +26,10 @@ from specify_cli._version import (
|
||||
_normalize_tag,
|
||||
)
|
||||
from tests.conftest import strip_ansi
|
||||
from tests.http_helpers import mock_urlopen_response
|
||||
from tests.http_helpers import (
|
||||
mock_urlopen_response,
|
||||
route_opener_open_through_urlopen, # noqa: F401 (autouse fixture)
|
||||
)
|
||||
|
||||
runner = CliRunner()
|
||||
|
||||
@@ -235,6 +240,46 @@ class TestFailureCategorization:
|
||||
_fetch_latest_release_tag()
|
||||
|
||||
|
||||
class TestBoundedRead:
|
||||
"""Regression test for the read_response_limited hardening.
|
||||
|
||||
A future refactor could silently revert `_fetch_latest_release_tag` to
|
||||
`resp.read()` (the unbounded form) — this test pins the contract that
|
||||
the response body is read through ``read_response_limited`` with a
|
||||
bounded ``max_bytes``.
|
||||
"""
|
||||
|
||||
def test_response_body_is_bounded(self):
|
||||
recorded: dict[str, int | str] = {}
|
||||
|
||||
def _spy(response, *, max_bytes: int, label: str, **kwargs):
|
||||
# max_bytes and label are keyword-only with no defaults: if the
|
||||
# caller forgets to pass either, the call raises TypeError here
|
||||
# (instead of recording a misleading None).
|
||||
recorded["max_bytes"] = max_bytes
|
||||
recorded["label"] = label
|
||||
# Forward to the real implementation so the function under test
|
||||
# still gets a parseable body.
|
||||
return _real_read_response_limited(
|
||||
response, max_bytes=max_bytes, label=label, **kwargs
|
||||
)
|
||||
|
||||
with patch(
|
||||
"specify_cli.authentication.http.urllib.request.urlopen",
|
||||
return_value=mock_urlopen_response({"tag_name": "v9.9.9"}),
|
||||
), patch("specify_cli._version.read_response_limited", side_effect=_spy):
|
||||
tag, reason = _fetch_latest_release_tag()
|
||||
|
||||
assert tag == "v9.9.9"
|
||||
assert reason is None
|
||||
# The cap (1 MiB) is a deliberate ceiling for the GitHub release
|
||||
# JSON — keep it explicit so a future refactor that drops the
|
||||
# `max_bytes=` argument fails this test instead of regressing
|
||||
# silently to the default.
|
||||
assert recorded["max_bytes"] == 1024 * 1024
|
||||
assert recorded["label"] == "GitHub latest release"
|
||||
|
||||
|
||||
_FAILURE_CASES = [
|
||||
("offline or timeout", urllib.error.URLError("down")),
|
||||
(_RATE_LIMITED_REASON, _http_error(403)),
|
||||
|
||||
@@ -1935,6 +1935,14 @@ def _force_gate_stdin(monkeypatch, *, tty: bool):
|
||||
class TestInitStep:
|
||||
"""Test the init step type."""
|
||||
|
||||
def test_docstring_lists_every_valid_script_type(self):
|
||||
# The `script` field docstring must not contradict the step's own
|
||||
# VALID_SCRIPT_TYPES (which includes 'py'); validate() accepts all three.
|
||||
from specify_cli.workflows.steps.init import InitStep, VALID_SCRIPT_TYPES
|
||||
|
||||
for script_type in VALID_SCRIPT_TYPES:
|
||||
assert f"``{script_type}``" in InitStep.__doc__
|
||||
|
||||
def test_builds_here_argv_and_bootstraps(self, tmp_path):
|
||||
from specify_cli.workflows.steps.init import InitStep
|
||||
from specify_cli.workflows.base import StepContext, StepStatus
|
||||
@@ -2100,6 +2108,15 @@ class TestInitStep:
|
||||
class TestGateStep:
|
||||
"""Test the gate step type."""
|
||||
|
||||
def test_docstring_lists_every_on_reject_behaviour(self):
|
||||
# The docstring must not contradict validate()/execute(): on_reject
|
||||
# accepts 'abort', 'skip', AND 'retry' (execute() has a dedicated
|
||||
# retry -> PAUSED branch), but the summary omitted 'retry'.
|
||||
from specify_cli.workflows.steps.gate import GateStep
|
||||
|
||||
for behaviour in ("abort", "skip", "retry"):
|
||||
assert behaviour in GateStep.__doc__
|
||||
|
||||
@pytest.fixture(autouse=True)
|
||||
def _non_tty_stdin_by_default(self, monkeypatch):
|
||||
# Default every gate test to a non-TTY stdin so none can drop into
|
||||
@@ -8658,18 +8675,15 @@ steps:
|
||||
class FakeResponse:
|
||||
def __init__(self, data, url=None):
|
||||
self._data = data
|
||||
self._pos = 0
|
||||
self._url = url or "https://api.github.com/repos/org/repo/releases/assets/42"
|
||||
|
||||
def read(self, amt=None):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
if amt is None:
|
||||
chunk = self._data[self._pos :]
|
||||
self._pos = len(self._data)
|
||||
return chunk
|
||||
chunk = self._data[self._pos : self._pos + amt]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
def read(self, size=-1):
|
||||
if size < 0:
|
||||
size = len(self._data) - self._pos
|
||||
out = self._data[self._pos : self._pos + size]
|
||||
self._pos += len(out)
|
||||
return out
|
||||
|
||||
def geturl(self):
|
||||
return self._url
|
||||
@@ -8729,18 +8743,15 @@ steps:
|
||||
class FakeResponse:
|
||||
def __init__(self, data, url=None):
|
||||
self._data = data
|
||||
self._pos = 0
|
||||
self._url = url or "https://api.github.com/repos/org/repo/releases/assets/42"
|
||||
|
||||
def read(self, amt=None):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
if amt is None:
|
||||
chunk = self._data[self._pos :]
|
||||
self._pos = len(self._data)
|
||||
return chunk
|
||||
chunk = self._data[self._pos : self._pos + amt]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
def read(self, size=-1):
|
||||
if size < 0:
|
||||
size = len(self._data) - self._pos
|
||||
out = self._data[self._pos : self._pos + size]
|
||||
self._pos += len(out)
|
||||
return out
|
||||
|
||||
def geturl(self):
|
||||
return self._url
|
||||
@@ -8780,18 +8791,15 @@ steps:
|
||||
class FakeResponse:
|
||||
def __init__(self, data, url=None):
|
||||
self._data = data
|
||||
self._pos = 0
|
||||
self._url = url or "https://api.github.com/repos/org/repo/releases/assets/55"
|
||||
|
||||
def read(self, amt=None):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
if amt is None:
|
||||
chunk = self._data[self._pos :]
|
||||
self._pos = len(self._data)
|
||||
return chunk
|
||||
chunk = self._data[self._pos : self._pos + amt]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
def read(self, size=-1):
|
||||
if size < 0:
|
||||
size = len(self._data) - self._pos
|
||||
out = self._data[self._pos : self._pos + size]
|
||||
self._pos += len(out)
|
||||
return out
|
||||
|
||||
def geturl(self):
|
||||
return self._url
|
||||
@@ -8873,18 +8881,15 @@ steps:
|
||||
class FakeResponse:
|
||||
def __init__(self, data, url=None):
|
||||
self._data = data
|
||||
self._pos = 0
|
||||
self._url = url or "https://ghes.example/api/v3/repos/org/repo/releases/assets/42"
|
||||
|
||||
def read(self, amt=None):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
if amt is None:
|
||||
chunk = self._data[self._pos :]
|
||||
self._pos = len(self._data)
|
||||
return chunk
|
||||
chunk = self._data[self._pos : self._pos + amt]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
def read(self, size=-1):
|
||||
if size < 0:
|
||||
size = len(self._data) - self._pos
|
||||
out = self._data[self._pos : self._pos + size]
|
||||
self._pos += len(out)
|
||||
return out
|
||||
|
||||
def geturl(self):
|
||||
return self._url
|
||||
@@ -8936,18 +8941,15 @@ steps:
|
||||
class FakeResponse:
|
||||
def __init__(self, data, url=None):
|
||||
self._data = data
|
||||
self._pos = 0
|
||||
self._url = url or "https://ghes.example/api/v3/repos/org/repo/releases/assets/55"
|
||||
|
||||
def read(self, amt=None):
|
||||
if not hasattr(self, "_pos"):
|
||||
self._pos = 0
|
||||
if amt is None:
|
||||
chunk = self._data[self._pos :]
|
||||
self._pos = len(self._data)
|
||||
return chunk
|
||||
chunk = self._data[self._pos : self._pos + amt]
|
||||
self._pos += len(chunk)
|
||||
return chunk
|
||||
def read(self, size=-1):
|
||||
if size < 0:
|
||||
size = len(self._data) - self._pos
|
||||
out = self._data[self._pos : self._pos + size]
|
||||
self._pos += len(out)
|
||||
return out
|
||||
|
||||
def geturl(self):
|
||||
return self._url
|
||||
@@ -11751,6 +11753,10 @@ steps:
|
||||
_reject_insecure_download_redirect(
|
||||
"https://example.com/wf.yml", "http://localhost:8000/wf.yml"
|
||||
)
|
||||
with pytest.raises(urllib.error.URLError):
|
||||
_reject_insecure_download_redirect(
|
||||
"https://example.com/wf.yml", "https://127.0.0.2/wf.yml"
|
||||
)
|
||||
# Allowed: HTTPS anywhere, or loopback HTTP that stays on loopback HTTP.
|
||||
_reject_insecure_download_redirect(
|
||||
"https://example.com/wf.yml", "https://cdn.example.com/wf.yml"
|
||||
@@ -11761,6 +11767,9 @@ steps:
|
||||
_reject_insecure_download_redirect(
|
||||
"http://127.0.0.1/source.yml", "http://127.0.0.1/wf.yml"
|
||||
)
|
||||
_reject_insecure_download_redirect(
|
||||
"http://127.0.0.2/source.yml", "http://127.255.255.254/wf.yml"
|
||||
)
|
||||
|
||||
def test_add_from_url_passes_redirect_validator(self, project_dir, monkeypatch):
|
||||
from unittest.mock import patch
|
||||
|
||||
@@ -73,6 +73,23 @@ def test_build_is_deterministic(tmp_path: Path):
|
||||
assert first.artifact_path.read_bytes() == second.artifact_path.read_bytes()
|
||||
|
||||
|
||||
def test_member_order_is_platform_independent(tmp_path: Path):
|
||||
# Members must be laid out in canonical POSIX-arcname order (the same key
|
||||
# build_bundle uses to NAME them), not pathlib.Path order — which folds case
|
||||
# on Windows and would otherwise reorder members across build hosts, breaking
|
||||
# the byte-for-byte reproducibility guarantee. Mixed-case names make the
|
||||
# difference observable: Path order on Windows groups differently than the
|
||||
# canonical string sort.
|
||||
bundle = _make_bundle(
|
||||
tmp_path / "b",
|
||||
extra_files={"Zeta.txt": "z", "apple.txt": "a", "Foo.txt": "f", "bar.txt": "b"},
|
||||
)
|
||||
result = build_bundle(bundle, output_dir=tmp_path / "out")
|
||||
with zipfile.ZipFile(result.artifact_path) as archive:
|
||||
names = archive.namelist()
|
||||
assert names == sorted(names)
|
||||
|
||||
|
||||
def test_output_dir_inside_bundle_excludes_prior_artifacts(tmp_path: Path):
|
||||
bundle = _make_bundle(tmp_path / "b", extra_files={"a.txt": "a"})
|
||||
out_dir = bundle / "dist"
|
||||
|
||||
26
tests/unit/test_bundler_yamlio.py
Normal file
26
tests/unit/test_bundler_yamlio.py
Normal file
@@ -0,0 +1,26 @@
|
||||
"""Unit tests for the bundler YAML I/O helpers."""
|
||||
from __future__ import annotations
|
||||
|
||||
from pathlib import Path
|
||||
|
||||
from specify_cli.bundler.lib.yamlio import dump_yaml, load_yaml
|
||||
|
||||
|
||||
def test_dump_yaml_preserves_unicode(tmp_path: Path):
|
||||
# dump_yaml must write literal UTF-8, not \xNN / \uXXXX escapes, so bundle
|
||||
# config stays human-readable — matching _utils.dump_frontmatter and the
|
||||
# extensions/presets config writers (all allow_unicode=True).
|
||||
path = tmp_path / "f.yml"
|
||||
data = {"note": "café-münchen", "url": "https://例え.example"}
|
||||
dump_yaml(path, data)
|
||||
raw = path.read_text(encoding="utf-8")
|
||||
assert "café-münchen" in raw
|
||||
assert "例え" in raw
|
||||
assert "\\x" not in raw and "\\u" not in raw
|
||||
|
||||
|
||||
def test_dump_yaml_round_trips_unicode(tmp_path: Path):
|
||||
path = tmp_path / "f.yml"
|
||||
data = {"note": "café", "city": "münchen"}
|
||||
dump_yaml(path, data)
|
||||
assert load_yaml(path) == data
|
||||
@@ -162,6 +162,23 @@ class TestMergeSteps:
|
||||
ComposedStep("low-step", "project:low"),
|
||||
]
|
||||
|
||||
def test_merge_steps_multiple_insert_after_same_overlay_preserves_order(self):
|
||||
# Two insert_after edits from ONE overlay on the same anchor must keep
|
||||
# their declared order (a, x, y, b) — mirroring insert_before. The old
|
||||
# reversed(edits) over the flat list flipped them to (a, y, x, b).
|
||||
base = [_step("a"), _step("b")]
|
||||
overlay = Overlay(
|
||||
id="ov1",
|
||||
extends="wf",
|
||||
priority=10,
|
||||
edits=[
|
||||
OverlayEdit("insert_after", "a", _step("x")),
|
||||
OverlayEdit("insert_after", "a", _step("y")),
|
||||
],
|
||||
)
|
||||
steps, _ = merge_steps(base, [_layer(overlay, "project:ov1")])
|
||||
assert [s["id"] for s in steps] == ["a", "x", "y", "b"]
|
||||
|
||||
def test_merge_steps_replace_wins_over_insert(self):
|
||||
"""Overlays apply to the original tree only; targeting an overlay-introduced step raises."""
|
||||
base = [_step("a")]
|
||||
|
||||
Reference in New Issue
Block a user