xmz7891/ykxmz
1
0
Fork 0
mirror of https://gitee.com/xmz7891/ykxmz.git synced 2026-10-18 21:50:44 +00:00
Code Issues Packages Projects Releases Wiki Activity

管理页新增：允许/禁止新设备接入 + 修改设备配对码

Browse Source 
== 功能一：允许新设备接入开关 ==

关掉之后只有已经登记过的电脑能继续上线，没见过的新设备一律拒绝。
状态存在 relay_state.json 里，重启不丢。

判断新设备用的是 deviceId，不是配对码 ——
这一点很关键：管理员给老设备改过码之后，
那台设备必须仍然能进来；按码判断的话它会被自己的开关挡在外面。
测试里专门有一条 A3b 守这个行为。

被拒时回 blocked=True，让客户端能区分被管理员挡了
和配对码冲突，而不是去匹配中文错误文案（文案一改就坏）。

== 功能二：修改任意设备的配对码 ==

管理页每台设备一个「改码」按钮，填 6 位数字即生效。
要点：
  * 新旧码冲突检查
  * 旧码映射必须删掉，否则变成孤儿条目，
    管理页配对码占用的数字会虚高
  * 改码踢掉该设备当前会话（会话绑在码上）
  * 立刻持久化

== 顺手修掉两个真 bug ==

1. 改码功能一上线就 500：Session 用了 __slots__，
   根本没有 session_id 属性（会话是以 token 为 key 存在 dict 里的），
   写 s.session_id 直接 AttributeError。
   改成遍历 sessions.items() 拿 key。

2. snapshot() 里有同样的错（清理超时设备那段）：
   平时 stale 列表是空的所以从没炸过，一旦真有设备超过 6 小时
   没上报就会 500。这次一起修了。

== 管理页 UI ==
  * 设备小节加了「允许新设备接入」勾选框
  * 每行加「改码」按钮（走 prompt，不引入弹窗层）
  * 改码按钮用 data-recode 而不是 data-device ——
    现有单台删除按 button.danger[data-device] 绑，
    带上 data-device 会被当成删除按钮，点改码变删设备。
    测试 C6 专门断言这一点。

