37 lines
1.2 KiB
Python
37 lines
1.2 KiB
Python
"""节点订阅的认证请求协议。"""
|
|
|
|
import base64
|
|
import json
|
|
import urllib.error
|
|
import urllib.request
|
|
|
|
|
|
class NodeAuthError(Exception):
|
|
"""订阅服务拒绝请求时携带的 HTTP 状态和用户可读原因。"""
|
|
|
|
def __init__(self, message, code=0):
|
|
super().__init__(message)
|
|
self.code = code
|
|
|
|
|
|
def b64url(value: str) -> str:
|
|
"""SSO 查询参数使用的无 padding URL-safe Base64。"""
|
|
return base64.urlsafe_b64encode(str(value).encode("utf-8")).decode("ascii").rstrip("=")
|
|
|
|
|
|
def fetch_nodes_auth(url, user, password, timeout=10):
|
|
"""直连拉取需 SSO 查询参数认证的 nodes.json。"""
|
|
opener = urllib.request.build_opener(urllib.request.ProxyHandler({}))
|
|
separator = "&" if "?" in url else "?"
|
|
request_url = f"{url}{separator}user={b64url(user)}&password={b64url(password)}"
|
|
try:
|
|
with opener.open(request_url, timeout=timeout) as response:
|
|
return json.loads(response.read().decode("utf-8"))
|
|
except urllib.error.HTTPError as error:
|
|
body = ""
|
|
try:
|
|
body = error.read().decode("utf-8", "replace").strip()
|
|
except Exception:
|
|
pass
|
|
raise NodeAuthError(body or f"请求被拒绝 (HTTP {error.code})", error.code) from None
|