378 lines
15 KiB
Python
378 lines
15 KiB
Python
"""机器码授权门禁测试（网关侧）。
|
||
|
||
核心验证目标 —— 未授权插件被拒绝时，**网关必须仍能正常启动**：
|
||
|
||
1. 门禁收口在 `_discover_plugins()`：未授权插件不出现在返回列表中，
|
||
且 `denied_plugin_types()` 能给出拒绝原因；
|
||
2. **契约校验不硬失败**：这是最容易踩的坑 —— `validate_contracts()` 会把
|
||
"TOML 路由没有对应 @rpc_method 注册"视为硬性错误并终止启动。若把"不加载
|
||
未授权插件"实现成"跳过 import 但保留在插件列表里"，该插件的 TOML 路由就会
|
||
变成孤儿 → 设备起不来。本测试确保收口位置正确。
|
||
3. 路由描述符门禁：未授权类型的 external 路由描述符被忽略（防"改 load_mode
|
||
成 external 绕过门禁"）；
|
||
4. 管理接口门禁：未授权插件的扩缩容/重启/配置接口一律 403；
|
||
5. 未启用校验时（开发/测试）全部放行。
|
||
|
||
运行方式：`python test/test_license_gate.py`，或 `uv run pytest test/test_license_gate.py`。
|
||
"""
|
||
import asyncio
|
||
import base64
|
||
import json
|
||
import os
|
||
import sys
|
||
import tempfile
|
||
from datetime import date, timedelta
|
||
from pathlib import Path
|
||
|
||
ROOT = Path(__file__).resolve().parent.parent
|
||
sys.path.insert(0, str(ROOT))
|
||
|
||
# 必须在导入网关模块前设置：licensing 的 enforce 在首次建管理器时读取
|
||
os.environ["CHONGMING_LICENSE_MODE"] = "enforce"
|
||
|
||
from chongming_config import get_config # noqa: E402
|
||
from cryptography.hazmat.primitives import serialization # noqa: E402
|
||
from cryptography.hazmat.primitives.asymmetric.ed25519 import Ed25519PrivateKey # noqa: E402
|
||
from chongming_license import collect_fingerprints, machine_code, sign_payload # noqa: E402
|
||
|
||
from chongming.app.core import licensing # noqa: E402
|
||
from chongming.app.core.lifespan_setup import ( # noqa: E402
|
||
_discover_plugins,
|
||
_scan_plugins,
|
||
denied_plugin_types,
|
||
)
|
||
|
||
failures: list[str] = []
|
||
|
||
|
||
def check(name, cond, detail=""):
|
||
status = "PASS" if cond else "FAIL"
|
||
print(f"[{status}] {name}" + (f" — {detail}" if detail else ""))
|
||
if not cond:
|
||
failures.append(name)
|
||
|
||
|
||
# ---------------------------------------------------------------------------
|
||
# 测试夹具：临时密钥对 + license 文件
|
||
# ---------------------------------------------------------------------------
|
||
def _make_keypair() -> tuple[str, Ed25519PrivateKey]:
|
||
priv = Ed25519PrivateKey.generate()
|
||
pub = base64.b64encode(
|
||
priv.public_key().public_bytes(
|
||
encoding=serialization.Encoding.Raw,
|
||
format=serialization.PublicFormat.Raw,
|
||
)
|
||
).decode("ascii")
|
||
return pub, priv
|
||
|
||
|
||
def _issue(tmpdir: Path, priv: Ed25519PrivateKey, plugins: list[str],
|
||
expires_days: int = 365, sources: dict | None = None) -> str:
|
||
src = sources if sources is not None else collect_fingerprints()
|
||
payload = {
|
||
"v": 1,
|
||
"lic_id": "L-TEST-0001",
|
||
"customer": "测试客户",
|
||
"machine_code": machine_code(src),
|
||
"sources": src,
|
||
"plugins": plugins,
|
||
"not_before": None,
|
||
"expires_at": (date.today() + timedelta(days=expires_days)).isoformat(),
|
||
"issued_at": "2026-09-20T00:00:00Z",
|
||
"note": "",
|
||
}
|
||
doc = {
|
||
"payload": payload,
|
||
"alg": "ed25519",
|
||
"kid": "test",
|
||
"sig": sign_payload(payload, priv),
|
||
}
|
||
path = tmpdir / "license.lic"
|
||
path.write_text(json.dumps(doc), encoding="utf-8")
|
||
return str(path)
|
||
|
||
|
||
def _use_license(public_key: str, path: str, min_match: int = 3) -> None:
|
||
"""把测试用的公钥/路径写进运行期配置并重建管理器。
|
||
|
||
直接改 `get_config()` 返回的 dict（chongming_config 缓存同一对象）。
|
||
"""
|
||
cfg = get_config()
|
||
cfg.setdefault("license", {})
|
||
cfg["license"]["public_key"] = public_key
|
||
cfg["license"]["path"] = path
|
||
cfg["license"]["min_match"] = min_match
|
||
licensing.reset_manager()
|
||
|
||
|
||
def test_gate_denies_unlicensed_plugins():
|
||
"""未授权插件被剔除，且原因可查。"""
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, priv = _make_keypair()
|
||
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column"]))
|
||
|
||
scanned = sorted(p["type"] for p in _scan_plugins())
|
||
allowed = sorted(p["type"] for p in _discover_plugins())
|
||
denied = denied_plugin_types()
|
||
|
||
check("盘扫到全部插件", scanned == ["aibox_sophon_adapter", "ip_sound_column"],
|
||
f"scanned={scanned}")
|
||
check("只放行已授权插件", allowed == ["ip_sound_column"], f"allowed={allowed}")
|
||
check("未授权插件给出拒绝原因",
|
||
"aibox_sophon_adapter" in denied
|
||
and "白名单" in denied["aibox_sophon_adapter"],
|
||
str(denied))
|
||
|
||
|
||
def test_contract_validation_survives_denial():
|
||
"""关键：有插件被拒时契约校验不得抛异常（否则设备起不来）。"""
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, priv = _make_keypair()
|
||
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column"]))
|
||
|
||
from chongming.app.core.contracts import validate_contracts
|
||
|
||
inline = [p for p in _discover_plugins() if p.get("load_mode", "inline") != "external"]
|
||
try:
|
||
validate_contracts(inline)
|
||
check("被拒插件的 TOML 路由不会导致契约校验硬失败", True)
|
||
except Exception as exc: # noqa: BLE001
|
||
check("被拒插件的 TOML 路由不会导致契约校验硬失败", False, repr(exc))
|
||
|
||
|
||
def test_all_denied_still_validates():
|
||
"""极端情况：完全未授权（无 license 文件）时启动流程仍不硬失败。"""
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, _priv = _make_keypair()
|
||
_use_license(pub, str(tmp / "not-exists.lic"))
|
||
|
||
from chongming.app.core.contracts import validate_contracts
|
||
|
||
allowed = _discover_plugins()
|
||
check("无授权文件时不放行任何插件", allowed == [], str(allowed))
|
||
try:
|
||
validate_contracts([])
|
||
check("完全未授权时契约校验不抛异常", True)
|
||
except Exception as exc: # noqa: BLE001
|
||
check("完全未授权时契约校验不抛异常", False, repr(exc))
|
||
|
||
|
||
def test_route_registry_denies_descriptors():
|
||
"""未授权类型的路由描述符被忽略（防改 load_mode 绕过）。"""
|
||
from fastapi import FastAPI
|
||
|
||
from chongming.app.core.route_registry import RouteRegistry
|
||
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, priv = _make_keypair()
|
||
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column"]))
|
||
|
||
denied = licensing.denied_types({p["type"] for p in _scan_plugins()})
|
||
check("denied_types 含未授权插件", "aibox_sophon_adapter" in denied, str(denied))
|
||
|
||
async def run():
|
||
registry = RouteRegistry(FastAPI(), worker_manager=None, prefix="/api/v1")
|
||
registry.set_denied_types(denied)
|
||
descriptor = {
|
||
"schema_version": 1, "type": "aibox_sophon_adapter",
|
||
"subject": "aibox_sophon_adapter.alarm_report", "method": "POST",
|
||
"path": "/aibox_sophon_adapter/alarm", "params": {},
|
||
}
|
||
await registry._handle_sample(
|
||
"chongming/routes/aibox_sophon_adapter/alarm_report",
|
||
__import__("zenoh").SampleKind.PUT,
|
||
json.dumps(descriptor).encode(),
|
||
)
|
||
return registry._groups, registry.route_types
|
||
|
||
groups, route_types = asyncio.run(run())
|
||
check("未授权类型的描述符未进入路由表", groups == {}, f"groups={list(groups)}")
|
||
check("未授权类型不出现在 route_types 中", route_types == [], str(route_types))
|
||
|
||
|
||
def test_admin_gate_403_for_denied():
|
||
"""未授权插件的管理操作一律 403。"""
|
||
from fastapi import HTTPException
|
||
|
||
from chongming.app.routes.worker_admin import _ensure_licensed
|
||
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, priv = _make_keypair()
|
||
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column"]))
|
||
|
||
try:
|
||
_ensure_licensed("aibox_sophon_adapter")
|
||
check("未授权插件的管理操作被拒", False, "未抛 403")
|
||
except HTTPException as exc:
|
||
check("未授权插件的管理操作被拒", exc.status_code == 403,
|
||
f"status={exc.status_code}")
|
||
|
||
try:
|
||
_ensure_licensed("ip_sound_column")
|
||
check("已授权插件的管理操作放行", True)
|
||
except HTTPException as exc: # noqa: BLE001
|
||
check("已授权插件的管理操作放行", False, f"status={exc.status_code}")
|
||
|
||
|
||
def test_expired_license_denies_all():
|
||
"""过期授权 -> 全部拒绝。"""
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, priv = _make_keypair()
|
||
_use_license(pub, _issue(tmp, priv, plugins=["ip_sound_column", "aibox_sophon_adapter"],
|
||
expires_days=-1))
|
||
|
||
allowed = _discover_plugins()
|
||
check("过期授权不放行任何插件", allowed == [], str(allowed))
|
||
check("拒绝原因为过期",
|
||
all("过期" in m for m in denied_plugin_types().values()),
|
||
str(denied_plugin_types()))
|
||
|
||
|
||
def test_disabled_mode_allows_everything():
|
||
"""未启用校验时（非打包态的默认行为）全部放行。"""
|
||
tmp = Path(tempfile.mkdtemp())
|
||
pub, priv = _make_keypair()
|
||
_use_license(pub, _issue(tmp, priv, plugins=[]))
|
||
licensing.reset_manager()
|
||
os.environ.pop("CHONGMING_LICENSE_MODE", None)
|
||
get_config()["license"]["enabled"] = False
|
||
try:
|
||
allowed = sorted(p["type"] for p in _discover_plugins())
|
||
check("未启用校验时全部放行",
|
||
allowed == ["aibox_sophon_adapter", "ip_sound_column"], str(allowed))
|
||
check("未启用校验时无拒绝记录", denied_plugin_types() == {},
|
||
str(denied_plugin_types()))
|
||
finally:
|
||
os.environ["CHONGMING_LICENSE_MODE"] = "enforce"
|
||
get_config()["license"]["enabled"] = False
|
||
licensing.reset_manager()
|
||
|
||
|
||
def test_http_endpoints_and_upload_safety():
|
||
"""HTTP 层：门禁、信息接口、以及"坏文件不覆盖现有授权"。
|
||
|
||
这一步覆盖最容易在现场出事的行为 —— 误传一个坏 license 文件把设备搞成未授权。
|
||
"""
|
||
import base64 as _b64
|
||
import json as _json
|
||
import tempfile as _tf
|
||
from datetime import date as _date, timedelta as _td
|
||
|
||
import orjson
|
||
from cryptography.hazmat.primitives import serialization as _ser
|
||
from fastapi.testclient import TestClient
|
||
|
||
from chongming_jwt import JWT
|
||
from chongming_license import collect_fingerprints as _collect
|
||
|
||
from chongming.app import app
|
||
|
||
tmp = Path(_tf.mkdtemp())
|
||
priv = Ed25519PrivateKey.generate()
|
||
pub = _b64.b64encode(
|
||
priv.public_key().public_bytes(
|
||
encoding=_ser.Encoding.Raw, format=_ser.PublicFormat.Raw
|
||
)
|
||
).decode("ascii")
|
||
|
||
src = _collect()
|
||
payload = {
|
||
"v": 1, "lic_id": "L-HTTP-1", "customer": "HTTP 测试",
|
||
"machine_code": machine_code(src), "sources": src,
|
||
"plugins": ["ip_sound_column"], "not_before": None,
|
||
"expires_at": (_date.today() + _td(days=30)).isoformat(),
|
||
"issued_at": "2026-09-20T00:00:00Z", "note": "",
|
||
}
|
||
lic = tmp / "license.lic"
|
||
lic.write_text(_json.dumps({
|
||
"payload": payload, "alg": "ed25519", "kid": "test",
|
||
"sig": sign_payload(payload, priv),
|
||
}), encoding="utf-8")
|
||
|
||
_use_license(pub, str(lic))
|
||
|
||
sec = get_config()["security"]
|
||
def _token(superuser: bool) -> str:
|
||
return JWT(sec["secret_key"], sec["algorithm"]).encode(
|
||
orjson.dumps({"sub": "u1", "username": "admin",
|
||
"is_superuser": superuser}).decode(),
|
||
expires_in_seconds=3600,
|
||
)
|
||
|
||
client = TestClient(app)
|
||
auth = {"Authorization": f"Bearer {_token(True)}"}
|
||
|
||
r = client.get("/api/v1/license/info")
|
||
check("无 token 时拒绝访问", r.status_code == 401, f"status={r.status_code}")
|
||
|
||
r = client.get("/api/v1/license/info",
|
||
headers={"Authorization": f"Bearer {_token(False)}"})
|
||
check("非超管拒绝访问", r.status_code == 403, f"status={r.status_code}")
|
||
|
||
r = client.get("/api/v1/license/info", headers=auth)
|
||
check("超管可读取授权信息", r.status_code == 200, f"status={r.status_code}")
|
||
info = r.json()
|
||
check("/license/info 含机器码", info["status"]["machine_code"] != "",
|
||
info["status"]["machine_code"])
|
||
check("/license/info 含申请码", info["request_code"].startswith("CM1-"),
|
||
info["request_code"][:16])
|
||
check("未授权插件在 denied 中", "aibox_sophon_adapter" in info["plugins"]["denied"],
|
||
str(list(info["plugins"]["denied"])))
|
||
check("来源明细含 weak 标注",
|
||
any(s["weak"] for s in info["sources_detail"]),
|
||
str([(s["name"], s["weak"]) for s in info["sources_detail"]]))
|
||
|
||
# 签名错误的文件：必须拒绝，且不得覆盖现有授权
|
||
tampered = dict(payload, plugins=["aibox_sophon_adapter"])
|
||
r = client.post(
|
||
"/api/v1/license/file", headers=auth,
|
||
files={"file": ("bad.lic", _json.dumps(
|
||
{"payload": tampered, "alg": "ed25519", "kid": "test", "sig": "AAAA"}
|
||
), "application/json")},
|
||
)
|
||
check("签名错误的文件被拒绝", r.status_code == 422, f"status={r.status_code}")
|
||
check("被拒文件未覆盖现有授权",
|
||
_json.loads(lic.read_text(encoding="utf-8"))["payload"]["lic_id"] == "L-HTTP-1")
|
||
|
||
# 非 JSON 文件
|
||
r = client.post("/api/v1/license/file", headers=auth,
|
||
files={"file": ("bad.lic", "not json at all", "application/json")})
|
||
check("非 JSON 文件被拒绝", r.status_code == 422, f"status={r.status_code}")
|
||
|
||
# 合法文件：安装成功，并提示需重启加载
|
||
payload2 = dict(payload, lic_id="L-HTTP-2",
|
||
plugins=["ip_sound_column", "aibox_sophon_adapter"])
|
||
r = client.post(
|
||
"/api/v1/license/file", headers=auth,
|
||
files={"file": ("good.lic", _json.dumps(
|
||
{"payload": payload2, "alg": "ed25519", "kid": "test",
|
||
"sig": sign_payload(payload2, priv)}
|
||
), "application/json")},
|
||
)
|
||
check("合法文件导入成功", r.status_code == 200, f"status={r.status_code}")
|
||
if r.status_code == 200:
|
||
body = r.json()
|
||
check("导入后授权插件更新",
|
||
body["licensed_plugins"] == ["aibox_sophon_adapter", "ip_sound_column"],
|
||
str(body["licensed_plugins"]))
|
||
check("导入后提示需重启加载",
|
||
body["restart_required"] and "aibox_sophon_adapter" in body["pending_plugins"],
|
||
str(body["pending_plugins"]))
|
||
|
||
|
||
def main():
|
||
tests = [v for k, v in sorted(globals().items()) if k.startswith("test_")]
|
||
for fn in tests:
|
||
print(f"\n--- {fn.__name__} ---")
|
||
fn()
|
||
print("\n" + "=" * 56)
|
||
if failures:
|
||
print(f"❌ {len(failures)} 项未通过: {failures}")
|
||
return 1
|
||
print("✅ 全部通过")
|
||
return 0
|
||
|
||
|
||
if __name__ == "__main__":
|
||
sys.exit(main())
|