Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
25 changes: 23 additions & 2 deletions src/specify_cli/commands/bundle/__init__.py
Original file line number Diff line number Diff line change
Expand Up @@ -794,22 +794,43 @@ def _download_remote_manifest(entry_id: str, url: str):
import tempfile

from ...authentication.http import open_url
from ..._github_http import resolve_github_release_asset_api_url

def _validate_redirect(old_url: str, new_url: str) -> None:
_require_https(f"bundle '{entry_id}'", new_url)

_require_https(f"bundle '{entry_id}'", url)

# For private/SSO-protected GitHub repos, browser release download URLs
# (https://github.com/<owner>/<repo>/releases/download/<tag>/<asset>)
# redirect to an HTML/SSO page instead of delivering the asset. Resolve
# such URLs to the GitHub REST API asset URL so the authenticated client
# can download the actual file.
extra_headers = None
effective_url = url
resolved = resolve_github_release_asset_api_url(url, open_url, timeout=30)
if resolved:
effective_url = resolved
extra_headers = {"Accept": "application/octet-stream"}

try:
with open_url(url, timeout=30, redirect_validator=_validate_redirect) as resp:
with open_url(
effective_url,
timeout=30,
redirect_validator=_validate_redirect,
extra_headers=extra_headers,
) as resp:
_require_https(f"bundle '{entry_id}'", resp.geturl())
raw = resp.read()
except BundlerError:
raise
except Exception as exc: # noqa: BLE001
raise BundlerError(f"Failed to download bundle '{entry_id}' from {url}: {exc}") from exc
raise BundlerError(f"Failed to download bundle '{entry_id}' from {effective_url}: {exc}") from exc

# A .zip artifact is written to a temp file and parsed via the local-source
# path (which extracts bundle.yml); any other payload is treated as YAML.
# Use the original catalog URL (``url``) to determine the artifact format
# since the resolved API URL does not carry the file extension.
if url.lower().endswith(".zip"):
with tempfile.TemporaryDirectory() as tmp:
artifact = Path(tmp) / "bundle.zip"
Expand Down
120 changes: 120 additions & 0 deletions tests/contract/test_bundle_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -8,6 +8,7 @@

import json
from pathlib import Path
from unittest.mock import patch

import pytest
import yaml
Expand Down Expand Up @@ -389,3 +390,122 @@ def test_install_integration_override_cannot_bypass_clash_guard(project: Path):
)
assert result.exit_code == 1
assert "claude" in result.output and "copilot" in result.output


# ===== Private GitHub release asset URL resolution =====


class FakeBundleResponse:
"""Minimal context-manager response stub for open_url fakes."""

def __init__(self, data: bytes, url: str = "https://api.github.com/repos/org/repo/releases/assets/99"):
self._data = data
self._url = url

def read(self) -> bytes:
return self._data

def geturl(self) -> str:
return self._url

def __enter__(self):
return self

def __exit__(self, *_):
return False


def _make_catalog_config(catalog_path: Path, project: Path) -> None:
"""Write a bundle-catalogs.yml pointing at *catalog_path* in *project*."""
config = {
"schema_version": "1.0",
"catalogs": [
{
"id": "test",
"url": str(catalog_path),
"priority": 1,
"install_policy": "install-allowed",
}
],
}
(project / ".specify" / "bundle-catalogs.yml").write_text(
yaml.safe_dump(config), encoding="utf-8"
)


def test_bundle_info_resolves_github_browser_release_url(project: Path):
"""bundle info resolves a private-repo browser release URL via the GitHub API."""
browser_url = "https://github.com/org/repo/releases/download/v1.0/bundle.yml"
api_asset_url = "https://api.github.com/repos/org/repo/releases/assets/99"

captured = []
manifest_yaml = yaml.safe_dump(valid_manifest_dict()).encode()

def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
captured.append((url, extra_headers))
if "releases/tags/" in url:
# GitHub API release-tags lookup — return asset list
return FakeBundleResponse(
json.dumps({
"assets": [{"name": "bundle.yml", "url": api_asset_url}]
}).encode(),
url=url,
)
# Actual asset download
return FakeBundleResponse(manifest_yaml, url=api_asset_url)

catalog = project / "catalog.json"
write_catalog_file(
catalog,
{"demo-bundle": catalog_entry_dict("demo-bundle", download_url=browser_url)},
)
_make_catalog_config(catalog, project)

with patch("specify_cli.authentication.http.open_url", side_effect=fake_open_url):
result = runner.invoke(app, ["bundle", "info", "demo-bundle", "--json"])

assert result.exit_code == 0, result.output

# The browser release URL must have been resolved via the GitHub tags API
tag_calls = [url for url, _ in captured if "releases/tags/" in url]
assert len(tag_calls) == 1, f"Expected exactly one tags API call; got {captured}"
assert "releases/tags/v1.0" in tag_calls[0]

# The actual download must use the resolved API asset URL with octet-stream
asset_calls = [(url, h) for url, h in captured if "releases/assets/" in url]
assert len(asset_calls) >= 1
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}


def test_bundle_info_passes_through_api_asset_url(project: Path):
"""bundle info passes a direct GitHub API asset URL through with octet-stream."""
api_asset_url = "https://api.github.com/repos/org/repo/releases/assets/77"

captured = []
manifest_yaml = yaml.safe_dump(valid_manifest_dict()).encode()

def fake_open_url(url, timeout=None, extra_headers=None, redirect_validator=None):
captured.append((url, extra_headers))
return FakeBundleResponse(manifest_yaml, url=api_asset_url)

catalog = project / "catalog.json"
write_catalog_file(
catalog,
{"demo-bundle": catalog_entry_dict("demo-bundle", download_url=api_asset_url)},
)
_make_catalog_config(catalog, project)

with patch("specify_cli.authentication.http.open_url", side_effect=fake_open_url):
result = runner.invoke(app, ["bundle", "info", "demo-bundle", "--json"])

assert result.exit_code == 0, result.output

# No tags API call — URL was already a REST asset URL
tag_calls = [url for url, _ in captured if "releases/tags/" in url]
assert len(tag_calls) == 0

# Exactly one download call to the asset URL with octet-stream
asset_calls = [(url, h) for url, h in captured if "releases/assets/" in url]
assert len(asset_calls) == 1
assert asset_calls[0][0] == api_asset_url
assert asset_calls[0][1] == {"Accept": "application/octet-stream"}