snowy-yang
a7725f8812
在引入 uv 依赖管理、ruff 静态检查与格式化、pyright 类型检查之前， 先把当前可正常运行的状态固定下来，作为后续大规模自动改动的回滚点。 此时状态：97 项测试全绿，HTTP + WebSocket 多牧场服务器可用。
64 lines
2.4 KiB
Python
64 lines
2.4 KiB
Python
"""网络小工具：本地回环请求不该走系统代理。
|
|
|
|
为什么需要这个模块
|
|
------------------
|
|
在有 ``HTTP_PROXY`` 的机器上（公司网络、容器、CI 沙箱都常见），标准库
|
|
``urllib`` 会把 **所有** 请求——包括发给 ``127.0.0.1`` 的——都塞给代理。
|
|
于是本机自己开的游戏服务器会莫名其妙地连不上，报的还是"502 上游连接失败"
|
|
这种看不出根因的错误。
|
|
|
|
游戏客户端只应该关心"我和服务器之间"这一跳：
|
|
* 目标是回环地址（127.0.0.1 / localhost / ::1）→ 直连，不经过任何代理；
|
|
* 目标是别的机器 → 尊重用户环境里的代理设置。
|
|
|
|
WebSocket 那边不需要处理：``websockets`` 库默认不读代理环境变量。
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import urllib.parse
|
|
import urllib.request
|
|
from typing import Any, Optional, Union
|
|
|
|
LOOPBACK_HOSTS = {"127.0.0.1", "localhost", "::1", "[::1]", "0.0.0.0", ""}
|
|
|
|
_no_proxy_opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
|
|
|
|
def is_loopback(host: Optional[str]) -> bool:
|
|
"""判断主机名是否是本机回环（含 127.x.x.x 网段与 IPv6 回环）。"""
|
|
if host is None:
|
|
return False
|
|
host = host.strip().lower().strip("[]")
|
|
if host in LOOPBACK_HOSTS:
|
|
return True
|
|
return host.startswith("127.")
|
|
|
|
|
|
def host_of(target: Union[str, urllib.request.Request]) -> Optional[str]:
|
|
url = target.full_url if isinstance(target, urllib.request.Request) else target
|
|
try:
|
|
return urllib.parse.urlsplit(url).hostname
|
|
except ValueError:
|
|
return None
|
|
|
|
|
|
def open_url(target: Union[str, urllib.request.Request], timeout: float = 6.0,
|
|
host: Optional[str] = None) -> Any:
|
|
"""打开一个 URL 或 Request，回环地址自动绕过代理。
|
|
|
|
返回值是 ``http.client.HTTPResponse``，用法与 ``urllib.request.urlopen`` 一致。
|
|
"""
|
|
if not isinstance(target, urllib.request.Request):
|
|
target = urllib.request.Request(target, headers={"Accept": "application/json"})
|
|
target_host = host or host_of(target)
|
|
if is_loopback(target_host):
|
|
return _no_proxy_opener.open(target, timeout=timeout)
|
|
return urllib.request.urlopen(target, timeout=timeout)
|
|
|
|
|
|
def endpoint_host(url: str, default: str = "127.0.0.1") -> str:
|
|
"""从 ``http://host:port`` 里取出 host（带默认值）。"""
|
|
parts = urllib.parse.urlsplit(url if "://" in url else "http://" + url)
|
|
return parts.hostname or default
|