== 测试 ==
新增 _tests/test_admin_device_control.py（23 项），
已注册进 _tests/verify_all.ps1 第 3e 阶段。
This commit is contained in:
xmz7891
2026-10-07 17:01:14 +08:00
parent 25e7780272
commit d4f335b8f9
21 changed files with 1520 additions and 7 deletions
Download Patch File Download Diff File Expand all files Collapse all files
_build/test_admin_features.py
+211
View File
@@ -0,0 +1,211 @@
#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r'''测试管理页新增的两个功能（在设备上跑，直连 127.0.0.1:8080）。
覆盖：
A. 「允许新设备接入」开关
A1 默认是开的
A2 关掉之后，**新**设备注册被拒（blocked=True）
A3 关掉之后，**已登记过的**设备仍然能上线
A4 重新打开之后，新设备又能注册
B. 「修改配对码」
B1 换成合法的 6 位数字 -> 成功
B2 新码立刻生效（能拿来注册/attach）
B3 旧码的映射被清掉（不会变成孤儿占位）
B4 换成已被别的设备占用的码 -> 被拒
B5 换成非 6 位数字 -> 被拒
B6 换成同一个码 -> 成功但 unchanged
B7 改码会把该设备的会话踢掉
C. 管理页 HTML 里有新控件
'''
from __future__ import annotations
import json
import sys
import time
import urllib.error
import urllib.request
BASE = "http://127.0.0.1:8080"
KEY_FILE = "/root/lanremote-relay/data/admin_key.txt"
PASS = 0
FAIL = 0
def check(ok: bool, label: str) -> None:
global PASS, FAIL
if ok:
PASS += 1
print(" [PASS] %s" % label)
else:
FAIL += 1
print(" [FAIL] %s" % label)
def call(path: str, payload: dict | None = None, method: str = "POST",
admin: bool = False, key: str = "") -> tuple:
data = json.dumps(payload, ensure_ascii=False).encode("utf-8") \
if payload is not None else None
headers = {"Content-Type": "application/json"}
if admin:
headers["X-Admin"] = key
req = urllib.request.Request(BASE + path, data=data, headers=headers,
method=method)
try:
with urllib.request.urlopen(req, timeout=15) as resp:
body = resp.read().decode("utf-8", "replace")
try:
return resp.status, json.loads(body)
except Exception: # noqa: BLE001
return resp.status, body
except urllib.error.HTTPError as exc:
body = exc.read().decode("utf-8", "replace")
try:
return exc.code, json.loads(body)
except Exception: # noqa: BLE001
return exc.code, body
except Exception as exc: # noqa: BLE001
return 0, {"error": str(exc)}
def main() -> int:
try:
key = open(KEY_FILE, encoding="utf-8").read().strip()
except OSError as exc:
print(" 读不到管理密钥: %s" % exc)
return 1
stamp = int(time.time())
dev_a = "featA-%d" % stamp
dev_b = "featB-%d" % stamp
dev_c = "featC-%d" % stamp
code_a = "%06d" % ((stamp + 11) % 1000000)
code_b = "%06d" % ((stamp + 22) % 1000000)
code_c = "%06d" % ((stamp + 33) % 1000000)
code_a2 = "%06d" % ((stamp + 44) % 1000000)
def reg(dev: str, code: str, name: str = "feat") -> tuple:
return call("/v1/agent/register", {
"deviceId": dev, "code": code, "name": name, "version": "1.8.2"})
print(" ================ A. 允许新设备接入开关 ================")
# A1 默认开
st, snap = call("/v1/health", method="GET")
check(snap.get("allowNew") is True, "A1 默认允许新设备接入")
# 先登记设备 A（此时开关是开的）
st, r = reg(dev_a, code_a, "设备A")
check(r.get("ok") is True, "A0 设备A 初始登记成功（码 %s）" % code_a)
# A2 关掉开关 + 已有会话
st, r = call("/v1/admin/allow_new", {"allow": False}, admin=True, key=key)
check(r.get("ok") is True and r.get("allowNew") is False,
"A2a 关闭开关成功")
st, r = reg(dev_b, code_b, "设备B")
check(r.get("ok") is False and r.get("blocked") is True,
"A2b 新设备B 被拒（blocked=True）: %s" % r.get("error", ""))
# A3 已登记过的 A 仍能上线
st, r = reg(dev_a, code_a, "设备A")
check(r.get("ok") is True,
"A3 已登记的设备A 仍能上线（关开关不影响老设备）")
# A4 重新打开
st, r = call("/v1/admin/allow_new", {"allow": True}, admin=True, key=key)
check(r.get("ok") is True and r.get("allowNew") is True, "A4a 重新打开成功")
st, r = reg(dev_c, code_c, "设备C")
check(r.get("ok") is True, "A4b 新设备C 又能注册了")
# ---------------- B ----------------
print()
print(" ================ B. 修改配对码 ================")
# B5 非 6 位 -> 拒
st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": "12ab"},
admin=True, key=key)
check(r.get("ok") is False, "B5 非 6 位数字被拒: %s" % r.get("error", ""))
# B4 已被占用 -> 拒
st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_c},
admin=True, key=key)
check(r.get("ok") is False, "B4 已被设备C占用的码被拒: %s" % r.get("error", ""))
# B1 正常改码
st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_a2},
admin=True, key=key)
check(r.get("ok") is True and r.get("code") == code_a2,
"B1 改码成功 %s -> %s" % (code_a, code_a2))
# B2 新码立刻生效
st, r = reg(dev_a, code_a2, "设备A")
check(r.get("ok") is True and r.get("code") == code_a2,
"B2 新码立刻可用于注册")
# B6 改成同一个 -> unchanged
st, r = call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_a2},
admin=True, key=key)
check(r.get("ok") is True and r.get("unchanged") is True,
"B6 改成同一个码 -> unchanged=True")
# B3 旧码不该还占着
st, snap = call("/v1/health", method="GET")
codes_used = {d["code"] for d in snap.get("devices", [])}
check(code_a2 in codes_used and code_a not in codes_used,
"B3 设备列表里只有新码（旧码 %s 已清）" % code_a)
# B7 会话被踢 —— 先 attach 一个
st, r = call("/v1/client/attach",
{"code": code_a2, "name": "测试手表"})
attached = r.get("ok") is True
if attached:
st2, r2 = call("/v1/admin/set_code",
{"deviceId": dev_a,
"code": "%06d" % ((stamp + 55) % 1000000)},
admin=True, key=key)
check(r2.get("ok") is True and r2.get("kicked", 0) >= 1,
"B7 改码踢掉了 %s 个会话" % r2.get("kicked"))
# 改回去，方便下面收尾
call("/v1/admin/set_code", {"deviceId": dev_a, "code": code_a2},
admin=True, key=key)
else:
check(True, "B7 跳过（attach 没成功：%s）" % r.get("error", ""))
# ---------------- C ----------------
print()
print(" ================ C. 管理页 HTML ================")
req = urllib.request.Request(BASE + "/zhongzhuanzhan",
headers={"X-Admin": key})
try:
with urllib.request.urlopen(req, timeout=15) as resp:
html = resp.read().decode("utf-8", "replace")
check('id="allowNew"' in html, "C1 页面有开关控件")
check("允许新设备接入" in html, "C2 页面有开关文案")
check("data-recode=" in html, "C3 设备行有改码按钮")
check("/v1/admin/allow_new" in html, "C4 JS 里调 allow_new 接口")
check("/v1/admin/set_code" in html, "C5 JS 里调 set_code 接口")
# 改码按钮不能带 data-device，否则会被删除逻辑吃掉
import re
recode_btns = re.findall(r'<button[^>]*data-recode="[^"]*"[^>]*>', html)
bad = [b for b in recode_btns if "data-device" in b]
check(not bad, "C6 改码按钮没有 data-device（不会误触发删除）")
except Exception as exc: # noqa: BLE001
check(False, "C 取管理页失败: %s" % exc)
# ---------------- 收尾 ----------------
print()
print(" 清理测试设备…")
for dev in (dev_a, dev_b, dev_c):
call("/v1/admin/delete", {"deviceId": dev}, admin=True, key=key)
call("/v1/admin/allow_new", {"allow": True}, admin=True, key=key)
print()
print("=" * 56)
print(" 通过 %d 项，失败 %d 项" % (PASS, FAIL))
return 0 if FAIL == 0 else 1
if __name__ == "__main__":
sys.exit(main())