From 79944ba9881f0c3b6ff66b9bee75165e60557c24 Mon Sep 17 00:00:00 2001 From: WOLIKIMCHENG <35391914+WOLIKIMCHENG@users.noreply.github.com> Date: Fri, 4 Sep 2026 14:44:11 +0800 Subject: [PATCH] fix(github-http): validate release asset metadata URLs Require metadata asset URLs to match the derived API origin, repository identity, and exact numeric release-asset endpoint. Reject malformed metadata URLs while preserving direct passthrough and browser-download fallback behavior. --- src/specify_cli/_github_http.py | 110 +++++++++++++++++++-- tests/test_github_http.py | 169 ++++++++++++++++++++++++++++++++ 2 files changed, 272 insertions(+), 7 deletions(-) diff --git a/src/specify_cli/_github_http.py b/src/specify_cli/_github_http.py index 8734528481..0b6b9be181 100644 --- a/src/specify_cli/_github_http.py +++ b/src/specify_cli/_github_http.py @@ -11,6 +11,7 @@ import os import urllib.request from fnmatch import fnmatch +from ipaddress import ip_address from typing import Callable, Dict, Optional from urllib.parse import quote, unquote, urlparse @@ -27,6 +28,19 @@ _MAX_RELEASE_METADATA_BYTES = 5 * 1024 * 1024 +def _has_valid_percent_escapes(value: str) -> bool: + """Return whether every percent sign in *value* starts a percent escape.""" + hex_digits = "0123456789abcdefABCDEF" + for index, character in enumerate(value): + if character == "%" and ( + index + 2 >= len(value) + or value[index + 1] not in hex_digits + or value[index + 2] not in hex_digits + ): + return False + return True + + def build_github_request(url: str) -> urllib.request.Request: """Build a urllib Request, adding a GitHub auth header when available. @@ -142,6 +156,16 @@ def _is_asset_path(segments: list[str]) -> bool: if hostname and parts[:2] == ["api", "v3"] and _is_asset_path(parts[2:]): return download_url + # Browser download URLs must be usable HTTP(S) URLs before they can cause + # a metadata request. Direct API-asset passthrough above intentionally + # retains its existing path-only behavior. + if parsed.scheme not in {"http", "https"}: + return None + try: + _browser_port = parsed.port + except ValueError: + return None + # Determine the REST API base for browser release-download URLs. if hostname == "github.com": api_base = "https://api.github.com" @@ -153,7 +177,10 @@ def _is_asset_path(segments: list[str]) -> bool: port = parsed.port except ValueError: return None - authority = hostname if port is None else f"{hostname}:{port}" + # ``urlparse().hostname`` removes IPv6 brackets. Restore them when + # constructing an authority so the derived API base remains a URL. + authority_host = f"[{hostname}]" if ":" in hostname else hostname + authority = authority_host if port is None else f"{authority_host}:{port}" api_base = f"{parsed.scheme}://{authority}/api/v3" else: return None @@ -168,6 +195,77 @@ def _is_asset_path(segments: list[str]) -> bool: encoded_tag = quote(tag, safe="") release_url = f"{api_base}/repos/{owner}/{repo}/releases/tags/{encoded_tag}" + def _is_expected_asset_url(asset_url: object) -> bool: + """Return whether metadata names this release's exact API asset endpoint.""" + if not isinstance(asset_url, str): + return False + # ``urlparse`` tolerates some raw spellings (for example whitespace) + # even though the original metadata value is returned to the caller. + # Reject those spellings before parsing rather than normalizing them. + if ( + any( + ord(character) <= 0x20 or ord(character) == 0x7F + for character in asset_url + ) + or any(delimiter in asset_url for delimiter in ("?", "#", ";")) + or not _has_valid_percent_escapes(asset_url) + ): + return False + try: + asset_parsed = urlparse(asset_url) + asset_host = asset_parsed.hostname + asset_port = asset_parsed.port + api_parsed = urlparse(api_base) + api_host = api_parsed.hostname + api_port = api_parsed.port + except ValueError: + return False + + if ( + asset_parsed.scheme not in {"http", "https"} + or not asset_host + or asset_parsed.username is not None + or asset_parsed.password is not None + or asset_parsed.query + or asset_parsed.fragment + or asset_parsed.params + ): + return False + + def _origin(parsed_url, host: str, port: int | None) -> tuple[str, str, int]: + default_port = 443 if parsed_url.scheme == "https" else 80 + try: + normalized_host = ip_address(host).compressed + except ValueError: + normalized_host = host.lower() + return ( + parsed_url.scheme, + normalized_host, + default_port if port is None else port, + ) + + if _origin(asset_parsed, asset_host, asset_port) != _origin( + api_parsed, api_host or "", api_port + ): + return False + + asset_parts = asset_parsed.path.split("/") + owner_index = 2 if api_base == "https://api.github.com" else 4 + expected_prefix = ( + ["", "repos"] + if api_base == "https://api.github.com" + else ["", "api", "v3", "repos"] + ) + return ( + len(asset_parts) == owner_index + 5 + and asset_parts[:owner_index] == expected_prefix + and unquote(asset_parts[owner_index]).casefold() == owner.casefold() + and unquote(asset_parts[owner_index + 1]).casefold() == repo.casefold() + and asset_parts[owner_index + 2:owner_index + 4] == ["releases", "assets"] + and asset_parts[-1].isascii() + and asset_parts[-1].isdigit() + ) + try: open_kwargs = {"timeout": timeout} if redirect_validator is not None: @@ -194,11 +292,9 @@ def _is_asset_path(segments: list[str]) -> bool: if not isinstance(assets, list): return None for asset in assets: - if ( - isinstance(asset, dict) - and asset.get("name") == asset_name - and asset.get("url") - ): - return str(asset["url"]) + if isinstance(asset, dict) and asset.get("name") == asset_name: + asset_url = asset.get("url") + if _is_expected_asset_url(asset_url): + return asset_url return None diff --git a/tests/test_github_http.py b/tests/test_github_http.py index fcf6ae9936..6131bacdd3 100644 --- a/tests/test_github_http.py +++ b/tests/test_github_http.py @@ -150,6 +150,61 @@ def test_resolves_browser_url_to_api_url(self): ) assert result == "https://api.github.com/repos/org/repo/releases/assets/99" + def test_accepts_case_variant_public_metadata_repository_identity(self): + """GitHub owner and repository names are case-insensitive identities.""" + asset_url = "https://api.github.com/repos/owner/repository/releases/assets/99" + result = resolve_github_release_asset_api_url( + "https://github.com/Owner/Repository/releases/download/v1.0/pack.zip", + self._make_open_url_fn( + {"assets": [{"name": "pack.zip", "url": asset_url}]} + ), + ) + assert result == asset_url + + @pytest.mark.parametrize( + "asset_url", + [ + None, + 99, + "http://metadata.example/repos/org/repo/releases/assets/99", + "https://wrong.example/repos/org/repo/releases/assets/99", + "http://api.github.com/repos/org/repo/releases/assets/99", + "https://api.github.com:8443/repos/org/repo/releases/assets/99", + "https://api.github.com:notaport/repos/org/repo/releases/assets/99", + "https://[not-an-ip]/repos/org/repo/releases/assets/99", + "https:///repos/org/repo/releases/assets/99", + "ftp://api.github.com/repos/org/repo/releases/assets/99", + "https://api.github.com/repos/other/repo/releases/assets/99", + "https://api.github.com/repos/org/other/releases/assets/99", + "https://api.github.com/repos/org/repo/releases/download/99", + "https://api.github.com/repos/org/repo/releases/assets/not-a-number", + "https://api.github.com/repos/org/repo/releases/assets/99/extra", + "https://api.github.com/repos/org/repo/releases/assets/99?download=1", + "https://api.github.com/repos/org/repo/releases/assets/99#fragment", + "https://user@api.github.com/repos/org/repo/releases/assets/99", + "https://api.github.com/repos/org/repo/releases/assets/99\n", + "https://api.github.com/repos/org/repo/releases/assets/99\t", + " https://api.github.com/repos/org/repo/releases/assets/99", + "https://api.github.com/repos/org/repo/releases/assets/99 ", + "https://api.github.com/repos/org/repo/releases/assets/99;", + "https://api.github.com/repos/org/repo/releases/assets/99?", + "https://api.github.com/repos/org/repo/releases/assets/99#", + "https://api.github.com/repos/org/repo/releases/assets/%", + "https://api.github.com/repos/org/repo/releases/assets/%9", + "https://api.github.com/repos/org/repo/releases/assets/%ZZ", + "https://api.github.com:65536/repos/org/repo/releases/assets/99", + ], + ) + def test_rejects_invalid_public_metadata_asset_url(self, asset_url): + """Metadata cannot replace a browser URL with a noncanonical API URL.""" + result = resolve_github_release_asset_api_url( + "https://github.com/org/repo/releases/download/v1.0/pack.zip", + self._make_open_url_fn( + {"assets": [{"name": "pack.zip", "url": asset_url}]} + ), + ) + assert result is None + def test_returns_none_when_asset_not_found(self): """Returns None when the release exists but asset name doesn't match.""" release_json = {"assets": [{"name": "other.zip", "url": "https://api.github.com/repos/org/repo/releases/assets/1"}]} @@ -174,6 +229,15 @@ def failing_open(url, timeout=None, extra_headers=None): ) assert result is None + @pytest.mark.parametrize("release_json", [[], {"assets": {}}]) + def test_returns_none_for_invalid_release_metadata(self, release_json): + """Malformed release metadata retains the existing None fallback.""" + result = resolve_github_release_asset_api_url( + "https://github.com/org/repo/releases/download/v1/pack.zip", + self._make_open_url_fn(release_json), + ) + assert result is None + def test_metadata_lookup_is_bounded_and_redirect_validated(self): """Release metadata reads stay bounded and use the caller's policy.""" captured = {} @@ -266,6 +330,18 @@ def test_resolves_ghes_browser_url_to_api_url(self): ) assert result == "https://ghes.example/api/v3/repos/o/r/releases/assets/7" + def test_accepts_case_variant_ghes_metadata_repository_identity(self): + """GHES owner and repository names are case-insensitive identities.""" + asset_url = ( + "https://ghes.example/api/v3/repos/owner/repository/releases/assets/7" + ) + result = resolve_github_release_asset_api_url( + "https://ghes.example/Owner/Repository/releases/download/v1/ext.zip", + self._make_open_url_fn({"assets": [{"name": "ext.zip", "url": asset_url}]}), + github_hosts=("ghes.example",), + ) + assert result == asset_url + 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" @@ -366,6 +442,99 @@ def capturing_open(url, timeout=None, extra_headers=None): ) assert captured == ["http://localhost:8000/api/v3/repos/o/r/releases/tags/v1"] + @pytest.mark.parametrize( + ("download_url", "asset_url", "expected_lookup_url"), + [ + ( + "http://[::1]/o/r/releases/download/v1/ext.zip", + "http://[::1]/api/v3/repos/o/r/releases/assets/7", + "http://[::1]/api/v3/repos/o/r/releases/tags/v1", + ), + ( + "https://[::1]:8443/o/r/releases/download/v1/ext.zip", + "https://[0:0:0:0:0:0:0:1]:8443/api/v3/repos/o/r/releases/assets/7", + "https://[::1]:8443/api/v3/repos/o/r/releases/tags/v1", + ), + ], + ) + def test_ghes_ipv6_api_base_and_metadata_origin( + self, download_url, asset_url, expected_lookup_url + ): + """GHES IPv6 API lookups retain brackets and compare equivalent literals.""" + captured = [] + + @contextmanager + def capturing_open(url, timeout=None, extra_headers=None): + captured.append(url) + resp = MagicMock() + resp.read.side_effect = io.BytesIO(json.dumps({ + "assets": [{"name": "ext.zip", "url": asset_url}] + }).encode()).read + yield resp + + result = resolve_github_release_asset_api_url( + download_url, + capturing_open, + github_hosts=("::1",), + ) + assert result == asset_url + assert captured == [expected_lookup_url] + + @pytest.mark.parametrize( + ("download_url", "asset_url", "github_hosts"), + [ + ( + "https://ghes.example/o/r/releases/download/v1/ext.zip", + "https://ghes.example/api/v3/repos/o/r/releases/assets/7", + ("ghes.example",), + ), + ( + "https://ghes.example:8443/o/r/releases/download/v1/ext.zip", + "https://GHES.EXAMPLE:8443/api/v3/repos/o/r/releases/assets/7", + ("ghes.example",), + ), + ( + "http://localhost:8000/o/r/releases/download/v1/ext.zip", + "http://localhost:8000/api/v3/repos/o/r/releases/assets/7", + ("localhost",), + ), + ( + "http://localhost/o/r/releases/download/v1/ext.zip", + "http://LOCALHOST:80/api/v3/repos/o/r/releases/assets/7", + ("localhost",), + ), + ], + ) + def test_accepts_same_origin_ghes_metadata_asset_url( + self, download_url, asset_url, github_hosts + ): + """GHES metadata URLs retain their derived scheme, host, and port.""" + result = resolve_github_release_asset_api_url( + download_url, + self._make_open_url_fn({"assets": [{"name": "ext.zip", "url": asset_url}]}), + github_hosts=github_hosts, + ) + assert result == asset_url + + @pytest.mark.parametrize( + "asset_url", + [ + "https://other.example/api/v3/repos/o/r/releases/assets/7", + "http://ghes.example/api/v3/repos/o/r/releases/assets/7", + "https://ghes.example:8443/api/v3/repos/o/r/releases/assets/7", + "https://ghes.example/api/v3/repos/o/other/releases/assets/7", + "https://ghes.example/api/v3/repos/o/r/releases/assets/7/extra", + ], + ) + def test_rejects_wrong_origin_or_path_for_ghes_metadata_asset_url(self, asset_url): + """GHES metadata URLs must match the derived API origin and endpoint.""" + result = resolve_github_release_asset_api_url( + "https://ghes.example/o/r/releases/download/v1/ext.zip", + self._make_open_url_fn({"assets": [{"name": "ext.zip", "url": asset_url}]}), + github_hosts=("ghes.example",), + ) + assert result is None + 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(