dekun/LocalNav
0
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

修复中控

Browse Source
This commit is contained in:
dekun
2026-05-30 15:19:38 +08:00
parent 4cd5a48dc1
commit a67d7aa58b
7 changed files with 447 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
app.py
+148 -45
View File
@@ -93,6 +93,103 @@ def _apply_session_cookie_settings(app: Flask) -> None:
app.config["REMEMBER_COOKIE_SECURE"] = False
_HUB_SESSION_COOKIE = "hub_sess"
def _hub_api_login(base: str, username: str, password: str) -> tuple[str | None, str | None]:
"""服务端登录中控，返回 (session_token, error_detail)。"""
payload = json.dumps({"username": username, "password": password}).encode("utf-8")
req = urllib.request.Request(
f"{base.rstrip('/')}/api/auth/login",
data=payload,
headers={"Content-Type": "application/json", "Accept": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
raw = resp.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as e:
try:
err_body = e.read().decode("utf-8", errors="replace")
detail = json.loads(err_body).get("detail", err_body)
except Exception:
detail = str(e)
return None, detail or "中控登录失败"
except Exception as e:
return None, f"无法连接中控: {e}"
try:
data = json.loads(raw)
except json.JSONDecodeError:
return None, "中控返回非 JSON"
if not data.get("ok"):
return None, data.get("detail") or "登录失败"
token = (data.get("session_token") or "").strip()
if not token:
return None, "中控未返回 session_token，请升级云端 hub 后重试"
return token, None
def _hub_instance_open_url(
base: str, session_token: str, exchange_id: str, next_path: str, *, embed: bool = True
) -> tuple[str | None, str | None]:
"""已登录中控会话下签发实例 SSO 打开链接。"""
params = {"exchange_id": str(exchange_id), "next": next_path or "/"}
if embed:
params["embed"] = "1"
q = urlencode(params)
req = urllib.request.Request(
f"{base.rstrip('/')}/api/instance/open-url?{q}",
headers={
"Accept": "application/json",
"Cookie": f"{_HUB_SESSION_COOKIE}={session_token}",
},
)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
raw = resp.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as e:
try:
err_body = e.read().decode("utf-8", errors="replace")
detail = json.loads(err_body).get("detail", err_body)
except Exception:
detail = str(e)
return None, detail or "无法签发实例链接"
except Exception as e:
return None, f"无法连接中控: {e}"
try:
data = json.loads(raw)
except json.JSONDecodeError:
return None, "中控返回非 JSON"
url = (data.get("url") or "").strip()
if not data.get("ok") or not url:
return None, data.get("detail") or "无法签发实例链接"
return url, None
def _resolve_hub_embed_service(body: dict) -> tuple[str | None, str | None, str | None]:
"""返回 (base_origin, default_next_path, error_detail)。"""
sid = body.get("service_id")
base = (body.get("base_url") or "").strip().rstrip("/")
next_path = (body.get("next") or "/monitor").strip() or "/monitor"
if not next_path.startswith("/"):
next_path = "/" + next_path
if sid:
svc = db.session.get(Service, int(sid))
if not svc or not svc.is_hub_embed():
return None, None, "服务不存在或未标记为中控"
base = svc.build_origin()
if not body.get("next") or next_path == "/monitor":
p = (svc.path or "/monitor").strip() or "/monitor"
next_path = p if p.startswith("/") else "/" + p
if not base:
return None, None, "缺少 base_url"
return base, next_path, None
def create_app() -> Flask:
app = Flask(__name__)
app.config["SECRET_KEY"] = os.environ.get("NAV_SECRET_KEY") or secrets.token_hex(32)
@@ -178,65 +275,71 @@ def create_app() -> Flask:
避免浏览器在跨站 iframe 里丢弃 Set-Cookie。
"""
body = request.get_json(silent=True) or {}
sid = body.get("service_id")
base = (body.get("base_url") or "").strip().rstrip("/")
next_path = (body.get("next") or "/monitor").strip() or "/monitor"
if not next_path.startswith("/"):
next_path = "/" + next_path
username = (body.get("username") or os.environ.get("NAV_HUB_USERNAME") or "").strip()
password = body.get("password")
if password is None:
password = os.environ.get("NAV_HUB_PASSWORD") or ""
password = str(password)
if sid:
svc = db.session.get(Service, int(sid))
if not svc or not svc.is_hub_embed():
return jsonify({"ok": False, "detail": "服务不存在或未标记为中控"}), 400
base = svc.build_origin()
if not next_path or next_path == "/monitor":
p = (svc.path or "/monitor").strip() or "/monitor"
next_path = p if p.startswith("/") else "/" + p
if not base:
return jsonify({"ok": False, "detail": "缺少 base_url"}), 400
base, next_path, err = _resolve_hub_embed_service(body)
if err:
return jsonify({"ok": False, "detail": err}), 400
if not username or not password:
return jsonify({"ok": False, "detail": "缺少中控用户名或密码（可配置 NAV_HUB_USERNAME / NAV_HUB_PASSWORD）"}), 400
payload = json.dumps({"username": username, "password": password}).encode("utf-8")
req = urllib.request.Request(
f"{base}/api/auth/login",
data=payload,
headers={"Content-Type": "application/json", "Accept": "application/json"},
method="POST",
)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
raw = resp.read().decode("utf-8", errors="replace")
except urllib.error.HTTPError as e:
try:
err_body = e.read().decode("utf-8", errors="replace")
detail = json.loads(err_body).get("detail", err_body)
except Exception:
detail = str(e)
return jsonify({"ok": False, "detail": detail or "中控登录失败"}), 401
except Exception as e:
return jsonify({"ok": False, "detail": f"无法连接中控: {e}"}), 502
try:
data = json.loads(raw)
except json.JSONDecodeError:
return jsonify({"ok": False, "detail": "中控返回非 JSON"}), 502
if not data.get("ok"):
return jsonify({"ok": False, "detail": data.get("detail") or "登录失败"}), 401
token = data.get("session_token")
if not token:
return jsonify({"ok": False, "detail": "中控未返回 session_token，请升级云端 hub 后重试"}), 502
token, err = _hub_api_login(base, username, password)
if err:
status = 401 if "登录" in err or "401" in err else 502
return jsonify({"ok": False, "detail": err}), status
q = urlencode({"token": token, "next": next_path})
embed_url = f"{base}/embed-auth?{q}"
return jsonify({"ok": True, "embed_auth_url": embed_url, "next": next_path})
@app.route("/api/embed/hub-instance-url", methods=["POST"])
@csrf.exempt
@login_required
def api_embed_hub_instance_url():
"""
本地导航代签实例 SSO：服务端登录中控后调用 /api/instance/open-url。
用于 LocalNav iframe 直接打开实例（避免中控内再嵌一层 iframe）。
"""
body = request.get_json(silent=True) or {}
exchange_id = str(body.get("exchange_id") or "").strip()
next_path = (body.get("next") or "/").strip() or "/"
if not next_path.startswith("/"):
next_path = "/" + next_path
if not exchange_id:
return jsonify({"ok": False, "detail": "缺少 exchange_id"}), 400
username = (body.get("username") or os.environ.get("NAV_HUB_USERNAME") or "").strip()
password = body.get("password")
if password is None:
password = os.environ.get("NAV_HUB_PASSWORD") or ""
password = str(password)
base, _, err = _resolve_hub_embed_service(body)
if err:
return jsonify({"ok": False, "detail": err}), 400
if not username or not password:
return jsonify({"ok": False, "detail": "缺少中控用户名或密码（可配置 NAV_HUB_USERNAME / NAV_HUB_PASSWORD）"}), 400
token, err = _hub_api_login(base, username, password)
if err:
status = 401 if "登录" in err or "401" in err else 502
return jsonify({"ok": False, "detail": err}), status
url, err = _hub_instance_open_url(
base,
token,
exchange_id,
next_path,
embed=(body.get("embed") or "1").strip().lower() not in ("0", "false", "no"),
)
if err:
return jsonify({"ok": False, "detail": err}), 502
return jsonify({"ok": True, "url": url, "exchange_id": exchange_id, "next": next_path})
# ---------- 分组管理 ----------
@app.route("/admin/groups")
@login_required