diff --git a/docs/reference/authentication.md b/docs/reference/authentication.md index e25bddff84..059052cd8f 100644 --- a/docs/reference/authentication.md +++ b/docs/reference/authentication.md @@ -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`) | Scheme | Header | Use for | diff --git a/src/specify_cli/__init__.py b/src/specify_cli/__init__.py index b2e8defb18..6713549d35 100644 --- a/src/specify_cli/__init__.py +++ b/src/specify_cli/__init__.py @@ -1128,9 +1128,10 @@ def _validate_and_install_local(yaml_path: Path, source_label: str) -> None: raise typer.Exit(1) 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 - _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: source = _resolved_wf_url _wf_url_extra_headers = {"Accept": "application/octet-stream"} @@ -1234,10 +1235,11 @@ def _validate_and_install_local(yaml_path: Path, source_label: str) -> None: try: 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 _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: workflow_url = _resolved_workflow_url _wf_cat_extra_headers = {"Accept": "application/octet-stream"} diff --git a/src/specify_cli/_github_http.py b/src/specify_cli/_github_http.py index d2030b57a8..31f6046395 100644 --- a/src/specify_cli/_github_http.py +++ b/src/specify_cli/_github_http.py @@ -10,6 +10,7 @@ import os import urllib.request +from fnmatch import fnmatch from typing import Callable, Dict, Optional 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) +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( download_url: str, open_url_fn: Callable, timeout: int = 60, + github_hosts: tuple[str, ...] = (), ) -> Optional[str]: - """Resolve a GitHub browser release URL to its REST API asset URL. - - For private or SSO-protected repositories, browser release download - URLs (``https://github.com///releases/download//``) - redirect to an HTML/SSO page instead of delivering the file. This - helper resolves such a URL to the matching GitHub REST API asset URL - (``https://api.github.com/repos/…/releases/assets/``), which can - then be downloaded with ``Accept: application/octet-stream`` and an - auth token to retrieve the actual file payload. - - If *download_url* is already a REST API asset URL, it is returned - as-is. Non-GitHub URLs and GitHub URLs that are not release-download - URLs return ``None``. If the API lookup fails (e.g. network error or - asset not found), ``None`` is returned so callers can fall back to the - original URL. + """Resolve a GitHub release browser-download URL to its REST API asset URL. + + Works for public ``github.com`` and for GitHub Enterprise Server (GHES) + hosts. A host is treated as GHES when it matches one of *github_hosts* + (exact hostname or ``*.suffix``) — supply the hosts the user has trusted + under a ``github`` provider in ``auth.json``. This allowlist is the + security gate: unlisted hosts never receive GHES API treatment, so a + malicious catalog cannot induce an API request to an arbitrary host. + + For a public URL the API base is ``https://api.github.com``; for a GHES + host it is ``{scheme}://{host[:port]}/api/v3``. Returns the API asset URL + (downloadable with ``Accept: application/octet-stream`` + a token), the + input unchanged if it is already an API asset URL, or ``None`` when the + URL is not a resolvable GitHub release download or the lookup fails. Args: download_url: The URL to resolve. open_url_fn: A callable compatible with - ``specify_cli.authentication.http.open_url`` used to make the - authenticated API request. + ``specify_cli.authentication.http.open_url`` used for the + authenticated release-metadata lookup. timeout: Per-request timeout in seconds. - - Returns: - The resolved REST API asset URL, or ``None`` if resolution is not - applicable or fails. + github_hosts: Host patterns to treat as GitHub Enterprise Server. """ import json import urllib.error parsed = urlparse(download_url) + hostname = (parsed.hostname or "").lower() parts = [unquote(part) for part in parsed.path.strip("/").split("/")] - # Already a REST API asset URL — use it directly - if ( - parsed.hostname == "api.github.com" - and len(parts) >= 6 - and parts[:1] == ["repos"] - and parts[3:5] == ["releases", "assets"] - ): + is_ghes = ( + bool(hostname) + and hostname not in GITHUB_HOSTS + and _host_matches(hostname, github_hosts) + ) + + 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 - # Only handle github.com browser release download URLs - if parsed.hostname != "github.com": + # Determine the REST API base for browser release-download URLs. + 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 # Expecting ///releases/download// @@ -114,7 +139,7 @@ def resolve_github_release_asset_api_url( owner, repo, tag = parts[0], parts[1], parts[4] asset_name = "/".join(parts[5:]) 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: with open_url_fn(release_url, timeout=timeout) as response: diff --git a/src/specify_cli/authentication/http.py b/src/specify_cli/authentication/http.py index e8ab8c1241..a2888bcce2 100644 --- a/src/specify_cli/authentication/http.py +++ b/src/specify_cli/authentication/http.py @@ -118,6 +118,20 @@ def build_request(url: str, extra_headers: dict[str, str] | None = None) -> urll 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( url: str, timeout: int = 10, diff --git a/src/specify_cli/extensions/__init__.py b/src/specify_cli/extensions/__init__.py index 3dd46ee6d2..9271a9fde6 100644 --- a/src/specify_cli/extensions/__init__.py +++ b/src/specify_cli/extensions/__init__.py @@ -2057,12 +2057,18 @@ def _resolve_github_release_asset_api_url( ) -> Optional[str]: """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.authentication.http import github_provider_hosts 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: diff --git a/src/specify_cli/presets/__init__.py b/src/specify_cli/presets/__init__.py index 07e31185ec..8d5c044193 100644 --- a/src/specify_cli/presets/__init__.py +++ b/src/specify_cli/presets/__init__.py @@ -1892,10 +1892,19 @@ def _resolve_github_release_asset_api_url( download_url: str, timeout: int = 60, ) -> 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.authentication.http import github_provider_hosts + 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: diff --git a/src/specify_cli/presets/_commands.py b/src/specify_cli/presets/_commands.py index 682bfe919d..eabfe650dd 100644 --- a/src/specify_cli/presets/_commands.py +++ b/src/specify_cli/presets/_commands.py @@ -144,10 +144,13 @@ def _validate_download_redirect(old_url, new_url): zip_path = Path(tmpdir) / "preset.zip" try: 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 _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: from_url = _resolved_from_url _preset_extra_headers = {"Accept": "application/octet-stream"} diff --git a/tests/test_authentication.py b/tests/test_authentication.py index 8b09245384..a89303d3d8 100644 --- a/tests/test_authentication.py +++ b/tests/test_authentication.py @@ -900,3 +900,45 @@ def test_accept_header_present(self, monkeypatch): with patch("specify_cli.authentication.http.urllib.request.urlopen", side_effect=side_effect): _fetch_latest_release_tag() 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") diff --git a/tests/test_extensions.py b/tests/test_extensions.py index 6b181a1204..e8dc2b7beb 100644 --- a/tests/test_extensions.py +++ b/tests/test_extensions.py @@ -16,8 +16,10 @@ import tempfile import shutil import tomllib +from contextlib import contextmanager from pathlib import Path from datetime import datetime, timezone +from unittest.mock import MagicMock from tests.conftest import strip_ansi from specify_cli.extensions import ( @@ -7280,3 +7282,36 @@ def test_add_dev_force_reinstall(self, tmp_path): ) assert result2.exit_code == 0, 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"] diff --git a/tests/test_github_http.py b/tests/test_github_http.py index e258f4917f..cd1b651aaa 100644 --- a/tests/test_github_http.py +++ b/tests/test_github_http.py @@ -188,3 +188,117 @@ def capturing_open(url, timeout=None, extra_headers=None): ) assert len(captured_urls) == 1 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"] diff --git a/tests/test_presets.py b/tests/test_presets.py index ff37dd3a96..0632fe3a89 100644 --- a/tests/test_presets.py +++ b/tests/test_presets.py @@ -17,9 +17,11 @@ import shutil import warnings import zipfile +from contextlib import contextmanager from pathlib import Path from datetime import datetime, timezone from types import SimpleNamespace +from unittest.mock import MagicMock import yaml @@ -4752,6 +4754,69 @@ def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None assert captured_urls[0][0] == "https://api.github.com/repos/org/repo/releases/assets/42" 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 ' 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: """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) 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"] diff --git a/tests/test_workflows.py b/tests/test_workflows.py index 5bbc9b6e53..988730d783 100644 --- a/tests/test_workflows.py +++ b/tests/test_workflows.py @@ -5477,6 +5477,137 @@ def fake_open_url(url, timeout=None, extra_headers=None): assert len(asset_calls) >= 1 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 ' 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 ' 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: """CLI-level tests for the run/resume process exit codes."""