-
Notifications
You must be signed in to change notification settings - Fork 10.7k
Expand file tree
/
Copy path_github_http.py
More file actions
161 lines (137 loc) · 6.45 KB
/
Copy path_github_http.py
File metadata and controls
161 lines (137 loc) · 6.45 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
"""Shared GitHub HTTP request helpers.
Provides ``build_github_request()`` for attaching GITHUB_TOKEN / GH_TOKEN
credentials to requests targeting GitHub-hosted domains, and
``resolve_github_release_asset_api_url()`` — used by extensions, presets,
and workflow URL resolution — to translate browser release-download URLs
into GitHub REST API asset URLs. Authenticated downloads themselves go
through the config-driven helpers in :mod:`specify_cli.authentication.http`.
"""
import os
import urllib.request
from fnmatch import fnmatch
from typing import Callable, Dict, Optional
from urllib.parse import quote, unquote, urlparse
# GitHub-owned hostnames that should receive the Authorization header.
# Includes codeload.github.com because GitHub archive URL downloads
# (e.g. /archive/refs/tags/<tag>.zip) redirect there and require auth
# for private repositories.
GITHUB_HOSTS = frozenset({
"raw.githubusercontent.com",
"github.com",
"api.github.com",
"codeload.github.com",
})
def build_github_request(url: str) -> urllib.request.Request:
"""Build a urllib Request, adding a GitHub auth header when available.
Reads GITHUB_TOKEN or GH_TOKEN from the environment and attaches an
``Authorization: Bearer <value>`` header when the target hostname is one
of the known GitHub-owned domains. Non-GitHub URLs are returned as plain
requests so credentials are never leaked to third-party hosts.
Raises:
ValueError: If ``url`` is empty or whitespace-only.
ValueError: If ``url`` does not use the ``http`` or ``https`` scheme.
ValueError: If ``url`` does not include a hostname.
"""
headers: Dict[str, str] = {}
url = url.strip()
if not url:
raise ValueError("url must not be empty")
parsed = urlparse(url)
if parsed.scheme not in {"http", "https"}:
raise ValueError(f"url must start with http:// or https://, got: {url!r}")
if not parsed.hostname:
raise ValueError(f"url must include a hostname, got: {url!r}")
github_token = (os.environ.get("GITHUB_TOKEN") or "").strip()
gh_token = (os.environ.get("GH_TOKEN") or "").strip()
token = github_token or gh_token or None
hostname = parsed.hostname.lower()
if token and hostname in GITHUB_HOSTS:
headers["Authorization"] = f"Bearer {token}"
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 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 for the
authenticated release-metadata lookup.
timeout: Per-request timeout in seconds.
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("/")]
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
# Determine the REST API base for browser release-download URLs.
if hostname == "github.com":
api_base = "https://api.github.com"
elif is_ghes:
# ``parsed.port`` raises ValueError on a malformed port (e.g.
# ``host:notaport``); the function's contract is to return None for
# anything it can't resolve, not to raise.
try:
port = parsed.port
except ValueError:
return None
authority = hostname if port is None else f"{hostname}:{port}"
api_base = f"{parsed.scheme}://{authority}/api/v3"
else:
return None
# Expecting /<owner>/<repo>/releases/download/<tag>/<asset>
if len(parts) < 6 or parts[2:4] != ["releases", "download"]:
return None
owner, repo, tag = parts[0], parts[1], parts[4]
asset_name = "/".join(parts[5:])
encoded_tag = quote(tag, safe="")
release_url = f"{api_base}/repos/{owner}/{repo}/releases/tags/{encoded_tag}"
try:
with open_url_fn(release_url, timeout=timeout) as response:
release_data = json.loads(response.read())
except (urllib.error.URLError, json.JSONDecodeError):
return None
for asset in release_data.get("assets", []):
if asset.get("name") == asset_name and asset.get("url"):
return str(asset["url"])
return None