User edition: prune docs, license gate, obfuscate core lib.
Keep deploy/basic docs only; integrate sq.bz121.com license client; encrypt strategy/trade/key_monitor/options/hedge_plan for release. Co-authored-by: Cursor <cursoragent@cursor.com>
This commit is contained in:
dekun
@@ -0,0 +1,19 @@
|
||||
"""整机许可：设备绑定、激活码兑换、联网校验。"""
|
||||
|
||||
from lib.license.license_lib import (
|
||||
ensure_license_or_raise,
|
||||
get_device_id,
|
||||
get_license_status,
|
||||
is_license_valid,
|
||||
redeem_code,
|
||||
validate_license,
|
||||
)
|
||||
|
||||
__all__ = [
|
||||
"ensure_license_or_raise",
|
||||
"get_device_id",
|
||||
"get_license_status",
|
||||
"is_license_valid",
|
||||
"redeem_code",
|
||||
"validate_license",
|
||||
]
|
||||
@@ -0,0 +1,115 @@
|
||||
"""FastAPI 中控许可中间件与 /license 页。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from fastapi import FastAPI, Request
|
||||
from fastapi.responses import HTMLResponse, JSONResponse, RedirectResponse
|
||||
|
||||
from lib.license.license_lib import (
|
||||
get_device_id,
|
||||
get_license_status,
|
||||
is_license_valid,
|
||||
redeem_code,
|
||||
validate_license,
|
||||
)
|
||||
|
||||
_TEMPLATE_PATH = Path(__file__).resolve().parent / "templates" / "license.html"
|
||||
|
||||
|
||||
def _allowed_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 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 = {}
|
||||
code = (data.get("code") if isinstance(data, dict) else "") or ""
|
||||
return redeem_code(str(code))
|
||||
|
||||
@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):
|
||||
msg = ""
|
||||
err = ""
|
||||
if request.method == "POST":
|
||||
form = await request.form()
|
||||
code = str(form.get("code") or "").strip()
|
||||
result = redeem_code(code)
|
||||
if result.get("ok"):
|
||||
msg = result.get("message") or "激活成功"
|
||||
else:
|
||||
err = result.get("message") or "激活失败"
|
||||
status = get_license_status()
|
||||
html = _TEMPLATE_PATH.read_text(encoding="utf-8")
|
||||
# 简单替换，避免 Jinja 依赖差异
|
||||
filled = (
|
||||
html.replace("{{ device_id }}", get_device_id())
|
||||
.replace("{{ api_url }}", str(status.get("api_url") or ""))
|
||||
.replace("{{ wechat }}", "dekun03")
|
||||
.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 "未授权")
|
||||
)
|
||||
# Flask 模板用 Jinja；FastAPI 路径用占位符版本
|
||||
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="dekun03",
|
||||
)
|
||||
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 _allowed_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)
|
||||
@@ -0,0 +1,88 @@
|
||||
"""Flask 实例许可门禁与 /license 页。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import os
|
||||
from pathlib import Path
|
||||
|
||||
from flask import Flask, jsonify, redirect, render_template_string, request
|
||||
|
||||
from lib.license.license_lib import (
|
||||
get_device_id,
|
||||
get_license_status,
|
||||
is_license_valid,
|
||||
redeem_code,
|
||||
validate_license,
|
||||
)
|
||||
|
||||
_TEMPLATE_PATH = Path(__file__).resolve().parent / "templates" / "license.html"
|
||||
|
||||
|
||||
def _allowed_path(path: str) -> bool:
|
||||
if path in ("/license", "/api/license/status", "/api/license/redeem", "/api/license/validate", "/health"):
|
||||
return True
|
||||
if path.startswith("/static/"):
|
||||
return True
|
||||
if path.startswith("/favicon"):
|
||||
return True
|
||||
return False
|
||||
|
||||
|
||||
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()
|
||||
return jsonify(redeem_code(code))
|
||||
|
||||
@app.post("/api/license/validate")
|
||||
def _license_validate_api():
|
||||
return jsonify(validate_license(force=True))
|
||||
|
||||
@app.route("/license", methods=["GET", "POST"])
|
||||
def _license_page():
|
||||
msg = ""
|
||||
err = ""
|
||||
if request.method == "POST":
|
||||
code = (request.form.get("code") or "").strip()
|
||||
result = redeem_code(code)
|
||||
if result.get("ok"):
|
||||
msg = result.get("message") or "激活成功"
|
||||
else:
|
||||
err = result.get("message") or "激活失败"
|
||||
status = get_license_status()
|
||||
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="dekun03",
|
||||
)
|
||||
|
||||
@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 _allowed_path(path):
|
||||
return None
|
||||
# hub bridge 内部调用：仍要求已授权（整机许可）
|
||||
if is_license_valid():
|
||||
return None
|
||||
if path.startswith("/api/"):
|
||||
return jsonify({"ok": False, "error": "license_required", "message": "请先激活许可"}), 403
|
||||
return redirect("/license")
|
||||
@@ -0,0 +1,393 @@
|
||||
"""用户端许可客户端：对接 https://sq.bz121.com （crypto_monitor_web）。"""
|
||||
|
||||
from __future__ import annotations
|
||||
|
||||
import hashlib
|
||||
import json
|
||||
import os
|
||||
import platform
|
||||
import threading
|
||||
import time
|
||||
import uuid
|
||||
from datetime import datetime, timezone
|
||||
from pathlib import Path
|
||||
from typing import Any
|
||||
from urllib.error import HTTPError, URLError
|
||||
from urllib.request import Request, urlopen
|
||||
|
||||
from lib.paths import REPO_ROOT
|
||||
|
||||
DEFAULT_API_URL = "https://sq.bz121.com"
|
||||
VALIDATE_INTERVAL_SEC = 3 * 24 * 3600 # 每 3 天
|
||||
DEFAULT_OFFLINE_GRACE_HOURS = 72
|
||||
STATE_FILENAME = "license_state.json"
|
||||
LICENSE_ENV_NAME = "license.env"
|
||||
|
||||
_lock = threading.RLock()
|
||||
_cached_device_id: str | None = None
|
||||
_env_loaded = False
|
||||
|
||||
|
||||
def _load_license_env() -> None:
|
||||
global _env_loaded
|
||||
if _env_loaded:
|
||||
return
|
||||
_env_loaded = True
|
||||
path = REPO_ROOT / LICENSE_ENV_NAME
|
||||
if not path.is_file():
|
||||
return
|
||||
try:
|
||||
text = path.read_text(encoding="utf-8")
|
||||
except OSError:
|
||||
return
|
||||
for line in text.splitlines():
|
||||
raw = line.strip()
|
||||
if not raw or raw.startswith("#") or "=" not in raw:
|
||||
continue
|
||||
key, value = raw.split("=", 1)
|
||||
key = key.strip().lstrip("\ufeff")
|
||||
if not key or key in os.environ:
|
||||
continue
|
||||
os.environ[key] = value.strip().strip('"').strip("'")
|
||||
|
||||
|
||||
def api_base_url() -> str:
|
||||
_load_license_env()
|
||||
url = (os.getenv("LICENSE_API_URL") or DEFAULT_API_URL).strip().rstrip("/")
|
||||
return url or DEFAULT_API_URL
|
||||
|
||||
|
||||
def client_key() -> str:
|
||||
_load_license_env()
|
||||
return (os.getenv("LICENSE_CLIENT_KEY") or "").strip()
|
||||
|
||||
|
||||
def offline_grace_hours() -> float:
|
||||
_load_license_env()
|
||||
raw = (os.getenv("LICENSE_OFFLINE_GRACE_HOURS") or str(DEFAULT_OFFLINE_GRACE_HOURS)).strip()
|
||||
try:
|
||||
return max(0.0, float(raw))
|
||||
except ValueError:
|
||||
return float(DEFAULT_OFFLINE_GRACE_HOURS)
|
||||
|
||||
|
||||
def state_path() -> Path:
|
||||
_load_license_env()
|
||||
override = (os.getenv("LICENSE_STATE_PATH") or "").strip()
|
||||
if override:
|
||||
p = Path(override)
|
||||
if not p.is_absolute():
|
||||
p = REPO_ROOT / p
|
||||
p.parent.mkdir(parents=True, exist_ok=True)
|
||||
return p
|
||||
data = REPO_ROOT / "data"
|
||||
data.mkdir(parents=True, exist_ok=True)
|
||||
return data / STATE_FILENAME
|
||||
|
||||
|
||||
def get_device_id() -> str:
|
||||
"""稳定设备指纹，长度 16–64，供 redeem/validate 使用。"""
|
||||
global _cached_device_id
|
||||
if _cached_device_id:
|
||||
return _cached_device_id
|
||||
parts = [
|
||||
platform.node() or "",
|
||||
platform.system() or "",
|
||||
platform.machine() or "",
|
||||
hex(uuid.getnode()),
|
||||
]
|
||||
try:
|
||||
machine_id = Path("/etc/machine-id")
|
||||
if machine_id.is_file():
|
||||
parts.append(machine_id.read_text(encoding="utf-8").strip())
|
||||
except OSError:
|
||||
pass
|
||||
try:
|
||||
win_id = Path(os.environ.get("SystemRoot", r"C:\Windows")) / "System32" / "drivers" / "etc"
|
||||
parts.append(str(win_id.resolve()))
|
||||
except OSError:
|
||||
pass
|
||||
digest = hashlib.sha256("|".join(parts).encode("utf-8", errors="ignore")).hexdigest()
|
||||
_cached_device_id = digest[:40]
|
||||
return _cached_device_id
|
||||
|
||||
|
||||
def _read_state() -> dict[str, Any]:
|
||||
path = state_path()
|
||||
if not path.is_file():
|
||||
return {}
|
||||
try:
|
||||
data = json.loads(path.read_text(encoding="utf-8"))
|
||||
return data if isinstance(data, dict) else {}
|
||||
except (OSError, json.JSONDecodeError):
|
||||
return {}
|
||||
|
||||
|
||||
def _write_state(data: dict[str, Any]) -> None:
|
||||
path = state_path()
|
||||
path.parent.mkdir(parents=True, exist_ok=True)
|
||||
tmp = path.with_suffix(".tmp")
|
||||
tmp.write_text(json.dumps(data, ensure_ascii=False, indent=2), encoding="utf-8")
|
||||
tmp.replace(path)
|
||||
|
||||
|
||||
def _now_ts() -> float:
|
||||
return time.time()
|
||||
|
||||
|
||||
def _parse_expires_at(value: str | None) -> float | None:
|
||||
if not value:
|
||||
return None
|
||||
s = value.strip()
|
||||
if not s:
|
||||
return None
|
||||
try:
|
||||
if s.endswith("Z"):
|
||||
s = s[:-1] + "+00:00"
|
||||
dt = datetime.fromisoformat(s)
|
||||
if dt.tzinfo is None:
|
||||
# 授权站按北京时间存，无 tz 时按 UTC+8
|
||||
from datetime import timedelta, timezone as _tz
|
||||
|
||||
dt = dt.replace(tzinfo=_tz(timedelta(hours=8)))
|
||||
return dt.timestamp()
|
||||
except ValueError:
|
||||
return None
|
||||
|
||||
|
||||
def _http_json(method: str, path: str, body: dict[str, Any]) -> dict[str, Any]:
|
||||
key = client_key()
|
||||
if not key:
|
||||
return {"ok": False, "message": "未配置 LICENSE_CLIENT_KEY（与授权站 CLIENT_API_KEY 一致）"}
|
||||
url = f"{api_base_url()}{path}"
|
||||
payload = json.dumps(body).encode("utf-8")
|
||||
req = Request(
|
||||
url,
|
||||
data=payload,
|
||||
method=method.upper(),
|
||||
headers={
|
||||
"Content-Type": "application/json",
|
||||
"X-Client-Key": key,
|
||||
"User-Agent": "crypto_monitor_user/license",
|
||||
},
|
||||
)
|
||||
try:
|
||||
with urlopen(req, timeout=20) as resp:
|
||||
raw = resp.read().decode("utf-8", errors="replace")
|
||||
data = json.loads(raw) if raw else {}
|
||||
return data if isinstance(data, dict) else {"ok": False, "message": "响应格式错误"}
|
||||
except HTTPError as e:
|
||||
try:
|
||||
detail = e.read().decode("utf-8", errors="replace")
|
||||
except Exception:
|
||||
detail = str(e)
|
||||
return {"ok": False, "message": f"HTTP {e.code}: {detail[:200]}", "network_error": True}
|
||||
except (URLError, TimeoutError, OSError) as e:
|
||||
return {"ok": False, "message": f"无法连接授权服务: {e}", "network_error": True}
|
||||
except json.JSONDecodeError:
|
||||
return {"ok": False, "message": "授权服务返回非 JSON", "network_error": True}
|
||||
|
||||
|
||||
def redeem_code(code: str) -> dict[str, Any]:
|
||||
code = (code or "").strip().upper().replace(" ", "")
|
||||
if len(code) < 8:
|
||||
return {"ok": False, "message": "激活码无效"}
|
||||
device_id = get_device_id()
|
||||
result = _http_json("POST", "/v1/redeem", {"device_id": device_id, "code": code})
|
||||
if not result.get("ok"):
|
||||
return {
|
||||
"ok": False,
|
||||
"message": result.get("message") or "兑换失败",
|
||||
"network_error": bool(result.get("network_error")),
|
||||
}
|
||||
now = _now_ts()
|
||||
with _lock:
|
||||
state = {
|
||||
"device_id": device_id,
|
||||
"subscription_id": result.get("subscription_id"),
|
||||
"plan": result.get("plan"),
|
||||
"expires_at": result.get("expires_at"),
|
||||
"last_ok_at": now,
|
||||
"last_validate_at": now,
|
||||
"last_reason": "",
|
||||
}
|
||||
_write_state(state)
|
||||
return {
|
||||
"ok": True,
|
||||
"message": result.get("message") or "激活成功",
|
||||
"subscription_id": result.get("subscription_id"),
|
||||
"plan": result.get("plan"),
|
||||
"expires_at": result.get("expires_at"),
|
||||
"device_id": device_id,
|
||||
}
|
||||
|
||||
|
||||
def validate_license(*, force: bool = False) -> dict[str, Any]:
|
||||
"""联网校验；更新本地状态。force=True 忽略 3 天间隔。"""
|
||||
with _lock:
|
||||
state = _read_state()
|
||||
sub_id = (state.get("subscription_id") or "").strip()
|
||||
if not sub_id:
|
||||
return {"ok": True, "valid": False, "reason": "not_activated", "message": "尚未激活"}
|
||||
|
||||
last_v = float(state.get("last_validate_at") or 0)
|
||||
if not force and last_v and (_now_ts() - last_v) < VALIDATE_INTERVAL_SEC:
|
||||
return get_license_status()
|
||||
|
||||
device_id = get_device_id()
|
||||
result = _http_json(
|
||||
"POST",
|
||||
"/v1/validate",
|
||||
{"device_id": device_id, "subscription_id": sub_id},
|
||||
)
|
||||
now = _now_ts()
|
||||
if result.get("network_error"):
|
||||
# 断网：在宽限期内沿用本地到期日
|
||||
status = get_license_status(skip_remote=True)
|
||||
status["message"] = result.get("message") or "网络异常，使用本地宽限"
|
||||
status["network_error"] = True
|
||||
with _lock:
|
||||
st = _read_state()
|
||||
st["last_network_error_at"] = now
|
||||
st["last_reason"] = "network_error"
|
||||
_write_state(st)
|
||||
return status
|
||||
|
||||
valid = bool(result.get("valid"))
|
||||
reason = (result.get("reason") or "").strip()
|
||||
with _lock:
|
||||
st = _read_state()
|
||||
st["device_id"] = device_id
|
||||
st["last_validate_at"] = now
|
||||
if valid:
|
||||
st["last_ok_at"] = now
|
||||
st["last_reason"] = ""
|
||||
if result.get("expires_at"):
|
||||
st["expires_at"] = result.get("expires_at")
|
||||
if result.get("plan"):
|
||||
st["plan"] = result.get("plan")
|
||||
else:
|
||||
st["last_reason"] = reason or "invalid"
|
||||
if reason in ("expired", "device_revoked", "subscription_not_found", "subscription_inactive"):
|
||||
# 明确失效：保留字段便于页面展示，但 valid=False
|
||||
pass
|
||||
_write_state(st)
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"valid": valid,
|
||||
"reason": reason if not valid else "",
|
||||
"expires_at": result.get("expires_at") or state.get("expires_at"),
|
||||
"plan": result.get("plan") or state.get("plan"),
|
||||
"device_id": device_id,
|
||||
"subscription_id": sub_id,
|
||||
"message": "有效" if valid else (reason or "无效"),
|
||||
}
|
||||
|
||||
|
||||
def get_license_status(*, skip_remote: bool = False) -> dict[str, Any]:
|
||||
"""本地状态（可触发到期判断）；不主动联网除非 skip_remote=False 且到期需刷新。"""
|
||||
with _lock:
|
||||
state = _read_state()
|
||||
device_id = get_device_id()
|
||||
sub_id = (state.get("subscription_id") or "").strip()
|
||||
if not sub_id:
|
||||
return {
|
||||
"ok": True,
|
||||
"valid": False,
|
||||
"reason": "not_activated",
|
||||
"message": "尚未激活",
|
||||
"device_id": device_id,
|
||||
"api_url": api_base_url(),
|
||||
}
|
||||
|
||||
exp_ts = _parse_expires_at(state.get("expires_at"))
|
||||
now = _now_ts()
|
||||
if exp_ts is not None and now > exp_ts:
|
||||
return {
|
||||
"ok": True,
|
||||
"valid": False,
|
||||
"reason": "expired",
|
||||
"message": "许可已过期",
|
||||
"expires_at": state.get("expires_at"),
|
||||
"plan": state.get("plan"),
|
||||
"device_id": device_id,
|
||||
"subscription_id": sub_id,
|
||||
"api_url": api_base_url(),
|
||||
}
|
||||
|
||||
reason = (state.get("last_reason") or "").strip()
|
||||
if reason in ("expired", "device_revoked", "subscription_not_found", "subscription_inactive"):
|
||||
return {
|
||||
"ok": True,
|
||||
"valid": False,
|
||||
"reason": reason,
|
||||
"message": reason,
|
||||
"expires_at": state.get("expires_at"),
|
||||
"plan": state.get("plan"),
|
||||
"device_id": device_id,
|
||||
"subscription_id": sub_id,
|
||||
"api_url": api_base_url(),
|
||||
}
|
||||
|
||||
last_ok = float(state.get("last_ok_at") or 0)
|
||||
grace = offline_grace_hours() * 3600
|
||||
if last_ok and grace > 0 and (now - last_ok) > grace:
|
||||
# 超过离线宽限且长时间未成功校验
|
||||
last_v = float(state.get("last_validate_at") or 0)
|
||||
if last_v and (now - last_v) >= VALIDATE_INTERVAL_SEC and not skip_remote:
|
||||
return validate_license(force=True)
|
||||
if (now - last_ok) > grace:
|
||||
return {
|
||||
"ok": True,
|
||||
"valid": False,
|
||||
"reason": "offline_grace_exceeded",
|
||||
"message": "超过离线宽限期，请联网校验",
|
||||
"expires_at": state.get("expires_at"),
|
||||
"plan": state.get("plan"),
|
||||
"device_id": device_id,
|
||||
"subscription_id": sub_id,
|
||||
"api_url": api_base_url(),
|
||||
}
|
||||
|
||||
return {
|
||||
"ok": True,
|
||||
"valid": True,
|
||||
"reason": "",
|
||||
"message": "已授权",
|
||||
"expires_at": state.get("expires_at"),
|
||||
"plan": state.get("plan"),
|
||||
"device_id": device_id,
|
||||
"subscription_id": sub_id,
|
||||
"last_ok_at": last_ok or None,
|
||||
"api_url": api_base_url(),
|
||||
}
|
||||
|
||||
|
||||
def is_license_valid() -> bool:
|
||||
status = get_license_status()
|
||||
if status.get("valid"):
|
||||
# 到期需周期性联网
|
||||
last_v = 0.0
|
||||
with _lock:
|
||||
st = _read_state()
|
||||
last_v = float(st.get("last_validate_at") or 0)
|
||||
if last_v and (_now_ts() - last_v) >= VALIDATE_INTERVAL_SEC:
|
||||
status = validate_license(force=True)
|
||||
return bool(status.get("valid"))
|
||||
# 未激活或已失效：尝试若有 subscription 则强制校验一次（避免本地过期标志陈旧）
|
||||
with _lock:
|
||||
st = _read_state()
|
||||
if (st.get("subscription_id") or "").strip() and status.get("reason") not in (
|
||||
"not_activated",
|
||||
"expired",
|
||||
"device_revoked",
|
||||
):
|
||||
status = validate_license(force=True)
|
||||
return bool(status.get("valid"))
|
||||
return False
|
||||
|
||||
|
||||
def ensure_license_or_raise() -> None:
|
||||
if not is_license_valid():
|
||||
raise RuntimeError("license_required")
|
||||
@@ -0,0 +1,100 @@
|
||||
<!DOCTYPE html>
|
||||
<html lang="zh-CN">
|
||||
<head>
|
||||
<meta charset="UTF-8">
|
||||
<meta name="viewport" content="width=device-width, initial-scale=1">
|
||||
<title>软件授权</title>
|
||||
<style>
|
||||
* { box-sizing: border-box; margin: 0; padding: 0; }
|
||||
body {
|
||||
min-height: 100vh;
|
||||
display: flex;
|
||||
align-items: center;
|
||||
justify-content: center;
|
||||
padding: 24px;
|
||||
background: #0a0a10;
|
||||
color: #e8e8f0;
|
||||
font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, sans-serif;
|
||||
}
|
||||
.box {
|
||||
width: 100%;
|
||||
max-width: 440px;
|
||||
background: #12121a;
|
||||
border: 1px solid #2a2a3a;
|
||||
border-radius: 14px;
|
||||
padding: 28px 24px;
|
||||
}
|
||||
h1 {
|
||||
font-size: 1.35rem;
|
||||
margin-bottom: 8px;
|
||||
background: linear-gradient(90deg, #4cc2ff, #7b42ff);
|
||||
-webkit-background-clip: text;
|
||||
-webkit-text-fill-color: transparent;
|
||||
}
|
||||
.muted { color: #9aa0b4; font-size: 0.9rem; margin-bottom: 18px; line-height: 1.5; }
|
||||
.row { margin-bottom: 14px; }
|
||||
label { display: block; font-size: 0.85rem; color: #b8bfd4; margin-bottom: 6px; }
|
||||
.id {
|
||||
word-break: break-all;
|
||||
background: #0c0c14;
|
||||
border: 1px solid #2a2a3a;
|
||||
border-radius: 8px;
|
||||
padding: 10px 12px;
|
||||
font-family: ui-monospace, Consolas, monospace;
|
||||
font-size: 0.85rem;
|
||||
}
|
||||
input[type=text] {
|
||||
width: 100%;
|
||||
padding: 10px 12px;
|
||||
border-radius: 8px;
|
||||
border: 1px solid #2a2a3a;
|
||||
background: #0c0c14;
|
||||
color: #fff;
|
||||
font-size: 1rem;
|
||||
}
|
||||
button {
|
||||
width: 100%;
|
||||
margin-top: 8px;
|
||||
padding: 11px 14px;
|
||||
border: 0;
|
||||
border-radius: 8px;
|
||||
background: linear-gradient(90deg, #3a8dff, #6b4dff);
|
||||
color: #fff;
|
||||
font-size: 1rem;
|
||||
cursor: pointer;
|
||||
}
|
||||
.ok { color: #5ddea8; margin: 10px 0; font-size: 0.9rem; }
|
||||
.err { color: #ff7b8a; margin: 10px 0; font-size: 0.9rem; }
|
||||
.meta { margin-top: 16px; font-size: 0.85rem; color: #9aa0b4; line-height: 1.6; }
|
||||
a { color: #7ec8ff; }
|
||||
</style>
|
||||
</head>
|
||||
<body>
|
||||
<div class="box">
|
||||
<h1>软件授权</h1>
|
||||
<p class="muted">复制下方设备 ID，联系微信 <strong>{{ wechat }}</strong> 购买激活码后粘贴兑换。授权站：{{ api_url }}</p>
|
||||
|
||||
{% if message %}<p class="ok">{{ message }}</p>{% endif %}
|
||||
{% if error %}<p class="err">{{ error }}</p>{% endif %}
|
||||
|
||||
<div class="row">
|
||||
<label>设备 ID</label>
|
||||
<div class="id" id="deviceId">{{ device_id }}</div>
|
||||
</div>
|
||||
|
||||
<form method="post" action="/license">
|
||||
<div class="row">
|
||||
<label>激活码</label>
|
||||
<input type="text" name="code" placeholder="粘贴激活码" autocomplete="off" required>
|
||||
</div>
|
||||
<button type="submit">兑换激活</button>
|
||||
</form>
|
||||
|
||||
<div class="meta">
|
||||
状态：{% if status.valid %}已授权{% else %}未授权{% endif %}
|
||||
（{{ status.message or status.reason or '—' }}）<br>
|
||||
套餐：{{ status.plan or '—' }} 到期：{{ status.expires_at or '—' }}
|
||||
</div>
|
||||
</div>
|
||||
</body>
|
||||
</html>
|
||||
Reference in New Issue
Block a user