Skip to content

fix(github-http): validate release asset metadata URLs #4438

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
WOLIKIMCHENG wants to merge 1 commit into
base: main
Choose a base branch
from
+272 −7
Open

fix(github-http): validate release asset metadata URLs #4438

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
1 commit
Select commit
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
110 changes: 103 additions & 7 deletions src/specify_cli/_github_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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

Expand All @@ -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.

Expand Down Expand Up @@ -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"
Expand All @@ -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
Expand All @@ -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:
Expand All @@ -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
169 changes: 169 additions & 0 deletions tests/test_github_http.py
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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"}]}
Expand All @@ -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 = {}
Expand Down Expand Up @@ -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"
Expand Down Expand Up @@ -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(
Expand Down