mirror of
https://github.com/github/spec-kit.git
synced 2026-07-03 12:28:06 +08:00
fix(extensions,presets,workflows): resolve private GHES release assets via /api/v3 (#3157)
* feat(auth): add github_provider_hosts() to enumerate GHES hosts from auth.json Assisted-by: Claude Code (model: claude-sonnet-4-6, autonomous) * fix(extensions): resolve GHES release assets via /api/v3 Generalizes resolve_github_release_asset_api_url to GitHub Enterprise Server hosts (gated by auth.json github hosts), fixing private GHES extension/preset downloads. github/spec-kit#3147 Assisted-by: Claude Code (model: claude-sonnet-4-6, autonomous) * fix(extensions,presets): pass auth.json github hosts into release resolver Assisted-by: Claude Code (model: claude-sonnet-4-6, autonomous) * docs(auth): document GHES private catalog + release-asset auth Assisted-by: Claude Code (model: claude-sonnet-4-6, autonomous) * fix(presets,workflows): pass auth.json github hosts into remaining release resolvers Wires preset add --from and workflow add through github_provider_hosts() so private GHES release assets resolve via /api/v3 there too. github/spec-kit#3147 Assisted-by: Claude Code (model: claude-sonnet-4-6, autonomous) * test(presets): use module-level io.BytesIO in GHES preset test Addresses Copilot review on PR #3157: drop unnecessary __import__("io") in test_preset_add_from_ghes_release_url_resolves_via_api_v3 since io is already imported at module level. * fix(github-http): pass through GHES asset API URLs by path shape Addresses Copilot review on PR #3157. A direct GHES /api/v3 release asset URL was only returned as already-resolved when its host was in the allowlist; otherwise the resolver returned None and the caller downloaded the same URL without 'Accept: application/octet-stream', fetching JSON metadata instead of the binary. Gate the passthrough on path shape alone, mirroring the github.com case. This is safe: passthrough returns the input URL unchanged and the caller fetches it either way, so no new request to an arbitrary host is induced; the token stays independently gated by auth.json in open_url. The allowlist remains the anti-SSRF gate on the tag-lookup resolving path. Add test_passthrough_for_unlisted_ghes_api_asset_url.
This commit is contained in:
@@ -69,6 +69,33 @@ Either `token` or `token_env` must be set for `bearer` and `basic-pat` schemes.
|
|||||||
}
|
}
|
||||||
```
|
```
|
||||||
|
|
||||||
|
### GitHub Enterprise Server (GHES)
|
||||||
|
|
||||||
|
To use a private catalog or extension hosted on a GitHub Enterprise Server
|
||||||
|
instance, add a `github` entry listing your GHES host(s). The same entry
|
||||||
|
authenticates both catalog JSON fetches **and** private release-asset
|
||||||
|
downloads — Specify recognizes the listed hosts as GitHub Enterprise and
|
||||||
|
resolves release downloads through the GHES REST API (`/api/v3`).
|
||||||
|
|
||||||
|
```json
|
||||||
|
{
|
||||||
|
"providers": [
|
||||||
|
{
|
||||||
|
"hosts": ["ghes.example.com", "raw.ghes.example.com", "codeload.ghes.example.com"],
|
||||||
|
"provider": "github",
|
||||||
|
"auth": "bearer",
|
||||||
|
"token_env": "GH_ENTERPRISE_TOKEN"
|
||||||
|
}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
```
|
||||||
|
|
||||||
|
List the **bare** web host (e.g. `ghes.example.com`) — release-download URLs
|
||||||
|
live there. If your instance uses subdomain isolation, also list the `raw.`
|
||||||
|
and `codeload.` subdomains your catalog/extension URLs use. A
|
||||||
|
`*.ghes.example.com` wildcard matches subdomains but **not** the bare host,
|
||||||
|
so always include the bare host explicitly.
|
||||||
|
|
||||||
### Azure DevOps (`azure-devops`)
|
### Azure DevOps (`azure-devops`)
|
||||||
|
|
||||||
| Scheme | Header | Use for |
|
| Scheme | Header | Use for |
|
||||||
|
|||||||
@@ -1128,9 +1128,10 @@ def workflow_add(
|
|||||||
raise typer.Exit(1)
|
raise typer.Exit(1)
|
||||||
|
|
||||||
from specify_cli._github_http import resolve_github_release_asset_api_url as _resolve_gh_asset
|
from specify_cli._github_http import resolve_github_release_asset_api_url as _resolve_gh_asset
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
|
||||||
_wf_url_extra_headers = None
|
_wf_url_extra_headers = None
|
||||||
_resolved_wf_url = _resolve_gh_asset(source, _open_url, timeout=30)
|
_resolved_wf_url = _resolve_gh_asset(source, _open_url, timeout=30, github_hosts=github_provider_hosts())
|
||||||
if _resolved_wf_url:
|
if _resolved_wf_url:
|
||||||
source = _resolved_wf_url
|
source = _resolved_wf_url
|
||||||
_wf_url_extra_headers = {"Accept": "application/octet-stream"}
|
_wf_url_extra_headers = {"Accept": "application/octet-stream"}
|
||||||
@@ -1234,10 +1235,11 @@ def workflow_add(
|
|||||||
|
|
||||||
try:
|
try:
|
||||||
from specify_cli.authentication.http import open_url as _open_url
|
from specify_cli.authentication.http import open_url as _open_url
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
from specify_cli._github_http import resolve_github_release_asset_api_url as _resolve_gh_asset
|
from specify_cli._github_http import resolve_github_release_asset_api_url as _resolve_gh_asset
|
||||||
|
|
||||||
_wf_cat_extra_headers = None
|
_wf_cat_extra_headers = None
|
||||||
_resolved_workflow_url = _resolve_gh_asset(workflow_url, _open_url, timeout=30)
|
_resolved_workflow_url = _resolve_gh_asset(workflow_url, _open_url, timeout=30, github_hosts=github_provider_hosts())
|
||||||
if _resolved_workflow_url:
|
if _resolved_workflow_url:
|
||||||
workflow_url = _resolved_workflow_url
|
workflow_url = _resolved_workflow_url
|
||||||
_wf_cat_extra_headers = {"Accept": "application/octet-stream"}
|
_wf_cat_extra_headers = {"Accept": "application/octet-stream"}
|
||||||
|
|||||||
@@ -10,6 +10,7 @@ through the config-driven helpers in :mod:`specify_cli.authentication.http`.
|
|||||||
|
|
||||||
import os
|
import os
|
||||||
import urllib.request
|
import urllib.request
|
||||||
|
from fnmatch import fnmatch
|
||||||
from typing import Callable, Dict, Optional
|
from typing import Callable, Dict, Optional
|
||||||
from urllib.parse import quote, unquote, urlparse
|
from urllib.parse import quote, unquote, urlparse
|
||||||
|
|
||||||
@@ -56,55 +57,79 @@ def build_github_request(url: str) -> urllib.request.Request:
|
|||||||
return urllib.request.Request(url, headers=headers)
|
return urllib.request.Request(url, headers=headers)
|
||||||
|
|
||||||
|
|
||||||
|
def _host_matches(hostname: str, patterns: tuple[str, ...]) -> bool:
|
||||||
|
"""Return True when *hostname* matches a pattern (exact or ``*.suffix``)."""
|
||||||
|
hostname = hostname.lower()
|
||||||
|
return any(p == hostname or fnmatch(hostname, p) for p in patterns)
|
||||||
|
|
||||||
|
|
||||||
def resolve_github_release_asset_api_url(
|
def resolve_github_release_asset_api_url(
|
||||||
download_url: str,
|
download_url: str,
|
||||||
open_url_fn: Callable,
|
open_url_fn: Callable,
|
||||||
timeout: int = 60,
|
timeout: int = 60,
|
||||||
|
github_hosts: tuple[str, ...] = (),
|
||||||
) -> Optional[str]:
|
) -> Optional[str]:
|
||||||
"""Resolve a GitHub browser release URL to its REST API asset URL.
|
"""Resolve a GitHub release browser-download URL to its REST API asset URL.
|
||||||
|
|
||||||
For private or SSO-protected repositories, browser release download
|
Works for public ``github.com`` and for GitHub Enterprise Server (GHES)
|
||||||
URLs (``https://github.com/<owner>/<repo>/releases/download/<tag>/<asset>``)
|
hosts. A host is treated as GHES when it matches one of *github_hosts*
|
||||||
redirect to an HTML/SSO page instead of delivering the file. This
|
(exact hostname or ``*.suffix``) — supply the hosts the user has trusted
|
||||||
helper resolves such a URL to the matching GitHub REST API asset URL
|
under a ``github`` provider in ``auth.json``. This allowlist is the
|
||||||
(``https://api.github.com/repos/…/releases/assets/<id>``), which can
|
security gate: unlisted hosts never receive GHES API treatment, so a
|
||||||
then be downloaded with ``Accept: application/octet-stream`` and an
|
malicious catalog cannot induce an API request to an arbitrary host.
|
||||||
auth token to retrieve the actual file payload.
|
|
||||||
|
|
||||||
If *download_url* is already a REST API asset URL, it is returned
|
For a public URL the API base is ``https://api.github.com``; for a GHES
|
||||||
as-is. Non-GitHub URLs and GitHub URLs that are not release-download
|
host it is ``{scheme}://{host[:port]}/api/v3``. Returns the API asset URL
|
||||||
URLs return ``None``. If the API lookup fails (e.g. network error or
|
(downloadable with ``Accept: application/octet-stream`` + a token), the
|
||||||
asset not found), ``None`` is returned so callers can fall back to the
|
input unchanged if it is already an API asset URL, or ``None`` when the
|
||||||
original URL.
|
URL is not a resolvable GitHub release download or the lookup fails.
|
||||||
|
|
||||||
Args:
|
Args:
|
||||||
download_url: The URL to resolve.
|
download_url: The URL to resolve.
|
||||||
open_url_fn: A callable compatible with
|
open_url_fn: A callable compatible with
|
||||||
``specify_cli.authentication.http.open_url`` used to make the
|
``specify_cli.authentication.http.open_url`` used for the
|
||||||
authenticated API request.
|
authenticated release-metadata lookup.
|
||||||
timeout: Per-request timeout in seconds.
|
timeout: Per-request timeout in seconds.
|
||||||
|
github_hosts: Host patterns to treat as GitHub Enterprise Server.
|
||||||
Returns:
|
|
||||||
The resolved REST API asset URL, or ``None`` if resolution is not
|
|
||||||
applicable or fails.
|
|
||||||
"""
|
"""
|
||||||
import json
|
import json
|
||||||
import urllib.error
|
import urllib.error
|
||||||
|
|
||||||
parsed = urlparse(download_url)
|
parsed = urlparse(download_url)
|
||||||
|
hostname = (parsed.hostname or "").lower()
|
||||||
parts = [unquote(part) for part in parsed.path.strip("/").split("/")]
|
parts = [unquote(part) for part in parsed.path.strip("/").split("/")]
|
||||||
|
|
||||||
# Already a REST API asset URL — use it directly
|
is_ghes = (
|
||||||
if (
|
bool(hostname)
|
||||||
parsed.hostname == "api.github.com"
|
and hostname not in GITHUB_HOSTS
|
||||||
and len(parts) >= 6
|
and _host_matches(hostname, github_hosts)
|
||||||
and parts[:1] == ["repos"]
|
)
|
||||||
and parts[3:5] == ["releases", "assets"]
|
|
||||||
):
|
def _is_asset_path(segments: list[str]) -> bool:
|
||||||
|
return (
|
||||||
|
len(segments) >= 6
|
||||||
|
and segments[:1] == ["repos"]
|
||||||
|
and segments[3:5] == ["releases", "assets"]
|
||||||
|
)
|
||||||
|
|
||||||
|
# Already a REST API asset URL — use it directly. Pure passthrough induces
|
||||||
|
# no new request: the caller fetches this same URL regardless, so it is
|
||||||
|
# gated on path shape alone rather than the GHES allowlist. The token stays
|
||||||
|
# independently gated by auth.json in the download helper, and only the
|
||||||
|
# resolving path below (which issues a tag-lookup request) needs the
|
||||||
|
# allowlist as its anti-SSRF gate.
|
||||||
|
if hostname == "api.github.com" and _is_asset_path(parts):
|
||||||
|
return download_url
|
||||||
|
if hostname and parts[:2] == ["api", "v3"] and _is_asset_path(parts[2:]):
|
||||||
return download_url
|
return download_url
|
||||||
|
|
||||||
# Only handle github.com browser release download URLs
|
# Determine the REST API base for browser release-download URLs.
|
||||||
if parsed.hostname != "github.com":
|
if hostname == "github.com":
|
||||||
|
api_base = "https://api.github.com"
|
||||||
|
elif is_ghes:
|
||||||
|
authority = hostname if parsed.port is None else f"{hostname}:{parsed.port}"
|
||||||
|
api_base = f"{parsed.scheme}://{authority}/api/v3"
|
||||||
|
else:
|
||||||
return None
|
return None
|
||||||
|
|
||||||
# Expecting /<owner>/<repo>/releases/download/<tag>/<asset>
|
# Expecting /<owner>/<repo>/releases/download/<tag>/<asset>
|
||||||
@@ -114,7 +139,7 @@ def resolve_github_release_asset_api_url(
|
|||||||
owner, repo, tag = parts[0], parts[1], parts[4]
|
owner, repo, tag = parts[0], parts[1], parts[4]
|
||||||
asset_name = "/".join(parts[5:])
|
asset_name = "/".join(parts[5:])
|
||||||
encoded_tag = quote(tag, safe="")
|
encoded_tag = quote(tag, safe="")
|
||||||
release_url = f"https://api.github.com/repos/{owner}/{repo}/releases/tags/{encoded_tag}"
|
release_url = f"{api_base}/repos/{owner}/{repo}/releases/tags/{encoded_tag}"
|
||||||
|
|
||||||
try:
|
try:
|
||||||
with open_url_fn(release_url, timeout=timeout) as response:
|
with open_url_fn(release_url, timeout=timeout) as response:
|
||||||
|
|||||||
@@ -118,6 +118,20 @@ def build_request(url: str, extra_headers: dict[str, str] | None = None) -> urll
|
|||||||
return urllib.request.Request(url, headers=headers)
|
return urllib.request.Request(url, headers=headers)
|
||||||
|
|
||||||
|
|
||||||
|
def github_provider_hosts() -> tuple[str, ...]:
|
||||||
|
"""Return host patterns from every ``github`` provider entry in ``auth.json``.
|
||||||
|
|
||||||
|
Used to classify which hosts are GitHub Enterprise Server instances when
|
||||||
|
resolving release-asset download URLs. Returns an empty tuple when no
|
||||||
|
``auth.json`` exists or it contains no ``github`` entries.
|
||||||
|
"""
|
||||||
|
hosts: list[str] = []
|
||||||
|
for entry in _load_config():
|
||||||
|
if entry.provider == "github":
|
||||||
|
hosts.extend(entry.hosts)
|
||||||
|
return tuple(hosts)
|
||||||
|
|
||||||
|
|
||||||
def open_url(
|
def open_url(
|
||||||
url: str,
|
url: str,
|
||||||
timeout: int = 10,
|
timeout: int = 10,
|
||||||
|
|||||||
@@ -2057,12 +2057,18 @@ class ExtensionCatalog(CatalogStackBase):
|
|||||||
) -> Optional[str]:
|
) -> Optional[str]:
|
||||||
"""Resolve a GitHub release asset URL to its API asset URL.
|
"""Resolve a GitHub release asset URL to its API asset URL.
|
||||||
|
|
||||||
Delegates to the shared helper in :mod:`specify_cli._github_http`.
|
Delegates to the shared helper in :mod:`specify_cli._github_http`,
|
||||||
|
passing the ``github`` provider hosts from ``auth.json`` so GitHub
|
||||||
|
Enterprise Server release assets resolve via ``/api/v3``.
|
||||||
"""
|
"""
|
||||||
from specify_cli._github_http import resolve_github_release_asset_api_url
|
from specify_cli._github_http import resolve_github_release_asset_api_url
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
|
||||||
return resolve_github_release_asset_api_url(
|
return resolve_github_release_asset_api_url(
|
||||||
download_url, self._open_url, timeout=timeout
|
download_url,
|
||||||
|
self._open_url,
|
||||||
|
timeout=timeout,
|
||||||
|
github_hosts=github_provider_hosts(),
|
||||||
)
|
)
|
||||||
|
|
||||||
def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None:
|
def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None:
|
||||||
|
|||||||
@@ -1892,10 +1892,19 @@ class PresetCatalog:
|
|||||||
download_url: str,
|
download_url: str,
|
||||||
timeout: int = 60,
|
timeout: int = 60,
|
||||||
) -> Optional[str]:
|
) -> Optional[str]:
|
||||||
"""Resolve a GitHub release asset URL to its REST API asset URL."""
|
"""Resolve a GitHub release asset URL to its REST API asset URL.
|
||||||
|
|
||||||
|
Passes the ``github`` provider hosts from ``auth.json`` so GitHub
|
||||||
|
Enterprise Server release assets resolve via ``/api/v3``.
|
||||||
|
"""
|
||||||
from specify_cli._github_http import resolve_github_release_asset_api_url
|
from specify_cli._github_http import resolve_github_release_asset_api_url
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
|
||||||
return resolve_github_release_asset_api_url(
|
return resolve_github_release_asset_api_url(
|
||||||
download_url, self._open_url, timeout=timeout
|
download_url,
|
||||||
|
self._open_url,
|
||||||
|
timeout=timeout,
|
||||||
|
github_hosts=github_provider_hosts(),
|
||||||
)
|
)
|
||||||
|
|
||||||
def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None:
|
def _validate_catalog_payload(self, catalog_data: Any, url: str) -> None:
|
||||||
|
|||||||
@@ -144,10 +144,13 @@ def preset_add(
|
|||||||
zip_path = Path(tmpdir) / "preset.zip"
|
zip_path = Path(tmpdir) / "preset.zip"
|
||||||
try:
|
try:
|
||||||
from specify_cli.authentication.http import open_url as _open_url
|
from specify_cli.authentication.http import open_url as _open_url
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
from specify_cli._github_http import resolve_github_release_asset_api_url
|
from specify_cli._github_http import resolve_github_release_asset_api_url
|
||||||
|
|
||||||
_preset_extra_headers = None
|
_preset_extra_headers = None
|
||||||
_resolved_from_url = resolve_github_release_asset_api_url(from_url, _open_url)
|
_resolved_from_url = resolve_github_release_asset_api_url(
|
||||||
|
from_url, _open_url, github_hosts=github_provider_hosts()
|
||||||
|
)
|
||||||
if _resolved_from_url:
|
if _resolved_from_url:
|
||||||
from_url = _resolved_from_url
|
from_url = _resolved_from_url
|
||||||
_preset_extra_headers = {"Accept": "application/octet-stream"}
|
_preset_extra_headers = {"Accept": "application/octet-stream"}
|
||||||
|
|||||||
@@ -900,3 +900,45 @@ class TestFetchLatestReleaseTagDelegation:
|
|||||||
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect):
|
with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect):
|
||||||
_fetch_latest_release_tag()
|
_fetch_latest_release_tag()
|
||||||
assert captured["request"].get_header("Accept") == "application/vnd.github+json"
|
assert captured["request"].get_header("Accept") == "application/vnd.github+json"
|
||||||
|
|
||||||
|
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
# github_provider_hosts
|
||||||
|
# ---------------------------------------------------------------------------
|
||||||
|
|
||||||
|
|
||||||
|
class TestGithubProviderHosts:
|
||||||
|
"""Tests for github_provider_hosts() — the GHES host allowlist source."""
|
||||||
|
|
||||||
|
def _set_config(self, monkeypatch, entries):
|
||||||
|
from specify_cli.authentication import http as _auth_http
|
||||||
|
monkeypatch.setattr(_auth_http, "_config_override", entries)
|
||||||
|
|
||||||
|
def test_returns_hosts_from_github_entries(self, monkeypatch):
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
self._set_config(monkeypatch, [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example", "raw.ghes.example"),
|
||||||
|
provider="github", auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
assert github_provider_hosts() == ("ghes.example", "raw.ghes.example")
|
||||||
|
|
||||||
|
def test_empty_when_no_config(self, monkeypatch):
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
self._set_config(monkeypatch, [])
|
||||||
|
assert github_provider_hosts() == ()
|
||||||
|
|
||||||
|
def test_ignores_non_github_providers(self, monkeypatch):
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
self._set_config(monkeypatch, [
|
||||||
|
AuthConfigEntry(hosts=("dev.azure.com",), provider="azure-devops",
|
||||||
|
auth="basic-pat", token="t"),
|
||||||
|
])
|
||||||
|
assert github_provider_hosts() == ()
|
||||||
|
|
||||||
|
def test_unions_multiple_github_entries(self, monkeypatch):
|
||||||
|
from specify_cli.authentication.http import github_provider_hosts
|
||||||
|
self._set_config(monkeypatch, [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example",), provider="github", auth="bearer", token="t"),
|
||||||
|
AuthConfigEntry(hosts=("github.com",), provider="github", auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
assert github_provider_hosts() == ("ghes.example", "github.com")
|
||||||
|
|||||||
@@ -16,8 +16,10 @@ import platform
|
|||||||
import tempfile
|
import tempfile
|
||||||
import shutil
|
import shutil
|
||||||
import tomllib
|
import tomllib
|
||||||
|
from contextlib import contextmanager
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
from tests.conftest import strip_ansi
|
from tests.conftest import strip_ansi
|
||||||
from specify_cli.extensions import (
|
from specify_cli.extensions import (
|
||||||
@@ -7280,3 +7282,36 @@ class TestExtensionForceCLI:
|
|||||||
)
|
)
|
||||||
assert result2.exit_code == 0, strip_ansi(result2.output)
|
assert result2.exit_code == 0, strip_ansi(result2.output)
|
||||||
assert "installed" in strip_ansi(result2.output)
|
assert "installed" in strip_ansi(result2.output)
|
||||||
|
|
||||||
|
|
||||||
|
def test_extension_wrapper_resolves_ghes_asset_when_host_configured(tmp_path, monkeypatch):
|
||||||
|
"""End-to-end wiring: auth.json github host → GHES asset resolution."""
|
||||||
|
from specify_cli.authentication import http as _auth_http
|
||||||
|
from specify_cli.authentication.config import AuthConfigEntry
|
||||||
|
from specify_cli.extensions import ExtensionCatalog
|
||||||
|
|
||||||
|
monkeypatch.setattr(_auth_http, "_config_override", [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example",), provider="github",
|
||||||
|
auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
catalog = ExtensionCatalog(tmp_path)
|
||||||
|
|
||||||
|
captured = []
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def fake_open(url, timeout=None, extra_headers=None):
|
||||||
|
captured.append(url)
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.read.return_value = json.dumps({
|
||||||
|
"assets": [{"name": "ext.zip",
|
||||||
|
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/7"}]
|
||||||
|
}).encode()
|
||||||
|
yield resp
|
||||||
|
|
||||||
|
monkeypatch.setattr(catalog, "_open_url", fake_open)
|
||||||
|
|
||||||
|
resolved = catalog._resolve_github_release_asset_api_url(
|
||||||
|
"https://ghes.example/o/r/releases/download/v1/ext.zip"
|
||||||
|
)
|
||||||
|
assert resolved == "https://ghes.example/api/v3/repos/o/r/releases/assets/7"
|
||||||
|
assert captured == ["https://ghes.example/api/v3/repos/o/r/releases/tags/v1"]
|
||||||
|
|||||||
@@ -188,3 +188,117 @@ class TestResolveGitHubReleaseAssetApiUrl:
|
|||||||
)
|
)
|
||||||
assert len(captured_urls) == 1
|
assert len(captured_urls) == 1
|
||||||
assert "releases/tags/v1%23beta" in captured_urls[0]
|
assert "releases/tags/v1%23beta" in captured_urls[0]
|
||||||
|
|
||||||
|
# --- GHES (GitHub Enterprise Server) ---
|
||||||
|
|
||||||
|
def test_resolves_ghes_browser_url_to_api_url(self):
|
||||||
|
"""A GHES browser release URL resolves to the /api/v3 asset URL."""
|
||||||
|
release_json = {
|
||||||
|
"assets": [
|
||||||
|
{"name": "ext.zip",
|
||||||
|
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/7"}
|
||||||
|
]
|
||||||
|
}
|
||||||
|
result = resolve_github_release_asset_api_url(
|
||||||
|
"https://ghes.example/o/r/releases/download/v1/ext.zip",
|
||||||
|
self._make_open_url_fn(release_json),
|
||||||
|
github_hosts=("ghes.example",),
|
||||||
|
)
|
||||||
|
assert result == "https://ghes.example/api/v3/repos/o/r/releases/assets/7"
|
||||||
|
|
||||||
|
def test_passthrough_for_existing_ghes_api_asset_url(self):
|
||||||
|
"""An already-resolved GHES /api/v3 asset URL is returned as-is."""
|
||||||
|
url = "https://ghes.example/api/v3/repos/o/r/releases/assets/7"
|
||||||
|
result = resolve_github_release_asset_api_url(
|
||||||
|
url, lambda *a, **kw: None, github_hosts=("ghes.example",)
|
||||||
|
)
|
||||||
|
assert result == url
|
||||||
|
|
||||||
|
def test_returns_none_for_ghes_host_not_in_allowlist(self):
|
||||||
|
"""Unlisted hosts get no GHES treatment and trigger no API call (anti-SSRF)."""
|
||||||
|
called = []
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def recording_open(url, timeout=None, extra_headers=None):
|
||||||
|
called.append(url)
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.read.return_value = b"{}"
|
||||||
|
yield resp
|
||||||
|
|
||||||
|
result = resolve_github_release_asset_api_url(
|
||||||
|
"https://ghes.example/o/r/releases/download/v1/ext.zip",
|
||||||
|
recording_open,
|
||||||
|
github_hosts=("other.example",),
|
||||||
|
)
|
||||||
|
assert result is None
|
||||||
|
assert called == []
|
||||||
|
|
||||||
|
def test_passthrough_for_unlisted_ghes_api_asset_url(self):
|
||||||
|
"""A direct GHES /api/v3 asset URL passes through even when the host is
|
||||||
|
not allowlisted: passthrough issues no API request, and the download
|
||||||
|
helper gates the token independently, so octet-stream resolution must
|
||||||
|
not be withheld."""
|
||||||
|
called = []
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def recording_open(url, timeout=None, extra_headers=None):
|
||||||
|
called.append(url)
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.read.return_value = b"{}"
|
||||||
|
yield resp
|
||||||
|
|
||||||
|
url = "https://ghes.example/api/v3/repos/o/r/releases/assets/7"
|
||||||
|
result = resolve_github_release_asset_api_url(
|
||||||
|
url, recording_open, github_hosts=("other.example",)
|
||||||
|
)
|
||||||
|
assert result == url
|
||||||
|
assert called == []
|
||||||
|
|
||||||
|
def test_ghes_api_base_preserves_scheme_and_port(self):
|
||||||
|
"""The GHES API base mirrors the URL scheme and keeps a non-standard port."""
|
||||||
|
captured = []
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def capturing_open(url, timeout=None, extra_headers=None):
|
||||||
|
captured.append(url)
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.read.return_value = json.dumps({"assets": []}).encode()
|
||||||
|
yield resp
|
||||||
|
|
||||||
|
resolve_github_release_asset_api_url(
|
||||||
|
"http://localhost:8000/o/r/releases/download/v1/ext.zip",
|
||||||
|
capturing_open,
|
||||||
|
github_hosts=("localhost",),
|
||||||
|
)
|
||||||
|
assert captured == ["http://localhost:8000/api/v3/repos/o/r/releases/tags/v1"]
|
||||||
|
|
||||||
|
def test_ghes_wildcard_does_not_match_bare_host(self):
|
||||||
|
"""A '*.suffix' pattern does not match the bare host (must list it explicitly)."""
|
||||||
|
result = resolve_github_release_asset_api_url(
|
||||||
|
"https://ghes.example/o/r/releases/download/v1/ext.zip",
|
||||||
|
lambda *a, **kw: None,
|
||||||
|
github_hosts=("*.ghes.example",),
|
||||||
|
)
|
||||||
|
assert result is None
|
||||||
|
|
||||||
|
def test_public_github_url_unaffected_by_github_hosts(self):
|
||||||
|
"""Public github.com still resolves via api.github.com even with github_hosts set."""
|
||||||
|
captured = []
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def capturing_open(url, timeout=None, extra_headers=None):
|
||||||
|
captured.append(url)
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.read.return_value = json.dumps({
|
||||||
|
"assets": [{"name": "pack.zip",
|
||||||
|
"url": "https://api.github.com/repos/org/repo/releases/assets/99"}]
|
||||||
|
}).encode()
|
||||||
|
yield resp
|
||||||
|
|
||||||
|
result = resolve_github_release_asset_api_url(
|
||||||
|
"https://github.com/org/repo/releases/download/v1.0/pack.zip",
|
||||||
|
capturing_open,
|
||||||
|
github_hosts=("ghes.example",),
|
||||||
|
)
|
||||||
|
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"]
|
||||||
|
|||||||
@@ -17,9 +17,11 @@ import tempfile
|
|||||||
import shutil
|
import shutil
|
||||||
import warnings
|
import warnings
|
||||||
import zipfile
|
import zipfile
|
||||||
|
from contextlib import contextmanager
|
||||||
from pathlib import Path
|
from pathlib import Path
|
||||||
from datetime import datetime, timezone
|
from datetime import datetime, timezone
|
||||||
from types import SimpleNamespace
|
from types import SimpleNamespace
|
||||||
|
from unittest.mock import MagicMock
|
||||||
|
|
||||||
import yaml
|
import yaml
|
||||||
|
|
||||||
@@ -4752,6 +4754,69 @@ class TestPresetAddFromUrlResolution:
|
|||||||
assert captured_urls[0][0] == "https://api.github.com/repos/org/repo/releases/assets/42"
|
assert captured_urls[0][0] == "https://api.github.com/repos/org/repo/releases/assets/42"
|
||||||
assert captured_urls[0][1] == {"Accept": "application/octet-stream"}
|
assert captured_urls[0][1] == {"Accept": "application/octet-stream"}
|
||||||
|
|
||||||
|
def test_preset_add_from_ghes_release_url_resolves_via_api_v3(self, project_dir, monkeypatch):
|
||||||
|
"""'preset add --from <ghes-release-url>' resolves via GHES /api/v3 endpoint."""
|
||||||
|
from typer.testing import CliRunner
|
||||||
|
from unittest.mock import patch
|
||||||
|
from specify_cli import app
|
||||||
|
from specify_cli.authentication import http as _auth_http
|
||||||
|
from specify_cli.authentication.config import AuthConfigEntry
|
||||||
|
|
||||||
|
monkeypatch.setattr(_auth_http, "_config_override", [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example",), provider="github", auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
|
||||||
|
manifest_content = yaml.dump({
|
||||||
|
"schema_version": "1.0",
|
||||||
|
"preset": {"id": "my-preset", "name": "My Preset", "version": "1.0.0", "description": "Test preset", "author": "Test", "license": "MIT"},
|
||||||
|
"requires": {"speckit_version": ">=0.1.0"},
|
||||||
|
"provides": {"templates": [{"type": "template", "name": "t", "file": "templates/t.md", "description": "t"}]},
|
||||||
|
})
|
||||||
|
zip_buf = io.BytesIO()
|
||||||
|
with zipfile.ZipFile(zip_buf, "w") as zf:
|
||||||
|
zf.writestr("preset.yml", manifest_content)
|
||||||
|
zip_bytes = zip_buf.getvalue()
|
||||||
|
|
||||||
|
captured_urls = []
|
||||||
|
|
||||||
|
class FakeResponse:
|
||||||
|
def __init__(self, data):
|
||||||
|
self._data = data
|
||||||
|
|
||||||
|
def read(self):
|
||||||
|
return self._data
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, *a):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
|
||||||
|
captured_urls.append((url, extra_headers))
|
||||||
|
if "releases/tags/" in url:
|
||||||
|
return FakeResponse(json.dumps({
|
||||||
|
"assets": [{"name": "preset.zip", "url": "https://ghes.example/api/v3/repos/org/repo/releases/assets/42"}]
|
||||||
|
}).encode())
|
||||||
|
return FakeResponse(zip_bytes)
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
with patch.object(Path, "cwd", return_value=project_dir), \
|
||||||
|
patch("specify_cli.get_speckit_version", return_value="1.0.0"), \
|
||||||
|
patch("specify_cli.authentication.http.open_url", side_effect=fake_open_url):
|
||||||
|
result = runner.invoke(app, [
|
||||||
|
"preset", "add",
|
||||||
|
"--from", "https://ghes.example/org/repo/releases/download/v1.0/preset.zip",
|
||||||
|
])
|
||||||
|
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
# The tag-lookup call must use the GHES /api/v3 endpoint
|
||||||
|
assert any("ghes.example/api/v3/repos/org/repo/releases/tags/v1.0" in url for url, _ in captured_urls)
|
||||||
|
# The asset download call must carry Accept: application/octet-stream
|
||||||
|
asset_calls = [(url, h) for url, h in captured_urls if "releases/assets/" in url]
|
||||||
|
assert len(asset_calls) >= 1
|
||||||
|
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}
|
||||||
|
|
||||||
|
|
||||||
class TestWrapStrategy:
|
class TestWrapStrategy:
|
||||||
"""Tests for strategy: wrap preset command substitution."""
|
"""Tests for strategy: wrap preset command substitution."""
|
||||||
@@ -6021,3 +6086,36 @@ def _create_pack(temp_dir, valid_pack_data, pack_id, content,
|
|||||||
(subdir / f"{template_name}.md").write_text(content)
|
(subdir / f"{template_name}.md").write_text(content)
|
||||||
|
|
||||||
return pack_dir
|
return pack_dir
|
||||||
|
|
||||||
|
|
||||||
|
def test_preset_wrapper_resolves_ghes_asset_when_host_configured(tmp_path, monkeypatch):
|
||||||
|
"""End-to-end wiring for presets: auth.json github host → GHES asset resolution."""
|
||||||
|
from specify_cli.authentication import http as _auth_http
|
||||||
|
from specify_cli.authentication.config import AuthConfigEntry
|
||||||
|
from specify_cli.presets import PresetCatalog
|
||||||
|
|
||||||
|
monkeypatch.setattr(_auth_http, "_config_override", [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example",), provider="github",
|
||||||
|
auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
catalog = PresetCatalog(tmp_path)
|
||||||
|
|
||||||
|
captured = []
|
||||||
|
|
||||||
|
@contextmanager
|
||||||
|
def fake_open(url, timeout=None, extra_headers=None):
|
||||||
|
captured.append(url)
|
||||||
|
resp = MagicMock()
|
||||||
|
resp.read.return_value = json.dumps({
|
||||||
|
"assets": [{"name": "pack.zip",
|
||||||
|
"url": "https://ghes.example/api/v3/repos/o/r/releases/assets/9"}]
|
||||||
|
}).encode()
|
||||||
|
yield resp
|
||||||
|
|
||||||
|
monkeypatch.setattr(catalog, "_open_url", fake_open)
|
||||||
|
|
||||||
|
resolved = catalog._resolve_github_release_asset_api_url(
|
||||||
|
"https://ghes.example/o/r/releases/download/v2/pack.zip"
|
||||||
|
)
|
||||||
|
assert resolved == "https://ghes.example/api/v3/repos/o/r/releases/assets/9"
|
||||||
|
assert captured == ["https://ghes.example/api/v3/repos/o/r/releases/tags/v2"]
|
||||||
|
|||||||
@@ -5477,6 +5477,137 @@ steps:
|
|||||||
assert len(asset_calls) >= 1
|
assert len(asset_calls) >= 1
|
||||||
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}
|
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}
|
||||||
|
|
||||||
|
def test_workflow_add_from_ghes_release_url_resolves_via_api_v3(self, project_dir, monkeypatch):
|
||||||
|
"""'workflow add <ghes-release-url>' resolves via GHES /api/v3 endpoint."""
|
||||||
|
from typer.testing import CliRunner
|
||||||
|
from unittest.mock import patch
|
||||||
|
from specify_cli import app
|
||||||
|
from specify_cli.authentication import http as _auth_http
|
||||||
|
from specify_cli.authentication.config import AuthConfigEntry
|
||||||
|
|
||||||
|
monkeypatch.setattr(_auth_http, "_config_override", [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example",), provider="github", auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
|
||||||
|
captured_urls = []
|
||||||
|
|
||||||
|
class FakeResponse:
|
||||||
|
def __init__(self, data, url=None):
|
||||||
|
self._data = data
|
||||||
|
self._url = url or "https://ghes.example/api/v3/repos/org/repo/releases/assets/42"
|
||||||
|
|
||||||
|
def read(self):
|
||||||
|
return self._data
|
||||||
|
|
||||||
|
def geturl(self):
|
||||||
|
return self._url
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, *a):
|
||||||
|
return False
|
||||||
|
|
||||||
|
def fake_open_url(url, timeout=None, extra_headers=None):
|
||||||
|
captured_urls.append((url, extra_headers))
|
||||||
|
if "releases/tags/" in url:
|
||||||
|
return FakeResponse(json.dumps({
|
||||||
|
"assets": [{"name": "workflow.yml", "url": "https://ghes.example/api/v3/repos/org/repo/releases/assets/42"}]
|
||||||
|
}).encode())
|
||||||
|
return FakeResponse(self.VALID_WORKFLOW_YAML.encode())
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
with patch.object(Path, "cwd", return_value=project_dir), \
|
||||||
|
patch("specify_cli.authentication.http.open_url", side_effect=fake_open_url):
|
||||||
|
result = runner.invoke(app, [
|
||||||
|
"workflow", "add",
|
||||||
|
"https://ghes.example/org/repo/releases/download/v1.0/workflow.yml",
|
||||||
|
])
|
||||||
|
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
# Tag lookup must use the GHES /api/v3 endpoint
|
||||||
|
assert any("ghes.example/api/v3/repos/org/repo/releases/tags/v1.0" in url for url, _ in captured_urls)
|
||||||
|
# Asset download must carry Accept: application/octet-stream
|
||||||
|
asset_calls = [(url, h) for url, h in captured_urls if "releases/assets/" in url]
|
||||||
|
assert len(asset_calls) >= 1
|
||||||
|
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}
|
||||||
|
|
||||||
|
def test_workflow_add_catalog_based_ghes_release_url_resolves_via_api_v3(self, project_dir, monkeypatch):
|
||||||
|
"""'workflow add <id>' with a GHES catalog URL resolves via /api/v3."""
|
||||||
|
from typer.testing import CliRunner
|
||||||
|
from unittest.mock import patch
|
||||||
|
from specify_cli import app
|
||||||
|
from specify_cli.authentication import http as _auth_http
|
||||||
|
from specify_cli.authentication.config import AuthConfigEntry
|
||||||
|
|
||||||
|
monkeypatch.setattr(_auth_http, "_config_override", [
|
||||||
|
AuthConfigEntry(hosts=("ghes.example",), provider="github", auth="bearer", token="t"),
|
||||||
|
])
|
||||||
|
|
||||||
|
captured_urls = []
|
||||||
|
|
||||||
|
class FakeResponse:
|
||||||
|
def __init__(self, data, url=None):
|
||||||
|
self._data = data
|
||||||
|
self._url = url or "https://ghes.example/api/v3/repos/org/repo/releases/assets/55"
|
||||||
|
|
||||||
|
def read(self):
|
||||||
|
return self._data
|
||||||
|
|
||||||
|
def geturl(self):
|
||||||
|
return self._url
|
||||||
|
|
||||||
|
def __enter__(self):
|
||||||
|
return self
|
||||||
|
|
||||||
|
def __exit__(self, *a):
|
||||||
|
return False
|
||||||
|
|
||||||
|
ghes_wf_yaml = """
|
||||||
|
schema_version: "1.0"
|
||||||
|
workflow:
|
||||||
|
id: "my-wf"
|
||||||
|
name: "My GHES Workflow"
|
||||||
|
version: "1.0.0"
|
||||||
|
description: "A GHES catalog workflow"
|
||||||
|
steps:
|
||||||
|
- id: step-one
|
||||||
|
type: shell
|
||||||
|
run: "echo hello"
|
||||||
|
"""
|
||||||
|
|
||||||
|
def fake_open_url(url, timeout=None, extra_headers=None):
|
||||||
|
captured_urls.append((url, extra_headers))
|
||||||
|
if "releases/tags/" in url:
|
||||||
|
return FakeResponse(json.dumps({
|
||||||
|
"assets": [{"name": "workflow.yml", "url": "https://ghes.example/api/v3/repos/org/repo/releases/assets/55"}]
|
||||||
|
}).encode())
|
||||||
|
return FakeResponse(ghes_wf_yaml.encode())
|
||||||
|
|
||||||
|
fake_catalog_info = {
|
||||||
|
"id": "my-wf",
|
||||||
|
"name": "My GHES Workflow",
|
||||||
|
"version": "1.0.0",
|
||||||
|
"url": "https://ghes.example/org/repo/releases/download/v2.0/workflow.yml",
|
||||||
|
"_install_allowed": True,
|
||||||
|
}
|
||||||
|
|
||||||
|
runner = CliRunner()
|
||||||
|
with patch.object(Path, "cwd", return_value=project_dir), \
|
||||||
|
patch("specify_cli.authentication.http.open_url", side_effect=fake_open_url), \
|
||||||
|
patch("specify_cli.workflows.catalog.WorkflowCatalog.get_workflow_info", return_value=fake_catalog_info):
|
||||||
|
result = runner.invoke(app, ["workflow", "add", "my-wf"])
|
||||||
|
|
||||||
|
assert result.exit_code == 0, result.output
|
||||||
|
# Tag lookup must use GHES /api/v3
|
||||||
|
tag_calls = [url for url, _ in captured_urls if "releases/tags/" in url]
|
||||||
|
assert len(tag_calls) == 1
|
||||||
|
assert "ghes.example/api/v3/repos/org/repo/releases/tags/v2.0" in tag_calls[0]
|
||||||
|
# Asset download must carry Accept: application/octet-stream
|
||||||
|
asset_calls = [(url, h) for url, h in captured_urls if "releases/assets/" in url]
|
||||||
|
assert len(asset_calls) >= 1
|
||||||
|
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}
|
||||||
|
|
||||||
|
|
||||||
class TestWorkflowRunExitCodes:
|
class TestWorkflowRunExitCodes:
|
||||||
"""CLI-level tests for the run/resume process exit codes."""
|
"""CLI-level tests for the run/resume process exit codes."""
|
||||||
|
|||||||
Reference in New Issue
Block a user