Spaces:
Running
Running
| """GitHub元数据预检传输：凭据仅留在HTTPS API源，不跟随外部跳转。""" | |
| import re | |
| import time | |
| import urllib.error | |
| import urllib.parse | |
| import urllib.request | |
| from github_source import repository_parts | |
| MAX_PROBE_TIMEOUT = 10 | |
| MAX_PROBE_REDIRECTS = 3 | |
| def _metadata_target(url): | |
| if (not isinstance(url, str) or len(url) > 2048 | |
| or any(ord(char) <= 32 or ord(char) >= 127 for char in url) | |
| or "\\" in url or "?" in url or "#" in url): | |
| raise ValueError("invalid GitHub metadata target") | |
| parsed = urllib.parse.urlsplit(url) | |
| if (parsed.scheme != "https" or parsed.hostname != "api.github.com" | |
| or parsed.username is not None or parsed.password is not None | |
| or parsed.port not in (None, 443) or parsed.netloc.endswith(":")): | |
| raise ValueError("invalid GitHub metadata origin") | |
| if parsed.path.startswith("/repos/"): | |
| repository_parts("https://github.com/" + parsed.path[len("/repos/"):]) | |
| elif not re.fullmatch(r"/repositories/[0-9]{1,20}", parsed.path): | |
| raise ValueError("invalid GitHub metadata path") | |
| return url | |
| class GithubMetadataRedirect(urllib.request.HTTPRedirectHandler): | |
| max_redirections = MAX_PROBE_REDIRECTS | |
| max_repeats = 1 | |
| def __init__(self, deadline): | |
| self.deadline = deadline | |
| def redirect_request(self, req, fp, code, msg, headers, newurl): | |
| try: | |
| _metadata_target(req.full_url) | |
| _metadata_target(newurl) | |
| except ValueError: | |
| raise urllib.error.HTTPError(req.full_url, code, "GitHub预检跳转不允许", {}, fp) from None | |
| remaining = self.deadline - time.monotonic() | |
| if remaining <= 0: | |
| fp.close() | |
| raise TimeoutError("GitHub metadata probe timed out") | |
| # urllib的父处理器使用旧req.timeout传给下一跳；收紧到剩余预算。 | |
| req.timeout = min(getattr(req, "timeout", MAX_PROBE_TIMEOUT), remaining) | |
| return super().redirect_request(req, fp, code, msg, headers, newurl) | |
| def http_error_302(self, req, fp, code, msg, headers): | |
| location = headers.get("location") or headers.get("uri") | |
| if not isinstance(location, str) or not location or len(location) > 2048: | |
| raise urllib.error.HTTPError(req.full_url, code, "GitHub预检跳转无效", {}, fp) | |
| try: | |
| target = urllib.parse.urljoin(req.full_url, location) | |
| except ValueError: | |
| raise urllib.error.HTTPError(req.full_url, code, "GitHub预检跳转无效", {}, fp) from None | |
| new_request = self.redirect_request(req, fp, code, msg, headers, target) | |
| count = getattr(req, "_ranking_probe_redirects", 0) | |
| seen = getattr(req, "_ranking_probe_seen", {req.full_url}) | |
| if new_request is None or count >= MAX_PROBE_REDIRECTS or target in seen: | |
| raise urllib.error.HTTPError(req.full_url, code, "GitHub预检跳转次数过多", {}, fp) | |
| new_request._ranking_probe_redirects = count + 1 | |
| new_request._ranking_probe_seen = seen | {target} | |
| # 只看元数据状态：关闭跳转正文，不使用urllib默认的无界fp.read()。 | |
| fp.close() | |
| return self.parent.open(new_request, timeout=req.timeout) | |
| http_error_301 = http_error_303 = http_error_307 = http_error_308 = http_error_302 | |
| def open_github_probe(request, *, timeout=MAX_PROBE_TIMEOUT): | |
| """创建本次请求专用opener，不修改全局；HTTP错误也关闭其响应。""" | |
| _metadata_target(request.full_url) | |
| if (request.get_method() != "GET" or isinstance(timeout, bool) | |
| or not isinstance(timeout, (int, float)) or not 0 < timeout <= MAX_PROBE_TIMEOUT): | |
| raise ValueError("invalid GitHub metadata request") | |
| redirect = GithubMetadataRedirect(time.monotonic() + timeout) | |
| opener = urllib.request.build_opener(redirect) | |
| try: | |
| return opener.open(request, timeout=timeout) | |
| except urllib.error.HTTPError as error: | |
| try: | |
| error.close() | |
| except Exception: | |
| pass # 清理失败不覆盖原HTTP错误，不输出响应或地址。 | |
| raise | |