已授权

"""FastAPI 中控许可中间件与 /license 页。""" from __future__ import annotations import html as html_lib import os from pathlib import Path from fastapi import FastAPI, Request from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse, Response from lib.license.license_lib import ( get_device_id, get_license_status, is_license_valid, redeem_code, usdt_address, usdt_network, validate_license, wechat_id, ) _TEMPLATE_PATH = Path(__file__).resolve().parent / "templates" / "license.html" _POST_LICENSE_TARGET = "/login" def _license_public_path(path: str) -> bool: """未授权时仍可访问的路径（授权页 / 接口 / 静态资源）。""" if path in ( "/license", "/api/license/status", "/api/license/redeem", "/api/license/validate", "/health", ): return True if path.startswith("/assets/") or path.startswith("/static/"): return True if path.startswith("/favicon"): return True return False def _license_manage_requested(request: Request) -> bool: """已授权时默认禁止进入 /license；续费/换机用 ?renew=1。""" q = request.query_params return (q.get("renew") or q.get("manage") or "").strip().lower() in ( "1", "true", "yes", "on", ) def _locally_licensed() -> bool: return bool(get_license_status(skip_remote=True).get("valid")) def _leave_license_response(target: str = _POST_LICENSE_TARGET) -> Response: safe = html_lib.escape(target, quote=True) body = f""" 已授权

已激活，正在进入系统… 点击进入

""" return Response( content=body, status_code=302, headers={ "Location": target, "Cache-Control": "no-store, no-cache, must-revalidate", "Content-Type": "text/html; charset=utf-8", }, ) def install_license_middleware(app: FastAPI) -> None: @app.get("/health") async def _license_health(): st = get_license_status(skip_remote=True) return {"ok": True, "license_valid": bool(st.get("valid")), "service": "manual_trading_hub"} @app.get("/api/license/status") async def _license_status_api(): return get_license_status() @app.post("/api/license/redeem") async def _license_redeem_api(request: Request): try: data = await request.json() except Exception: data = {} if not isinstance(data, dict): data = {} code = str(data.get("code") or "").strip() ckey = str(data.get("client_api_key") or "").strip() return redeem_code(code, client_api_key=ckey or None) @app.post("/api/license/validate") async def _license_validate_api(): return validate_license(force=True) @app.api_route("/license", methods=["GET", "POST"]) async def _license_page(request: Request): # 已授权：默认不可再进授权页（续费/换机：/license?renew=1） if _locally_licensed() and request.method == "GET" and not _license_manage_requested(request): return _leave_license_response() msg = "" err = "" if request.method == "POST": form = await request.form() code = str(form.get("code") or "").strip() ckey = str(form.get("client_api_key") or "").strip() result = redeem_code(code, client_api_key=ckey or None) if result.get("ok"): return _leave_license_response() err = result.get("message") or "激活失败" status = get_license_status(skip_remote=True) html = _TEMPLATE_PATH.read_text(encoding="utf-8") filled = ( html.replace("{{ device_id }}", get_device_id()) .replace("{{ api_url }}", str(status.get("api_url") or "")) .replace("{{ wechat }}", wechat_id()) .replace("{{ usdt_address }}", usdt_address()) .replace("{{ usdt_network }}", usdt_network()) .replace("{{ message }}", msg) .replace("{{ error }}", err) .replace("{{ status_message }}", str(status.get("message") or "")) .replace("{{ expires_at }}", str(status.get("expires_at") or "—")) .replace("{{ plan }}", str(status.get("plan") or "—")) .replace("{{ valid_text }}", "已授权" if status.get("valid") else "未授权") ) if "{%" in filled or "{{" in filled: from jinja2 import Template filled = Template(html).render( device_id=get_device_id(), status=status, message=msg, error=err, api_url=status.get("api_url") or "", wechat=wechat_id(), usdt_address=usdt_address(), usdt_network=usdt_network(), ) return HTMLResponse(filled) @app.middleware("http") async def _license_http_middleware(request: Request, call_next): if os.getenv("LICENSE_DISABLED", "").strip().lower() in ("1", "true", "yes", "on"): return await call_next(request) path = request.url.path or "/" if _license_public_path(path): return await call_next(request) if is_license_valid(): return await call_next(request) if path.startswith("/api/"): return JSONResponse( {"ok": False, "error": "license_required", "message": "请先激活许可"}, status_code=403, ) return RedirectResponse(url="/license", status_code=302)