dekun
103 lines
3.1 KiB
Python
103 lines
3.1 KiB
Python
"""代理调用策略机。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import logging
|
|
from typing import Any
|
|
|
|
import httpx
|
|
|
|
from .config import get_control_settings
|
|
from .crypto import unseal
|
|
|
|
logger = logging.getLogger(__name__)
|
|
|
|
|
|
def node_token_plain(node: dict[str, Any]) -> str | None:
|
|
sealed = (node.get("token_sealed") or "").strip()
|
|
if not sealed:
|
|
return None
|
|
secret = get_control_settings().control_auth_secret
|
|
try:
|
|
return unseal(sealed, secret)
|
|
except Exception:
|
|
logger.exception("unseal fleet token failed node=%s", node.get("id"))
|
|
return None
|
|
|
|
|
|
async def call_node(
|
|
node: dict[str, Any],
|
|
method: str,
|
|
path: str,
|
|
*,
|
|
require_token: bool = True,
|
|
json_body: dict | None = None,
|
|
timeout: float | None = None,
|
|
) -> tuple[int, Any]:
|
|
settings = get_control_settings()
|
|
base = str(node["base_url"]).rstrip("/")
|
|
headers: dict[str, str] = {}
|
|
if require_token:
|
|
tok = node_token_plain(node)
|
|
if not tok:
|
|
return 400, {"detail": "该机尚未生成/保存中控 Token"}
|
|
headers["X-Fleet-Token"] = tok
|
|
headers["Authorization"] = f"Fleet {tok}"
|
|
url = f"{base}{path}"
|
|
to = float(timeout if timeout is not None else settings.control_http_timeout_sec)
|
|
try:
|
|
async with httpx.AsyncClient(timeout=to, follow_redirects=False) as client:
|
|
res = await client.request(method, url, headers=headers, json=json_body)
|
|
# 少数反代会 301/302 补尾斜杠；手动跟一次并保留 Token 头
|
|
if res.status_code in (301, 302, 307, 308) and res.headers.get("location"):
|
|
loc = res.headers["location"]
|
|
if loc.startswith("/"):
|
|
from urllib.parse import urljoin
|
|
|
|
loc = urljoin(url, loc)
|
|
res = await client.request(method, loc, headers=headers, json=json_body)
|
|
try:
|
|
data = res.json()
|
|
except Exception:
|
|
data = {"detail": res.text[:500]}
|
|
return res.status_code, data
|
|
except Exception as e:
|
|
return 502, {"detail": f"连接失败: {e}"}
|
|
|
|
|
|
async def probe_health(
|
|
node: dict[str, Any],
|
|
*,
|
|
timeout: float | None = None,
|
|
) -> dict[str, Any]:
|
|
code, data = await call_node(
|
|
node, "GET", "/health", require_token=False, timeout=timeout
|
|
)
|
|
if code == 200 and isinstance(data, dict):
|
|
return {
|
|
"online": True,
|
|
"health": data,
|
|
"error": None,
|
|
"from_fleet": False,
|
|
}
|
|
# fallback fleet status if health blocked
|
|
code2, data2 = await call_node(
|
|
node, "GET", "/api/fleet/status", require_token=True, timeout=timeout
|
|
)
|
|
if code2 == 200 and isinstance(data2, dict):
|
|
return {
|
|
"online": True,
|
|
"health": data2,
|
|
"error": None,
|
|
"from_fleet": True,
|
|
}
|
|
detail = ""
|
|
if isinstance(data, dict):
|
|
detail = str(data.get("detail") or "")
|
|
return {
|
|
"online": False,
|
|
"health": None,
|
|
"error": detail or f"HTTP {code}",
|
|
"from_fleet": False,
|
|
}
|