louis/ReinLoopTest
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Merge https://git.dominatedconvergence.com/Epifnne/ReinLoop into reinlooptest

Browse Source
This commit is contained in:
rangang
2026-08-03 15:50:43 +08:00
parent 1093980b77 5edb0b2d69
commit 270895aaf0
12 changed files with 514 additions and 26 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ReinLoop/core/__init__.py
+1 -1
View File
@@ -30,7 +30,7 @@ def _init_license():
# 生产环境（exe 打包）→ 严格执行验签
from license_utils import check_license
check_license() # 验签并启动唯一的后台巡检线程，失败直接退出
check_license() # 仅启动时验签，失败直接退出
_LICENSE_CHECKED = True
ReinLoop/core/model_manager.py
+1
View File
@@ -6,6 +6,7 @@
import threading
import io
import requests
import torch
from stable_baselines3 import SAC
ReinLoop/license_utils.py
+35 -18
View File
@@ -41,8 +41,8 @@ Ue6JWRU4j3Wg37WDPbwkO3tQba2jbUQsLYomLGuohfkVAgMBAAE=
-----END PUBLIC KEY-----"""
# {{LICENSE_PUBLIC_KEY_END}}
# 许可证文件相对路径
LICENSE_FILE = "license.lic"
# 许可证文件检索模式（默认在程序目录中匹配）
LICENSE_GLOB = "*license.lic"
# 巡检间隔（分钟）
DEFAULT_CHECK_INTERVAL = 5
@@ -164,11 +164,38 @@ def _validate_payload(payload):
return has_new_format
def _default_license_dir():
"""返回默认许可证搜索目录。"""
# PyInstaller 打包后 sys.executable 是 exe 路径
return Path(sys.executable).parent if getattr(sys, 'frozen', False) else Path.cwd()
def _resolve_license_path(lic_path=None):
"""解析许可证路径。
- 显式传入 ``lic_path`` 时直接使用该路径。
- 未传入时，在默认目录按 ``*license.lic`` 匹配，优先选择最近修改的文件。
"""
if lic_path is not None:
return Path(lic_path)
search_dir = _default_license_dir()
matches = [path for path in search_dir.glob(LICENSE_GLOB) if path.is_file()]
if not matches:
raise FileNotFoundError(
f"未找到许可证文件（模式: {LICENSE_GLOB}，目录: {search_dir})"
)
# 多个候选时优先取最新文件；同修改时间再按文件名稳定排序。
matches.sort(key=lambda path: (path.stat().st_mtime, path.name), reverse=True)
return matches[0]
def verify_license(lic_path=None):
"""验证许可证签名 + 有效期。
Args:
lic_path: 许可证文件路径，默认 exe 同级目录下的 license.lic
lic_path: 许可证文件路径，默认在程序目录按 *license.lic 自动匹配
Returns:
dict: 许可证 payload（customer, expiry, issued 等）
@@ -179,10 +206,7 @@ def verify_license(lic_path=None):
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
lic_path = _resolve_license_path(lic_path)
if not os.path.exists(lic_path):
raise FileNotFoundError(f"许可证文件不存在: {lic_path}")
@@ -531,7 +555,7 @@ def check_license(lic_path=None):
"""启动时调用：验证许可证，通过则返回 payload。
在 main.py 的 main() 函数开头调用一次即可。
内部会自动启动后台巡检线程。
仅启动时执行校验，不自动启动后台巡检线程。
Returns:
dict: 许可证载荷
@@ -549,7 +573,6 @@ def check_license(lic_path=None):
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}")
@@ -559,7 +582,7 @@ def check_license(lic_path=None):
_log(f"❌ {e}")
_show_error_and_exit(
"未找到许可证文件",
"请将 license.lic 放到软件根目录，然后重新启动程序。\n\n"
"请将许可证文件放到软件根目录（文件名需匹配 *license.lic），然后重新启动程序。\n\n"
"如有疑问，请联系厂商获取有效的许可证文件。"
)
@@ -568,7 +591,7 @@ def check_license(lic_path=None):
_show_error_and_exit(
"许可证验证失败",
"许可证签名校验不通过，文件可能已被篡改。\n\n"
"请使用原始签发的 license.lic 文件，\n"
"请使用原始签发且文件名匹配 *license.lic 的许可证文件，\n"
"或联系厂商重新签发。"
)
@@ -626,13 +649,7 @@ def get_license_info(lic_path=None):
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
lic_path = _resolve_license_path(lic_path)
if not os.path.exists(lic_path):
return None
ReinLoop/tests/test_license_protocol.py
+39
View File
@@ -7,6 +7,7 @@ import os
from pathlib import Path
import sys
import tempfile
import time
import types
import unittest
from unittest.mock import patch
@@ -107,6 +108,44 @@ class LicenseProtocolTests(unittest.TestCase):
side_effect=LICENSE.requests.ConnectionError("offline")):
LICENSE.validate_license_online(NEW_LICENSE)
def test_verify_license_uses_default_glob_pattern_when_path_missing(self):
with tempfile.TemporaryDirectory() as tmp_dir:
payload_b64 = base64.b64encode(json.dumps(NEW_LICENSE).encode()).decode()
license_path = Path(tmp_dir) / "customer-license.lic"
license_path.write_text(
f"{payload_b64}|{base64.b64encode(b'signature').decode()}",
encoding="utf-8",
)
with patch.object(LICENSE, "_load_public_key", return_value=FakePublicKey()), \
patch.object(LICENSE, "_default_license_dir", return_value=Path(tmp_dir)):
payload = LICENSE.verify_license()
self.assertEqual(payload["license_id"], NEW_LICENSE["license_id"])
def test_verify_license_prefers_latest_matching_file(self):
with tempfile.TemporaryDirectory() as tmp_dir:
old_payload = dict(NEW_LICENSE, license_id="old-license")
new_payload = dict(NEW_LICENSE, license_id="new-license")
def write_license(path, payload):
payload_b64 = base64.b64encode(json.dumps(payload).encode()).decode()
path.write_text(
f"{payload_b64}|{base64.b64encode(b'signature').decode()}",
encoding="utf-8",
)
old_path = Path(tmp_dir) / "a-license.lic"
new_path = Path(tmp_dir) / "z-license.lic"
write_license(old_path, old_payload)
time.sleep(0.01)
write_license(new_path, new_payload)
with patch.object(LICENSE, "_load_public_key", return_value=FakePublicKey()), \
patch.object(LICENSE, "_default_license_dir", return_value=Path(tmp_dir)):
payload = LICENSE.verify_license()
self.assertEqual(payload["license_id"], "new-license")
def test_api_rejects_environment_device_id_mismatch(self):
fake_license_utils = types.ModuleType("license_utils")
fake_license_utils.get_verified_license = lambda: dict(NEW_LICENSE)