dekun
160 lines
5.1 KiB
Python
160 lines
5.1 KiB
Python
"""Flask 实例许可门禁与 /license 页。"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import html as html_lib
|
||
import os
|
||
from pathlib import Path
|
||
|
||
from flask import Flask, Response, jsonify, redirect, render_template_string, request
|
||
|
||
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"
|
||
|
||
# 已授权离开授权页时优先去登录页（有完整 UI）；已登录会再跳进系统
|
||
_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",
|
||
"/api/admin/health",
|
||
):
|
||
return True
|
||
if path.startswith("/static/"):
|
||
return True
|
||
if path.startswith("/favicon"):
|
||
return True
|
||
return False
|
||
|
||
|
||
|
||
def _license_manage_requested() -> bool:
|
||
"""已授权时默认禁止进入 /license；续费/换机用 ?renew=1。"""
|
||
return (request.args.get("renew") or request.args.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:
|
||
"""已授权离开 /license：302 + 可见提示，避免浏览器停在全黑空页。"""
|
||
safe = html_lib.escape(target, quote=True)
|
||
body = f"""<!DOCTYPE html>
|
||
<html lang="zh-CN">
|
||
<head>
|
||
<meta charset="UTF-8">
|
||
<meta http-equiv="refresh" content="0;url={safe}">
|
||
<title>已授权</title>
|
||
<style>
|
||
body {{
|
||
margin: 0; min-height: 100vh; display: flex; align-items: center; justify-content: center;
|
||
background: #0a0a10; color: #e8e8f0; font-family: sans-serif;
|
||
}}
|
||
a {{ color: #7ec8ff; }}
|
||
</style>
|
||
</head>
|
||
<body>
|
||
<p>已激活，正在进入系统… <a href="{safe}">点击进入</a></p>
|
||
<script>location.replace({target!r});</script>
|
||
</body>
|
||
</html>
|
||
"""
|
||
return Response(
|
||
body,
|
||
status=302,
|
||
headers={
|
||
"Location": target,
|
||
"Cache-Control": "no-store, no-cache, must-revalidate",
|
||
"Content-Type": "text/html; charset=utf-8",
|
||
},
|
||
)
|
||
|
||
|
||
def install_license_gate(app: Flask) -> None:
|
||
"""注册 /license 与 before_request 门禁。三所 Flask 共用。"""
|
||
|
||
@app.get("/health")
|
||
def _license_health():
|
||
st = get_license_status(skip_remote=True)
|
||
return jsonify({"ok": True, "license_valid": bool(st.get("valid"))})
|
||
|
||
@app.get("/api/license/status")
|
||
def _license_status_api():
|
||
return jsonify(get_license_status())
|
||
|
||
@app.post("/api/license/redeem")
|
||
def _license_redeem_api():
|
||
data = request.get_json(silent=True) or {}
|
||
code = (data.get("code") or request.form.get("code") or "").strip()
|
||
ckey = (data.get("client_api_key") or request.form.get("client_api_key") or "").strip()
|
||
return jsonify(redeem_code(code, client_api_key=ckey or None))
|
||
|
||
@app.post("/api/license/validate")
|
||
def _license_validate_api():
|
||
return jsonify(validate_license(force=True))
|
||
|
||
@app.route("/license", methods=["GET", "POST"])
|
||
def _license_page():
|
||
# 已授权：默认不可再进授权页（续费/换机：/license?renew=1）
|
||
if _locally_licensed() and request.method == "GET" and not _license_manage_requested():
|
||
return _leave_license_response()
|
||
|
||
msg = ""
|
||
err = ""
|
||
if request.method == "POST":
|
||
code = (request.form.get("code") or "").strip()
|
||
ckey = (request.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")
|
||
return render_template_string(
|
||
html,
|
||
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(),
|
||
)
|
||
|
||
@app.before_request
|
||
def _license_before_request():
|
||
if os.getenv("LICENSE_DISABLED", "").strip().lower() in ("1", "true", "yes", "on"):
|
||
return None
|
||
path = request.path or "/"
|
||
if _license_public_path(path):
|
||
return None
|
||
if is_license_valid():
|
||
return None
|
||
if path.startswith("/api/"):
|
||
return jsonify({"ok": False, "error": "license_required", "message": "请先激活许可"}), 403
|
||
return redirect("/license")
|