657 lines
24 KiB
Python
657 lines
24 KiB
Python
# license_utils.py
|
||
"""RSA 验签许可证模块 —— 嵌入 exe，验签 + 每日巡检 + 防改系统时间。
|
||
|
||
用法（在 main.py 或主窗口 __init__ 中调用一次即可）:
|
||
from license_utils import check_license, start_license_watchdog
|
||
|
||
check_license() # 启动时验签（失败则抛异常退出）
|
||
start_license_watchdog(interval_minutes=1440) # 后台每天巡检
|
||
"""
|
||
|
||
import json
|
||
import base64
|
||
import os
|
||
import sys
|
||
import threading
|
||
import time as _time_module
|
||
import datetime
|
||
from pathlib import Path
|
||
|
||
import requests
|
||
|
||
from cryptography.hazmat.primitives.asymmetric import rsa, padding
|
||
from cryptography.hazmat.primitives import hashes, serialization
|
||
from cryptography.exceptions import InvalidSignature
|
||
|
||
# ============================================================
|
||
# 公钥（编译进 exe，可公开）—— 与 ControlPanel 使用的签名私钥配对
|
||
# 公钥更新由受控的发布流程完成，客户端不包含任何签发能力。
|
||
# ============================================================
|
||
# {{LICENSE_PUBLIC_KEY_START}}
|
||
PUBLIC_KEY_PEM = """-----BEGIN PUBLIC KEY-----
|
||
MIIBojANBgkqhkiG9w0BAQEFAAOCAY8AMIIBigKCAYEArQ4l2ePnl+p9yuUWcuOK
|
||
CQ4bzixyJKh8tPSVMwqI7u02v9qdrIPu5j3T0tl+qCNTLMP6WHd6s09M7bWuVTnU
|
||
220ljrdBb8opIzNGxTTri1k3JqMI95ljLwdEB6vp+ISyrdFKG4o+B+hDDvxkbyGH
|
||
2vKBG71Wrws4ujOEI1H8MDWDRMyGrHFQZZk1Sz6WkgWT+yjoZ8L0K3o0afIYw8F9
|
||
J50aaRQ9fvNW4EB+Pa5Yy5DQrNIs1smMVDpegdYt5uUMwEnfoS6Y6l98Gz7ljZ5n
|
||
6/WVaFb55XquAwsF/zq6oDfKBrAOBqzT2YYZklr8swlKKIJ3ExA1sd/dxhfZhidi
|
||
pqCvye6+cYa5GTRu9knzBsVPdzzhQC5AqKUuPglVJV8dQPfH7Nb7EP5wvNSgzpLT
|
||
9wsIoZXm9GXD0hApHvobSiZnpqY5g9InV7fQZyold2zFhHWpDieNjX0844gQafnH
|
||
Ue6JWRU4j3Wg37WDPbwkO3tQba2jbUQsLYomLGuohfkVAgMBAAE=
|
||
-----END PUBLIC KEY-----"""
|
||
# {{LICENSE_PUBLIC_KEY_END}}
|
||
|
||
# 许可证文件相对路径
|
||
LICENSE_FILE = "license.lic"
|
||
|
||
# 巡检间隔（分钟）
|
||
DEFAULT_CHECK_INTERVAL = 5
|
||
ONLINE_CHECK_TIMEOUT_SECONDS = 5
|
||
DEFAULT_OFFLINE_HOURS = 72
|
||
|
||
# 过期后宽限期（小时），给用户保存工作的时间
|
||
GRACE_PERIOD_HOURS = 2
|
||
|
||
# ============================================================
|
||
# 内部状态
|
||
# ============================================================
|
||
_startup_monotonic = _time_module.monotonic() # 软件启动时刻（不受系统时间影响）
|
||
_last_check_result = None
|
||
_verified_license = None
|
||
_verified_license_lock = threading.RLock()
|
||
_watchdog_started = False
|
||
_watchdog_lock = threading.Lock()
|
||
_last_online_success_monotonic = None
|
||
_on_expired_callback = None # 过期回调，可由外部设置
|
||
_on_grace_callback = None # 缓冲期回调
|
||
_on_log_callback = None # 日志回调，供 UI 状态栏显示
|
||
_warning_shown_states = {
|
||
"expiring_today": False, # 到期当天警告是否已弹窗
|
||
"expiring_soon": False, # 即将到期（30天内）警告是否已弹窗
|
||
"expired_grace": False, # 过期缓冲期警告是否已弹窗
|
||
}
|
||
|
||
# ============================================================
|
||
# GUI 线程安全工具
|
||
# ============================================================
|
||
def _invoke_on_qt_thread(func):
|
||
"""在 Qt 主线程中安全执行 func。
|
||
|
||
使用 QTimer.singleShot 将回调排队到主线程事件循环。
|
||
关键：必须传入 QApplication 作为 context（receiver），否则 QTimer 会被调度到
|
||
当前线程（watchdog 是 daemon 线程，没有 Qt event loop），导致静默永不触发。
|
||
"""
|
||
try:
|
||
from PySide6.QtCore import QTimer
|
||
from PySide6.QtWidgets import QApplication
|
||
app = QApplication.instance()
|
||
if app is not None:
|
||
QTimer.singleShot(0, app, func)
|
||
return True
|
||
except Exception:
|
||
pass
|
||
|
||
# 兜底：没有 QApplication 时直接调用（GUI 还未初始化）
|
||
try:
|
||
func()
|
||
except Exception:
|
||
pass
|
||
return False
|
||
|
||
|
||
def _show_message_box(icon, title, message):
|
||
"""线程安全地显示 QMessageBox。
|
||
|
||
Args:
|
||
icon: 'critical', 'warning', 'information'
|
||
title: 弹窗标题
|
||
message: 弹窗内容
|
||
"""
|
||
def _show():
|
||
try:
|
||
from PySide6.QtWidgets import QApplication, QMessageBox
|
||
app = QApplication.instance()
|
||
if app is None:
|
||
return
|
||
if icon == 'critical':
|
||
QMessageBox.critical(None, title, message)
|
||
elif icon == 'warning':
|
||
QMessageBox.warning(None, title, message)
|
||
else:
|
||
QMessageBox.information(None, title, message)
|
||
except Exception:
|
||
pass
|
||
|
||
_invoke_on_qt_thread(_show)
|
||
|
||
|
||
# ============================================================
|
||
# 验签核心
|
||
# ============================================================
|
||
def _load_public_key():
|
||
"""从内嵌的 PEM 加载公钥"""
|
||
return serialization.load_pem_public_key(PUBLIC_KEY_PEM.encode())
|
||
|
||
|
||
def _validate_device_id(device_id):
|
||
"""确保新许可证中的目录键只能是 ``company/line``。"""
|
||
if not isinstance(device_id, str):
|
||
raise ValueError("许可证 device_id 必须是字符串")
|
||
segments = device_id.split("/")
|
||
if len(segments) != 2 or any(
|
||
not segment or segment in (".", "..") or "\\" in segment
|
||
for segment in segments):
|
||
raise ValueError("许可证 device_id 必须是 company/production-line 格式")
|
||
|
||
|
||
def _validate_payload(payload):
|
||
if not isinstance(payload, dict):
|
||
raise ValueError("许可证内容必须是对象")
|
||
|
||
new_fields = ("license_id", "company_id", "production_line_id", "device_id")
|
||
has_new_format = any(field in payload for field in new_fields)
|
||
if has_new_format:
|
||
missing = [field for field in new_fields
|
||
if not isinstance(payload.get(field), str) or not payload[field].strip()]
|
||
if missing:
|
||
raise ValueError(f"新许可证缺少标识字段: {', '.join(missing)}")
|
||
_validate_device_id(payload["device_id"])
|
||
else:
|
||
_log("旧许可证不支持在线撤销")
|
||
|
||
if not isinstance(payload.get("customer"), str) or not payload["customer"].strip():
|
||
raise ValueError("许可证缺少客户信息")
|
||
return has_new_format
|
||
|
||
|
||
def verify_license(lic_path=None):
|
||
"""验证许可证签名 + 有效期。
|
||
|
||
Args:
|
||
lic_path: 许可证文件路径，默认 exe 同级目录下的 license.lic
|
||
|
||
Returns:
|
||
dict: 许可证 payload（customer, expiry, issued 等）
|
||
|
||
Raises:
|
||
FileNotFoundError: 许可证文件不存在
|
||
InvalidSignature: 签名不匹配（伪造/篡改）
|
||
RuntimeError: 许可证已过期
|
||
ValueError: 许可证格式错误
|
||
"""
|
||
if lic_path is None:
|
||
# PyInstaller 打包后 sys.executable 是 exe 路径
|
||
exe_dir = Path(sys.executable).parent if getattr(sys, 'frozen', False) else Path.cwd()
|
||
lic_path = exe_dir / LICENSE_FILE
|
||
|
||
if not os.path.exists(lic_path):
|
||
raise FileNotFoundError(f"许可证文件不存在: {lic_path}")
|
||
|
||
with open(lic_path, "r", encoding="utf-8") as f:
|
||
raw = f.read().strip()
|
||
|
||
# 解析: payload_base64 | signature_base64
|
||
if "|" not in raw:
|
||
raise ValueError("许可证格式错误")
|
||
|
||
payload_b64, signature_b64 = raw.split("|", 1)
|
||
signature = base64.b64decode(signature_b64)
|
||
|
||
# RSA-PSS SHA256 验签
|
||
try:
|
||
pub_key = _load_public_key()
|
||
pub_key.verify(
|
||
signature,
|
||
payload_b64.encode(), # 签名的是 base64 字符串本身
|
||
padding.PSS(
|
||
mgf=padding.MGF1(hashes.SHA256()),
|
||
salt_length=padding.PSS.MAX_LENGTH,
|
||
),
|
||
hashes.SHA256(),
|
||
)
|
||
except InvalidSignature:
|
||
raise InvalidSignature("许可证签名验证失败：文件可能被篡改")
|
||
|
||
# 解析载荷
|
||
try:
|
||
payload = json.loads(base64.b64decode(payload_b64))
|
||
except Exception:
|
||
raise ValueError("许可证内容解析失败")
|
||
|
||
_validate_payload(payload)
|
||
|
||
# 有效期检查（精确到小时）
|
||
expiry_str = payload.get("expiry")
|
||
if not expiry_str:
|
||
raise ValueError("许可证缺少过期时间")
|
||
|
||
# 兼容旧格式 YYYY-MM-DD（视为当天 23:59）
|
||
if len(expiry_str) == 10:
|
||
expiry = datetime.datetime.strptime(expiry_str + " 23:59", "%Y-%m-%d %H:%M")
|
||
else:
|
||
expiry = datetime.datetime.strptime(expiry_str, "%Y-%m-%d %H:%M")
|
||
|
||
trusted_now = _get_trusted_now()
|
||
|
||
if trusted_now > expiry:
|
||
gap_days = (trusted_now - expiry).days
|
||
uptime_days = _get_uptime_days()
|
||
|
||
# 反时钟篡改：需同时满足两个条件才怀疑用户拨慢系统时间
|
||
# ① 软件运行不到 1 天（刚启动）
|
||
# ② 过期时间差超过 7 天（差距巨大 → 文件 mtime 暴露了真实时间）
|
||
# 缺失任一条件 → 真过期，直接报错：
|
||
# - gap 小（几分钟~几小时）→ 刚过期，正常报错
|
||
# - 软件跑了很久 → 正常使用中过期，正常报错
|
||
if uptime_days < 1 and gap_days > 7:
|
||
issued_str = payload.get("issued", expiry_str)
|
||
if len(issued_str) == 10:
|
||
issued = datetime.datetime.strptime(issued_str + " 00:00", "%Y-%m-%d %H:%M")
|
||
else:
|
||
issued = datetime.datetime.strptime(issued_str, "%Y-%m-%d %H:%M")
|
||
estimated_now = issued + datetime.timedelta(days=uptime_days)
|
||
if estimated_now <= expiry:
|
||
return payload
|
||
|
||
raise ExpiredError(
|
||
f"许可证已过期 (到期: {expiry_str})",
|
||
expiry=expiry,
|
||
)
|
||
|
||
return payload
|
||
|
||
|
||
def get_verified_license():
|
||
"""返回本进程已通过本地验证的许可证载荷，尚未验证时返回 ``None``。"""
|
||
with _verified_license_lock:
|
||
return dict(_verified_license) if _verified_license is not None else None
|
||
|
||
|
||
def _is_new_license(payload):
|
||
return all(payload.get(field) for field in (
|
||
"license_id", "company_id", "production_line_id", "device_id"))
|
||
|
||
|
||
def _api_url():
|
||
base_url = os.environ.get(
|
||
"REINLOOP_SERVER_URL", "http://ReinLoop.dominatedconvergence.com"
|
||
).rstrip("/")
|
||
return os.environ.get("REINLOOP_API_URL", f"{base_url}/api")
|
||
|
||
|
||
def _offline_limit_seconds():
|
||
try:
|
||
hours = float(os.environ.get("REINLOOP_LICENSE_OFFLINE_HOURS", DEFAULT_OFFLINE_HOURS))
|
||
except ValueError:
|
||
hours = DEFAULT_OFFLINE_HOURS
|
||
return max(0, hours) * 3600
|
||
|
||
|
||
def validate_license_online(payload):
|
||
"""验证可撤销许可证的在线状态。
|
||
|
||
网络故障只在超过离线时限后失效；服务端明确拒绝则立即返回 ExpiredError。
|
||
"""
|
||
global _last_online_success_monotonic
|
||
if not _is_new_license(payload):
|
||
return
|
||
|
||
try:
|
||
response = requests.post(_api_url(), json={
|
||
"type": "validateLicense",
|
||
"licenseId": payload["license_id"],
|
||
"deviceId": payload["device_id"],
|
||
}, timeout=ONLINE_CHECK_TIMEOUT_SECONDS)
|
||
response.raise_for_status()
|
||
result = response.json()
|
||
except (requests.RequestException, ValueError) as exc:
|
||
now = _time_module.monotonic()
|
||
if _last_online_success_monotonic is None:
|
||
_last_online_success_monotonic = _startup_monotonic
|
||
if now - _last_online_success_monotonic > _offline_limit_seconds():
|
||
raise ExpiredError("许可证在线校验超过离线宽限期") from exc
|
||
_log(f"许可证在线校验暂不可用: {exc}")
|
||
return
|
||
|
||
status = result.get("status")
|
||
if not result.get("success") or not result.get("valid") or status != "active":
|
||
reason = status or result.get("errMsg") or "invalid"
|
||
raise ExpiredError(f"许可证在线状态无效: {reason}")
|
||
if result.get("licenseId") not in (None, payload["license_id"]):
|
||
raise ExpiredError("许可证在线校验返回了不匹配的许可证")
|
||
_last_online_success_monotonic = _time_module.monotonic()
|
||
|
||
|
||
# ============================================================
|
||
# 时间可信度
|
||
# ============================================================
|
||
def _get_uptime_days():
|
||
"""软件已连续运行的天数（基于 monotonic，不受系统时间影响）"""
|
||
return (_time_module.monotonic() - _startup_monotonic) / 86400
|
||
|
||
|
||
def _get_trusted_now():
|
||
"""多源交叉校验获取可信日期时间（精确到小时）。
|
||
|
||
取系统时间和文件时间的最大值，防止用户回拨系统时间。
|
||
"""
|
||
candidates = []
|
||
|
||
# 1. 系统时间
|
||
candidates.append(datetime.datetime.now())
|
||
|
||
# 2. 软件启动时记录的"最早可能时间"
|
||
# monotonic 计时推导出的启动时间
|
||
startup_guess = datetime.datetime.now() - datetime.timedelta(
|
||
days=_get_uptime_days()
|
||
)
|
||
candidates.append(startup_guess)
|
||
|
||
# 3. 系统文件的修改时间（不易被用户修改）
|
||
system_files = []
|
||
if sys.platform == "darwin":
|
||
system_files = [
|
||
"/System/Library/CoreServices/SystemVersion.plist",
|
||
"/usr/bin/python3",
|
||
]
|
||
elif sys.platform == "win32":
|
||
system_files = [
|
||
r"C:\Windows\System32\ntoskrnl.exe",
|
||
r"C:\Windows\explorer.exe",
|
||
]
|
||
|
||
for sf in system_files:
|
||
if os.path.exists(sf):
|
||
try:
|
||
mtime = os.path.getmtime(sf)
|
||
candidates.append(datetime.datetime.fromtimestamp(mtime))
|
||
except OSError:
|
||
pass
|
||
|
||
# 取最大值（真时间 ≥ 所有候选值。用户可能回拨，但不能让其他文件"变新"）
|
||
return max(candidates)
|
||
|
||
|
||
# ============================================================
|
||
# 后台巡检
|
||
# ============================================================
|
||
def start_license_watchdog(interval_minutes=DEFAULT_CHECK_INTERVAL):
|
||
"""启动后台许可证巡检线程。
|
||
|
||
Args:
|
||
interval_minutes: 检查间隔（分钟），默认 1440（24小时）
|
||
"""
|
||
global _watchdog_started
|
||
with _watchdog_lock:
|
||
if _watchdog_started:
|
||
return
|
||
_watchdog_started = True
|
||
t = threading.Thread(
|
||
target=_watchdog_loop,
|
||
args=(max(5, interval_minutes),),
|
||
daemon=True,
|
||
name="license-watchdog",
|
||
)
|
||
t.start()
|
||
|
||
|
||
def _watchdog_loop(interval_minutes):
|
||
while True:
|
||
_time_module.sleep(interval_minutes * 60)
|
||
|
||
try:
|
||
payload = verify_license()
|
||
with _verified_license_lock:
|
||
global _verified_license
|
||
_verified_license = dict(payload)
|
||
validate_license_online(payload)
|
||
expiry_str = payload["expiry"]
|
||
# 解析到期时间（兼容旧格式）
|
||
if len(expiry_str) == 10:
|
||
expiry = datetime.datetime.strptime(expiry_str + " 23:59", "%Y-%m-%d %H:%M")
|
||
else:
|
||
expiry = datetime.datetime.strptime(expiry_str, "%Y-%m-%d %H:%M")
|
||
|
||
trusted_now = _get_trusted_now()
|
||
hours_left = (expiry - trusted_now).total_seconds() / 3600
|
||
days_left = int(hours_left / 24)
|
||
|
||
if hours_left < 0:
|
||
# 已过期（不应走到这里，verify_license 会抛 ExpiredError）
|
||
_log(f"⚠ 许可证已过期 ({expiry_str})")
|
||
_show_message_box('critical', "许可证已过期",
|
||
f"您的许可证已于 {expiry_str} 到期！\n请尽快联系厂商续期。")
|
||
|
||
elif hours_left <= 24:
|
||
if not _warning_shown_states["expiring_today"]:
|
||
_warning_shown_states["expiring_today"] = True
|
||
_log(f"⚠ 许可证将在今天到期 ({expiry_str})")
|
||
_show_message_box('warning', "许可证即将到期",
|
||
f"您的许可证将于今天 {expiry_str} 到期！\n"
|
||
f"剩余约 {int(hours_left)} 小时，请及时联系厂商续期。")
|
||
|
||
elif days_left <= 30:
|
||
if not _warning_shown_states["expiring_soon"]:
|
||
_warning_shown_states["expiring_soon"] = True
|
||
_log(f"⚠ 许可证将在 {days_left} 天后到期 ({expiry_str})")
|
||
_show_message_box('warning', "许可证即将到期",
|
||
f"您的许可证将在 {days_left} 天后（{expiry_str}）到期\n请提前联系厂商续期。")
|
||
|
||
except ExpiredError as e:
|
||
_handle_expired(e)
|
||
|
||
except Exception as e:
|
||
_log(f"许可证巡检异常: {e}")
|
||
|
||
|
||
def _handle_expired(error):
|
||
"""处理许可证过期"""
|
||
_log(f"❌ {error}")
|
||
|
||
# 弹窗：许可证已过期
|
||
_show_message_box('critical', "许可证已过期",
|
||
f"{error}\n\n软件将在 {GRACE_PERIOD_HOURS} 小时缓冲期后自动退出，\n请及时保存工作并联系厂商续期。")
|
||
|
||
if _on_expired_callback:
|
||
_on_expired_callback(str(error))
|
||
|
||
# 宽限期：给用户时间保存工作
|
||
grace_start = _time_module.monotonic()
|
||
grace_seconds = GRACE_PERIOD_HOURS * 3600
|
||
|
||
if not _warning_shown_states["expired_grace"]:
|
||
_warning_shown_states["expired_grace"] = True
|
||
if _on_grace_callback:
|
||
_on_grace_callback(GRACE_PERIOD_HOURS)
|
||
|
||
# 半小时后提醒一次
|
||
warned_half = False
|
||
while _time_module.monotonic() - grace_start < grace_seconds:
|
||
elapsed = _time_module.monotonic() - grace_start
|
||
if not warned_half and elapsed > grace_seconds / 2:
|
||
warned_half = True
|
||
_show_message_box('warning', "许可证已过期",
|
||
f"缓冲期剩余约 {GRACE_PERIOD_HOURS // 2} 小时，\n请尽快保存工作！")
|
||
_time_module.sleep(60) # 每分钟检查一次
|
||
|
||
# 宽限期过，强制退出
|
||
_log("宽限期已过，软件即将退出")
|
||
_show_message_box('critical', "许可证已过期",
|
||
"缓冲期已结束，软件即将退出。\n请联系厂商续期后重新启动。")
|
||
|
||
# 给 5 秒做最后的清理
|
||
_time_module.sleep(5)
|
||
os._exit(1)
|
||
|
||
|
||
def set_on_expired(callback):
|
||
"""设置过期回调: callback(message: str)"""
|
||
global _on_expired_callback
|
||
_on_expired_callback = callback
|
||
|
||
|
||
def set_on_grace(callback):
|
||
"""设置宽限期回调: callback(hours: int)"""
|
||
global _on_grace_callback
|
||
_on_grace_callback = callback
|
||
|
||
|
||
def set_on_log(callback):
|
||
"""设置日志回调: callback(message: str)
|
||
所有许可证关键日志会同时输出到此回调，供 UI 状态栏显示。
|
||
"""
|
||
global _on_log_callback
|
||
_on_log_callback = callback
|
||
|
||
|
||
# ============================================================
|
||
# 便捷入口
|
||
# ============================================================
|
||
class ExpiredError(RuntimeError):
|
||
"""许可证过期异常"""
|
||
def __init__(self, message, expiry=None):
|
||
super().__init__(message)
|
||
self.expiry = expiry
|
||
|
||
|
||
def _log(msg):
|
||
"""终端日志 + UI 回调（不依赖任何 UI 层）"""
|
||
timestamp = datetime.datetime.now().strftime("%Y-%m-%d %H:%M:%S")
|
||
formatted = f"[License {timestamp}] {msg}"
|
||
print(formatted)
|
||
# 同步推送到 UI 状态栏（如果已注册回调）
|
||
if _on_log_callback:
|
||
try:
|
||
_on_log_callback(str(msg))
|
||
except Exception:
|
||
pass
|
||
|
||
|
||
def check_license(lic_path=None):
|
||
"""启动时调用：验证许可证，通过则返回 payload。
|
||
|
||
在 main.py 的 main() 函数开头调用一次即可。
|
||
内部会自动启动后台巡检线程。
|
||
|
||
Returns:
|
||
dict: 许可证载荷
|
||
|
||
Raises:
|
||
SystemExit: 验签失败或已过期（启动阶段直接退出）
|
||
"""
|
||
try:
|
||
payload = verify_license(lic_path)
|
||
environment_device_id = os.environ.get("REINLOOP_DEVICE_ID", "").strip()
|
||
if (_is_new_license(payload) and environment_device_id
|
||
and environment_device_id != payload["device_id"]):
|
||
raise ValueError("REINLOOP_DEVICE_ID 与许可证 device_id 不一致")
|
||
validate_license_online(payload)
|
||
with _verified_license_lock:
|
||
global _verified_license
|
||
_verified_license = dict(payload)
|
||
start_license_watchdog()
|
||
expiry = payload.get("expiry", "未知")
|
||
customer = payload.get("customer", "未知")
|
||
_log(f"✅ 许可证有效 | 客户: {customer} | 到期: {expiry}")
|
||
return payload
|
||
|
||
except FileNotFoundError as e:
|
||
_log(f"❌ {e}")
|
||
_show_error_and_exit(
|
||
"未找到许可证文件",
|
||
"请将 license.lic 放到软件根目录，然后重新启动程序。\n\n"
|
||
"如有疑问，请联系厂商获取有效的许可证文件。"
|
||
)
|
||
|
||
except InvalidSignature as e:
|
||
_log(f"❌ {e}")
|
||
_show_error_and_exit(
|
||
"许可证验证失败",
|
||
"许可证签名校验不通过，文件可能已被篡改。\n\n"
|
||
"请使用原始签发的 license.lic 文件，\n"
|
||
"或联系厂商重新签发。"
|
||
)
|
||
|
||
except ExpiredError as e:
|
||
_log(f"❌ {e}")
|
||
_show_error_and_exit(
|
||
"许可证已过期",
|
||
f"您的许可证已于 {e.expiry} 到期。\n\n"
|
||
"请联系厂商续期，获取新的许可证文件后重新启动。"
|
||
)
|
||
|
||
except Exception as e:
|
||
_log(f"❌ 许可证检查异常: {e}")
|
||
_show_error_and_exit(f"许可证校验失败: {e}", str(e))
|
||
|
||
|
||
def _show_error_and_exit(title, detail=""):
|
||
"""显示错误弹窗并退出（兼容 GUI 和无 GUI 模式）。
|
||
|
||
Args:
|
||
title: 弹窗标题（简短概要）
|
||
detail: 弹窗正文（详细说明和操作建议）
|
||
"""
|
||
message = f"{title}\n\n{detail}" if detail else title
|
||
|
||
# 尝试 GUI 弹窗
|
||
try:
|
||
from PySide6.QtWidgets import QApplication, QMessageBox
|
||
app = QApplication.instance()
|
||
if app is not None:
|
||
QMessageBox.critical(None, title, detail or title)
|
||
else:
|
||
# 无 QApplication 实例时，尝试创建一个临时的
|
||
try:
|
||
app = QApplication(sys.argv[:1])
|
||
QMessageBox.critical(None, title, detail or title)
|
||
except Exception:
|
||
pass
|
||
except Exception:
|
||
pass
|
||
|
||
print(f"\n{'='*50}")
|
||
print(message)
|
||
print(f"{'='*50}\n")
|
||
sys.exit(1)
|
||
|
||
|
||
# ============================================================
|
||
# 辅助：获取许可证信息（供 UI 显示）
|
||
# ============================================================
|
||
def get_license_info(lic_path=None):
|
||
"""读取许可证信息（不做过期检查），供 UI 显示。
|
||
|
||
Returns:
|
||
dict | None: 许可证信息，文件不存在则返回 None
|
||
"""
|
||
try:
|
||
if lic_path is None:
|
||
exe_dir = (
|
||
Path(sys.executable).parent
|
||
if getattr(sys, 'frozen', False)
|
||
else Path.cwd()
|
||
)
|
||
lic_path = exe_dir / LICENSE_FILE
|
||
|
||
if not os.path.exists(lic_path):
|
||
return None
|
||
|
||
with open(lic_path, "r", encoding="utf-8") as f:
|
||
raw = f.read().strip()
|
||
|
||
payload_b64 = raw.split("|")[0]
|
||
payload = json.loads(base64.b64decode(payload_b64))
|
||
|
||
# 先验签保证内容可信
|
||
verify_license(lic_path)
|
||
|
||
return {
|
||
"customer": payload.get("customer", "未知"),
|
||
"expiry": payload.get("expiry", "未知"),
|
||
"issued": payload.get("issued", "未知"),
|
||
"features": payload.get("features", "*"),
|
||
}
|
||
except Exception:
|
||
return None
|