feat(权限): 收敛普通账号写权限至本人凭证

将调度时刻、采集参数等实例级配置收归管理员,普通账号仅可维护本人 Cookie 与 User-Agent。
新增 config.writable_by 作为唯一写权限入口,set_setting 强制全局键落到 user_id=0,
消除「管理员改了只有自己生效」的静默缺陷。新增 tools/check_docs.py 文档自检,
smoke 断言扩至 215 项、check_live 扩至 122 项并支持普通账号越权验收,
忽略 backups/、data/*.bak* 与 legacy-v1/,版本升至 v1.3.0。
这个提交包含在:
2026-09-18 08:46:00 +08:00
父节点 df7db3582e
当前提交 1bf961f6b3
共修改 42 个文件,包含 2615 行新增和 854 行删除
+334
查看文件
@@ -0,0 +1,334 @@
#!/usr/bin/env python3
# SPDX-License-Identifier: MIT
"""文档自检:内部链接 / 跨文件锚点 / 图片引用 / 绝对路径泄漏 / 版本一致性 / 产品名硬编码。
文档一旦互相引用(README → docs/DEPLOYMENT.md#某节),章节重排就会让锚点**静默失效** ——
Markdown 不会报错,页面只是不跳转、图片只是显示裂图。这个脚本把这类问题变成可执行的断言。
用法:
python tools/check_docs.py # 有问题则退出码 1
python tools/check_docs.py --no-fail # 只看报告,不因问题而失败
退出码:0 通过,1 发现问题。
"""
from __future__ import annotations
import argparse
import os
import re
import sys
# ---------------------------------------------------------------- 常量
# 递归扫描时跳过的目录。**自动发现所有 .md**,而不是写死文件名列表 ——
# 写死列表的版本曾漏掉 THIRD-PARTY-NOTICES.md 与 CODE_OF_CONDUCT.md
# (新增文档时必然漏,而且没人会发现)。
SKIP_DIRS = {
".git", ".hg", ".svn", "node_modules", "vendor", "dist", "build",
".venv", "venv", "__pycache__", ".mypy_cache", ".pytest_cache", ".idea", ".vscode",
}
# markdown 链接:[文本](目标)
LINK_RE = re.compile(r"\[([^\]]*)\]\(([^)\s]+)(?:\s+\"[^\"]*\")?\)")
# markdown 图片:![alt](目标)
IMG_RE = re.compile(r"!\[[^\]]*\]\(([^)\s]+)\)")
# markdown 标题:# / ## / ... (用于生成锚点)
HEADING_RE = re.compile(r"^(#{1,6})\s+(.*?)\s*$")
# 显式 HTML 锚点:<a name="x"></a> 或 <a id="x"></a>
HTML_ANCHOR_RE = re.compile(r"<a\s+(?:name|id)=[\"']([^\"']+)[\"']")
# 代码块围栏(三反引号或三波浪线)
FENCE_RE = re.compile(r"^\s*(```|~~~)")
# 绝对路径泄漏:正向白名单写不出来,反着匹配已知模式足够有效。
# 每个模式的**第 1 个捕获组**是「用户名那一段」,用来判占位符。
LEAK_PATTERNS = [
(re.compile(r"[A-Za-z]:[\\/](?:Users|users)[\\/]([^\\/\s\"'`)]+)"),
"用户目录绝对路径"),
(re.compile(r"[A-Za-z]:[\\/]Documents and Settings[\\/]([^\\/\s\"'`)]+)"),
"用户目录绝对路径"),
(re.compile(r"/(?:home|Users)/([A-Za-z0-9._-]+)/"), "用户目录绝对路径"),
]
# 占位符豁免:`C:\Users\<用户名>\…` 是**良好实践**,不是泄漏。
PLACEHOLDER_RE = re.compile(
r"[<>%*{}]|^\.{2,}$|^[-_]+$"
r"|^(?:user|users|username|user-?name|your-?name|youruser|"
r"用户名|你的用户名|xxx+|yyy+|zzz+|aaa+|example|placeholder|"
r"me|someone|nobody)$",
re.IGNORECASE,
)
# 版本一致性:这四处必须互相一致
VERSION_INIT = os.path.join("workbuddy_portal", "__init__.py")
INIT_VER_RE = re.compile(r'^__version__\s*=\s*["\']([^"\']+)["\']', re.M)
DOCKERFILE = "Dockerfile"
OCI_VER_RE = re.compile(r'org\.opencontainers\.image\.version\s*=\s*"([^"]+)"')
README_VER_RE = re.compile(r"^\|\s*版本\s*\|\s*v?([0-9][^\s|]*)\s*\|", re.M)
CHANGELOG_VER_RE = re.compile(r"^##\s*\[?v?([0-9][^\s\]—-]*)", re.M)
# 产品名硬编码:应走 config.PROJECT_NAME 等上下文变量,不写进模板/JS
HARDCODE_NEEDLES = ["WorkBuddy Portal", "WorkBuddy 用量"]
HARDCODE_DIRS = [
os.path.join("workbuddy_portal", "web", "templates"),
os.path.join("workbuddy_portal", "web", "static", "js"),
]
# ---------------------------------------------------------------- 基础工具
def read(path: str) -> str:
with open(path, "r", encoding="utf-8", errors="replace") as fh:
return fh.read()
def strip_code_blocks(text: str) -> str:
"""去掉围栏代码块内容 —— 里面的 `#` 不是标题,里面的链接不该被当链接。
用空串占位(保留换行),这样**行号不会错位**,报错才能定位到真实位置。
"""
out, in_fence = [], False
for line in text.splitlines():
if FENCE_RE.match(line):
in_fence = not in_fence
out.append("")
continue
out.append("" if in_fence else line)
return "\n".join(out)
def slugify(text: str) -> str:
"""把标题/锚点文本转成可比对的 key。
**刻意不逐字复刻 GitHub/Gitea 的 slug 算法。** 各家的标点处理规则并不一致
(GitHub 会删 `+`/`:` 却保留 `、`/`:`,而「连续空格是折叠成一个连字符
还是每个空格一个连字符」也随实现而变)。照着某个实现写死,换个托管平台
就会批量误报,反而让真问题淹掉。
这里只保留「有效字符」:小写字母、数字、汉字。标点、空格、连字符**全部丢弃**。
于是 `八、配置系统:写时校验 + 读时兜底` 与链接里的
`八配置系统写时校验--读时兜底` 归一后相等 —— 章节被重命名时,
有效字符会变,锚点仍然照抓不误。
代价:仅标点不同的两个标题会被视为同一锚点(极端罕见,可接受)。
"""
return re.sub(r"[^0-9a-z\u4e00-\u9fff]", "", text.lower())
def is_external(target: str) -> bool:
"""外部链接(http: / mailto: 等)不检查。"""
return bool(re.match(r"^[a-zA-Z][a-zA-Z0-9+.-]*:", target))
def looks_like_placeholder(segment: str) -> bool:
return bool(PLACEHOLDER_RE.search(segment.strip()))
def discover(root: str) -> list[str]:
"""递归找出所有 .md,返回相对 root 的 posix 路径。"""
found: list[str] = []
for dirpath, dirnames, filenames in os.walk(root):
dirnames[:] = [d for d in dirnames if d not in SKIP_DIRS]
for fn in filenames:
if fn.lower().endswith((".md", ".markdown")):
rel = os.path.relpath(os.path.join(dirpath, fn), root)
found.append(rel.replace("\\", "/"))
return sorted(found)
_anchor_cache: dict[str, set[str]] = {}
def anchors_of(abs_path: str) -> set[str]:
"""一个 md 文件里所有可跳转的锚点(标题 + 显式 HTML 锚点)。"""
if abs_path not in _anchor_cache:
text = strip_code_blocks(read(abs_path))
got: set[str] = set()
for line in text.splitlines():
m = HEADING_RE.match(line)
if m:
got.add(slugify(m.group(2)))
for m in HTML_ANCHOR_RE.finditer(text):
got.add(slugify(m.group(1)))
_anchor_cache[abs_path] = got
return _anchor_cache[abs_path]
# ---------------------------------------------------------------- 各检查项
def check_links(root: str, files: list[str]) -> list[str]:
problems: list[str] = []
for rel in files:
abs_path = os.path.join(root, rel.replace("/", os.sep))
base_dir = os.path.dirname(abs_path)
text = strip_code_blocks(read(abs_path))
for lineno, line in enumerate(text.splitlines(), 1):
for m in LINK_RE.finditer(line):
target = m.group(2)
if is_external(target):
continue
# 纯页内锚点:指回本文件
if target.startswith("#"):
frag = target[1:]
if frag and slugify(frag) not in anchors_of(abs_path):
problems.append(f"{rel}:{lineno} 页内锚点失效 {target}")
continue
path_part, _, frag = target.partition("#")
if not path_part:
continue
resolved = os.path.normpath(os.path.join(base_dir, path_part))
if not os.path.exists(resolved):
problems.append(f"{rel}:{lineno} 链接目标不存在 {target}")
continue
if frag and resolved.lower().endswith((".md", ".markdown")):
if slugify(frag) not in anchors_of(resolved):
problems.append(f"{rel}:{lineno} 跨文件锚点失效 {target}")
return problems
def check_images(root: str, files: list[str]) -> list[str]:
problems: list[str] = []
for rel in files:
abs_path = os.path.join(root, rel.replace("/", os.sep))
base_dir = os.path.dirname(abs_path)
text = strip_code_blocks(read(abs_path))
for lineno, line in enumerate(text.splitlines(), 1):
for m in IMG_RE.finditer(line):
src = m.group(1)
if is_external(src):
continue
resolved = os.path.normpath(os.path.join(base_dir, src))
if not os.path.exists(resolved):
problems.append(f"{rel}:{lineno} 图片不存在 {src}")
return problems
def check_path_leaks(root: str, files: list[str]) -> list[str]:
"""扫绝对路径 —— 文档里出现多半是从本机命令里抄进来的(会连带泄漏用户名)。
命中后请**逐条人工判断**:占位符(`C:\\Users\\<用户名>`)已豁免,
但 `os.path.join(home, "AppData", ...)` 这类合法的路径发现代码不会命中
(它不含盘符或 `/home/` 前缀)。这里只报告,不自动修改。
"""
problems: list[str] = []
for rel in files:
abs_path = os.path.join(root, rel.replace("/", os.sep))
for lineno, line in enumerate(read(abs_path).splitlines(), 1):
for pat, label in LEAK_PATTERNS:
for m in pat.finditer(line):
seg = m.group(1) if m.groups() else ""
if seg and looks_like_placeholder(seg):
continue
problems.append(f"{rel}:{lineno} {label} {m.group(0)}")
return problems
def check_versions(root: str) -> list[str]:
"""版本号四处(__init__ / Dockerfile / README / CHANGELOG)必须一致。"""
found: dict[str, str] = {}
sources = [
(VERSION_INIT, INIT_VER_RE),
(DOCKERFILE, OCI_VER_RE),
("README.md", README_VER_RE),
(os.path.join("docs", "CHANGELOG.md"), CHANGELOG_VER_RE),
]
for rel, pat in sources:
p = os.path.join(root, rel)
if os.path.isfile(p):
m = pat.search(read(p))
if m:
found[rel.replace("\\", "/")] = m.group(1)
if not found:
return ["版本一致性: 一个版本号都没找到,检查脚本本身"]
if len(set(found.values())) > 1:
detail = "、".join("%s=%s" % (k, v) for k, v in sorted(found.items()))
return ["版本不一致: %s" % detail]
return []
def check_name_hardcode(root: str) -> list[str]:
"""产品名应走上下文变量(config.PROJECT_NAME),不该硬编码进模板/JS。"""
problems: list[str] = []
for d in HARDCODE_DIRS:
full = os.path.join(root, d)
if not os.path.isdir(full):
continue
for dirpath, _dirnames, filenames in os.walk(full):
for fn in filenames:
if not fn.lower().endswith((".html", ".js")):
continue
fp = os.path.join(dirpath, fn)
rel = os.path.relpath(fp, root).replace("\\", "/")
for i, line in enumerate(read(fp).splitlines(), 1):
for n in HARDCODE_NEEDLES:
if n in line:
problems.append(
f"{rel}:{i} 疑似硬编码产品名「{n}」(应走上下文变量)")
return problems
# ---------------------------------------------------------------- main
def main() -> int:
ap = argparse.ArgumentParser(description="文档自检")
ap.add_argument("--root", default=".", help="仓库根(默认当前目录)")
ap.add_argument("--no-fail", action="store_true",
help="即使发现问题也返回 0(仅用于人工查看报告)")
ap.add_argument("--strict", action="store_true", help=argparse.SUPPRESS)
ap.add_argument("--quiet", action="store_true", help="只打印问题")
args = ap.parse_args()
root = os.path.abspath(args.root)
if not os.path.isdir(root):
print("--root 不是目录: %s" % root)
return 1
files = discover(root)
if not files:
print("没找到任何 .md 文件。--root 是否正确?"
"(注意:Git Bash 的 /tmp/x 交给原生 Python 会变成 C:\\tmp\\x)")
return 1
sections = [
("内部链接与锚点", check_links(root, files)),
("图片引用", check_images(root, files)),
("绝对路径泄漏", check_path_leaks(root, files)),
("版本一致性", check_versions(root)),
("产品名硬编码", check_name_hardcode(root)),
]
total = sum(len(p) for _, p in sections)
if not args.quiet:
print("扫描 %d 个 Markdown 文件:" % len(files))
for rel in files:
print(" - %s" % rel)
print()
for title, problems in sections:
if args.quiet and not problems:
continue
print("=== %s ===" % title)
if problems:
for p in problems:
print(" [!!] %s" % p)
else:
print(" OK")
print()
print("RESULT: %d 处问题" % total)
if total == 0:
print("文档自检全部通过。")
# 默认「有问题就失败」——「报出 7 处问题却返回 0」是个静默无用的陷阱,
# 想只看报告请显式加 --no-fail。
if total and not args.no_fail:
return 1
return 0
if __name__ == "__main__":
sys.exit(main())
+93 -3
查看文件
@@ -5,18 +5,23 @@
"""端到端验收:对**运行中的**服务发真实 HTTP 请求,走完整登录/CSRF/API 链路。
与 tests 里用 Flask test_client 的冒烟测试互补——这里验证的是「真的起起来了、
真的能登录、真的能取到数」,适合部署到局域网后随手跑一遍。
与 tools/smoke.py 的分工:smoke 用 Flask test_client 直接渲染模板、不发网络请求;
本脚本确认真的是「起起来了、能登录、能取到数」,适合部署到局域网后随手跑一遍。
用法:
python tools/check_live.py # 默认 http://127.0.0.1:8848
python tools/check_live.py --base http://192.168.1.50:8848 # 换成你的部署主机
python tools/check_live.py -u admin -p 你的密码
python tools/check_live.py --as alice:她的密码 # 额外跑一遍**普通账号**的越权面
python tools/check_live.py --from 2026-09-08 --to 2026-09-14
`--as` 那一节会真的发越权请求(改调度 / 改采集参数 / 读日志),
期望全部被拒;不会创建或删除任何账号,所以请自己先准备一个普通账号。
退出码:0 全通过;1 有失败项(会打印失败清单)。
注意:脚本会读取窗口数据但**不写库**(不触发采集、不改配置),可安全反复运行。
注意:脚本会读窗口数据、会走登录(登录本身会更新 last_login_at),
但**不触发采集、不改任何配置**,可安全反复运行。
"""
from __future__ import annotations
@@ -313,6 +318,10 @@ def run(L: Live, user: str, pwd: str, frm: str, to: str) -> None:
chk("settings 回传实例级键清单", isinstance(stj.get("_globalKeys"), list)
and bool(stj.get("_globalKeys")), "%s" % stj.get("_globalKeys"))
chk("settings 标明能否改实例级配置", stj.get("_canEditGlobal") is True)
chk("settings 回传个人可写键清单(应为 cookie/user_agent)",
set(stj.get("_userKeys") or []) == {"cookie", "user_agent"},
"%s" % stj.get("_userKeys"))
chk("settings 标明角色", stj.get("_role") == "admin", "%s" % stj.get("_role"))
chk("配置页 HTML 不含 cookie 明文", "eyJ" not in L.get("/config")[1])
# 密文形态:v1.<b64salt>.<b64nonce>.<b64ct>.<b64tag>,恰好用正则判定,
# 免得把版本号 "v1.2.0" 当成泄漏(这两者前缀撞车)
@@ -417,6 +426,74 @@ def run(L: Live, user: str, pwd: str, frm: str, to: str) -> None:
"status=%s" % code)
def run_nonadmin(base: str, timeout: int, db_path: str, user: str, pwd: str) -> None:
"""普通账号的越权面(真实 HTTP 链路,--as 才跑)。
规则只有一条:普通账号**只能维护本人凭证**,其余配置 / 日志 / 用户管理
全部不可达。期望值是 403(页面)与 400(写配置)—— 不是「看得到但改不了」,
更不是「写进去但只对自己生效」。
"""
print("== 12. 普通账号越权面(--as %s) ==" % user)
L = Live(base, timeout, db_path)
st, _, _, _ = L.login(user, pwd)
chk("普通账号登录成功", st in (200, 302), "status=%s" % st)
st, html = L.get("/")
if st != 200 or "概览" not in html:
chk("普通账号登录后能看到概览", False, "status=%s(后续断言已跳过)" % st)
return
chk("普通账号登录后能看到概览", True)
chk("导航不出现「日志管理」", "日志管理" not in html)
chk("导航不出现「用户管理」", "用户管理" not in html)
# 可达页面(都只渲染本人数据)
for p, kw in [("/records", "记录"), ("/tasks", "任务"),
("/config", "配置"), ("/profile", "个人")]:
st, body = L.get(p)
chk("GET %-9s 普通账号=200" % p, st == 200 and kw in body, "status=%s" % st)
st, _ = L.get("/dashboard")
chk("GET /dashboard 普通账号=200", st == 200, "status=%s" % st)
# 不可达:日志与用户管理
for p in ("/logs", "/logs/tail?lines=10", "/users", "/api/users"):
st, _ = L.get(p)
chk("GET %-21s 普通账号=403" % p, st == 403, "status=%s" % st)
# 角色标记:页面之外还有静态页(大屏)与前端要靠它决定显隐
stj = L.jget("/api/settings")
chk("/api/settings _role=user", stj.get("_role") == "user", "%s" % stj.get("_role"))
chk("/api/settings _canEditGlobal=False", stj.get("_canEditGlobal") is False)
mf = L.jget("/api/manifest")
chk("/api/manifest role=user(大屏据此隐掉日志入口)",
mf.get("role") == "user", "%s" % mf.get("role"))
sta = L.jget("/api/status")
chk("/api/status is_admin=False", sta.get("is_admin") is False, "%s" % sta.get("is_admin"))
chk("/api/status can_edit_schedule=False", sta.get("can_edit_schedule") is False)
# 越权写:调度 / 采集参数 / 实例级键 -> 400
csrf = L.form_csrf("/config")
chk("拿到普通账号自己的 CSRF", bool(csrf))
for key, val in (("schedule_times", "23:59"), ("schedule_enabled", "0"),
("page_size", "1000"), ("ssl_verify", "0"),
("api_base", "http://evil.invalid"), ("allow_register", "0")):
st, body = L.post("/api/settings", {key: val}, csrf=csrf, as_json=True)
chk("越权 POST %-16s =400" % key, st == 400, "status=%s" % st)
chk(" └ 且点名 %s" % key, key in body)
# 本人 UA 必须写得进去;写回原值,不给对方留副作用
cur_ua = str(stj.get("user_agent") or "")
st, body = L.post("/api/settings", {"user_agent": cur_ua}, csrf=csrf, as_json=True)
chk("本人 user_agent 可写=200", st == 200, "status=%s %s" % (st, body[:100]))
# 页面只给凭证表单,采集参数与调度都渲染成只读
st, cf = L.get("/config")
chk("配置页有凭证表单", 'id="formCred"' in cf)
chk("配置页无采集参数表单", 'id="formCollect"' not in cf)
chk("配置页无实例级设置表单", 'id="formGlobal"' not in cf)
st, tk = L.get("/tasks")
chk("任务页调度只读(没有保存按钮)", "保存调度配置" not in tk)
chk("任务页标注调度仅管理员可改", "仅管理员可改" in tk)
def main() -> int:
ap = argparse.ArgumentParser(description="对运行中的用量门户做端到端验收")
ap.add_argument("--base", default="http://127.0.0.1:8848", help="服务地址")
@@ -429,12 +506,25 @@ def main() -> int:
"验证码策略为 always 时用它取答案以完成自动登录;"
"指向不存在的文件则跳过需要验证码的登录")
ap.add_argument("--timeout", type=int, default=20)
ap.add_argument("--as", dest="as_user", default=None, metavar="USER:PASS",
help="额外用一个**普通账号**跑一遍越权验收(第 12 节)。"
"不会创建/删除账号,请自己先备好一个普通账号")
a = ap.parse_args()
db_path = a.db or os.path.join(
os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "data", "usage.sqlite")
print("目标:%s 窗口:%s ~ %s\n验证码答案源:%s\n" % (a.base, a.frm, a.to, db_path))
run(Live(a.base, a.timeout, db_path), a.user, a.password, a.frm, a.to)
if a.as_user:
if ":" not in a.as_user:
print(" [FAIL] --as 需要写成 用户名:密码")
FAILS.append("--as 参数格式")
else:
nu, np_ = a.as_user.split(":", 1)
try:
run_nonadmin(a.base, a.timeout, db_path, nu, np_)
except Exception as e: # noqa: BLE001
chk("第 12 节执行未抛异常", False, "%s: %s" % (type(e).__name__, e))
print("\nRESULT: ok=%d fail=%d" % (OK, FAIL))
if FAILS:
print("失败项:%s" % "、".join(FAILS))
+4 -3
查看文件
@@ -184,9 +184,10 @@ def build(out_dir: str, days: int, seed: int, admin_password: str,
# 经 set_secret 落库 = 真的走一遍加密,所以示例库里也是密文。
db.set_secret(conn, "cookie", DEMO_COOKIE, admin_uid)
db.set_secret(conn, "cookie", DEMO_COOKIE_2, demo_uid)
# 两个账号各有一套调度时刻,界面上能看出「每人可改自己的」
db.set_setting(conn, "schedule_times", "09:00,17:00", admin_uid)
db.set_setting(conn, "schedule_times", "08:30,20:00", demo_uid)
# 调度与采集参数是**实例级**的(v1.3.0 起普通账号只读),所以只写一次;
# 这里刻意不按账号各写一份 —— set_setting 会把全局键归到 user_id=0,
# 写两次只会后一次覆盖前一次,看起来「每人一套」其实没有。
db.set_setting(conn, "schedule_times", "09:00,17:00", 0)
rng = random.Random(seed)
now = datetime.now().replace(second=0, microsecond=0)
+57 -19
查看文件
@@ -49,6 +49,14 @@ CAPTURES = [
("10-profile.png", "/profile", "个人中心", True),
]
# v1.3.0 起「任务管理 / 配置管理」在普通账号下是**只读**形态,
# 与管理员看到的表单不是同一个页面。文档要同时展示两种视角,
# 所以单独跑一趟普通账号的登录会话(换个 context = 干净 Cookie)。
USER_CAPTURES = [
("03b-tasks-user.png", "/tasks", "任务管理(普通账号:调度只读)"),
("04b-config-user.png", "/config", "配置管理(普通账号:仅凭证可改)"),
]
def _find_browser() -> str | None:
"""找一个可用的 Chromium 可执行文件。
@@ -118,6 +126,9 @@ def main() -> int:
ap.add_argument("--base", default="http://127.0.0.1:8849")
ap.add_argument("-u", "--user", default="admin")
ap.add_argument("-p", "--password", default="admin123")
ap.add_argument("--user2", default="demo",
help="普通账号用户名(截只读视角用;设为空则跳过)")
ap.add_argument("--password2", default="admin123")
ap.add_argument("--out", default=os.path.join(BASE, "data", "shots"))
ap.add_argument("--db", default=None,
help="SQLite 路径(默认 <repo>/data/usage.sqlite),用于取验证码答案")
@@ -148,38 +159,42 @@ def main() -> int:
page.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
page.on("pageerror", lambda e: errors.append(str(e)))
def shoot(name, path, label):
def shoot(name, path, label, pg=None):
pg = pg or page
errors.clear()
page.goto(a.base + path, wait_until="networkidle")
page.wait_for_timeout(900) # 等 ECharts / 表格渲染稳下来
page.screenshot(path=os.path.join(a.out, name), full_page=a.full)
pg.goto(a.base + path, wait_until="networkidle")
pg.wait_for_timeout(900) # 等 ECharts / 表格渲染稳下来
pg.screenshot(path=os.path.join(a.out, name), full_page=a.full)
js_err = [e for e in errors if "favicon" not in e.lower()]
if js_err:
problems.append("%s: %s" % (label, js_err[:3]))
print("[ok] %-22s %s%s" % (name, label,
"" if not js_err else " [JS错误] " + " | ".join(js_err[:3])))
def login(pg, ctx, user, password):
"""登录并跨过验证码(策略为 always 时从本地库取答案)。"""
pg.goto(a.base + "/login", wait_until="networkidle")
pg.fill('input[name=username]', user)
pg.fill('input[name=password]', password)
if pg.query_selector('input[name=captcha]'):
ans = _captcha_answer(ctx, db_path, "login")
if not ans:
print("[FAIL] 需要验证码但取不到答案(--db 是否指向本实例的库?):%s" % db_path)
return False
pg.fill('input[name=captcha]', ans)
print("[ok] 已用库里的答案通过验证码(%s)" % user)
pg.click('button[type=submit]')
pg.wait_for_load_state("networkidle")
return "/login" not in pg.url
# 1) 未登录的两页
for name, path, label, need_auth in CAPTURES:
if need_auth:
break
shoot(name, path, label)
# 2) 登录(策略为 always 时自动解验证码)
page.goto(a.base + "/login", wait_until="networkidle")
page.fill('input[name=username]', a.user)
page.fill('input[name=password]', a.password)
if page.query_selector('input[name=captcha]'):
ans = _captcha_answer(ctx, db_path, "login")
if not ans:
print("[FAIL] 需要验证码但取不到答案(--db 是否指向本实例的库?):%s" % db_path)
br.close()
return 1
page.fill('input[name=captcha]', ans)
print("[ok] 已用库里的答案通过验证码")
page.click('button[type=submit]')
page.wait_for_load_state("networkidle")
if "/login" in page.url:
# 2) 以管理员登录(策略为 always 时自动解验证码)
if not login(page, ctx, a.user, a.password):
print("[FAIL] 登录失败,后续截图无意义")
br.close()
return 1
@@ -208,6 +223,29 @@ def main() -> int:
problems.append("大屏交互: %s" % js_err[:2])
break
# 5) 换一个干净 context,用普通账号再跑一趟只读视角
if a.user2:
ctx2 = br.new_context(viewport={"width": a.width, "height": a.height},
device_scale_factor=2, locale="zh-CN")
page2 = ctx2.new_page()
page2.on("console", lambda m: errors.append(m.text) if m.type == "error" else None)
page2.on("pageerror", lambda e: errors.append(str(e)))
if not login(page2, ctx2, a.user2, a.password2):
print("[warn] 普通账号 %s 登录失败,跳过只读视角截图" % a.user2)
problems.append("普通账号 %s 登录失败" % a.user2)
else:
for name, path, label in USER_CAPTURES:
shoot(name, path, label, pg=page2)
# 顺带把越权面再验一次:普通账号访问这些必须不是 200
for probe in ("/logs", "/logs/tail", "/users"):
r = page2.goto(a.base + probe, wait_until="domcontentloaded")
code = r.status if r else 0
ok = code in (403, 401)
print("[%s] 越权面 %-12s -> %s" % ("ok" if ok else "!!", probe, code))
if not ok:
problems.append("越权面未关死:%s 返回 %s" % (probe, code))
ctx2.close()
br.close()
print("\n截图目录:%s" % a.out)
+156 -18
查看文件
@@ -11,13 +11,18 @@
因此能覆盖到「页面模板渲染是否正确」,且不需要先起服务、不需要密码。
覆盖内容:
1. 全页面渲染(含 /users,需管理员身份)——模板报错会直接暴露成 500
1. 全页面渲染(含 /users 与 /logs,均需管理员身份)——模板报错会直接暴露成 500
2. 模板未渲染残留(HTML 里出现 {{ / {% 说明有变量名写错)
3. 历史缺陷防回归(见 4. 的 ①~⑭)
4. 多用户:数据隔离 / 凭证保密 / 注册与验证码 / 权限边界
3. 历史缺陷防回归(见 5. 的 ①~⑭)
4. 多用户:数据隔离 / 凭证保密 / 注册与验证码 / **普通账号的越权面**(4. 与 4b.)
5. CSV 导出可被标准 csv 解析、列数一致
6. 页面 HTML 里的 class 与 app.css 的选择器做差集(抓类名拼写错误)
权限模型(改断言前先读这一行):
普通账号**只能写** `config.USER_EDITABLE_KEYS`(本人的 cookie / user_agent);
调度、采集参数、接口地址、注册策略全部只有管理员能写,且一律存在实例级
`user_id=0`。所以「越权写」的期望结果是 **400**,而不是「写进去但看不到」。
写库说明:会写少量 audit_log 行;另外会**临时**建两个普通账号
(一个用来验权限边界,一个用来走完整注册链路),无论成功失败都在 finally 里删掉。
不会改动任何用量数据。
@@ -107,6 +112,17 @@ def run() -> None:
return
ADMIN = admin_row["id"]
# 实例级配置快照:整轮跑完必须一条不少。
# 历史缺陷:清理语句写成 `user_id NOT IN (SELECT id FROM users)`,会把
# user_id=0(实例级)当成孤儿一起删掉 —— 表现为「跑一次 smoke,全实例的
# 调度/采集参数被重置」,而且不报任何错。所以这里前后各取一次快照。
inst_before = {r["key"] for r in conn.execute("SELECT key FROM settings WHERE user_id=0")}
chk("实例级配置非空(否则下面那条断言会空转)", len(inst_before) > 0,
"keys=%d" % len(inst_before))
chk("实例级不含凭证类键(cookie/user_agent 恒为个人级)",
not (inst_before & config.USER_EDITABLE_KEYS),
"混入=%s" % sorted(inst_before & config.USER_EDITABLE_KEYS))
# ---------------- 1. 未登录 ----------------
print("== 1. 未登录:受保护页应跳登录、API 应 401 ==")
with app.test_client() as cli:
@@ -241,15 +257,31 @@ def run() -> None:
# User-Agent 本身不是秘密,新账号拿到 DEFAULTS 里的**通用** UA 是对的;
# 要守住的是「不能继承别人存下来的那一份」。用一个哨兵值把这点钉死:
sentinel = "SMOKE-SENTINEL-UA/%s" % _rand()
saved_ua_inst = db.get_setting(conn, "user_agent", "", 0)
# 注意:get_setting 在没有行时会回落到 DEFAULTS,所以「还原是否成功」
# 要拿**行为**比(读出来一样),而不是拿「行在不在」比 —— 这两件事
# 在实例级是分开的,混起来会让断言永远失败。
before_ua = db.get_setting(conn, "user_agent", "", 0)
had_row = conn.execute("SELECT 1 FROM settings WHERE user_id=0 AND key='user_agent'"
).fetchone() is not None
saved_ua_inst = before_ua if had_row else None
db.set_setting(conn, "user_agent", sentinel, 0) # 写实例级
chk("② 实例级放哨兵后,新账号仍看不到它",
db.get_setting(conn, "user_agent", "", VIEWER) != sentinel,
"new=%s…" % (db.get_setting(conn, "user_agent", "", VIEWER) or "")[:22])
chk("② 哨兵在实例级确实生效(证明上面的断言不是在空跑)",
db.get_setting(conn, "user_agent", "", 0) == sentinel)
db.set_setting(conn, "user_agent", saved_ua_inst, 0) # 还原
chk("② 已还原实例级 UA", db.get_setting(conn, "user_agent", "", 0) == saved_ua_inst)
# 还原:**原本没有这一行就删掉**。实例级本不该存在凭证类键
# (v1.3.0 起 cookie / user_agent 恒为个人级),写回空串只会留下
# 一个多余行,下次跑就会让「实例级不含凭证键」的断言失败。
if had_row:
db.set_setting(conn, "user_agent", before_ua, 0)
else:
conn.execute("DELETE FROM settings WHERE user_id=0 AND key='user_agent'")
chk("② 已还原实例级 UA(读出来与放哨兵前一致)",
db.get_setting(conn, "user_agent", "", 0) == before_ua)
chk("② 还原后实例级不留 user_agent 行(原本有则保留)",
(conn.execute("SELECT 1 FROM settings WHERE user_id=0 AND key='user_agent'"
).fetchone() is not None) == had_row)
# 普通配置应当能回落到实例级(否则每个新账号都拿到空配置)
chk("② 普通配置仍回落实例级",
db.get_setting(conn, "page_size", None, VIEWER) ==
@@ -362,13 +394,22 @@ def run() -> None:
chk("⑦ totals(uid=0) 不含任何人的数据",
query.totals(conn, 0)["records"] == 0)
finally:
# 哨兵 UA 一定要还原(否则下次真采集会带着测试字符串发出去)
# 哨兵 UA 一定要还原(否则下次真采集会带着测试字符串发出去)。
# 原本实例级没有这一行时,**删掉**而不是写回空串 —— 见第 ② 条断言。
if saved_ua_inst is not None:
db.set_setting(conn, "user_agent", saved_ua_inst, 0)
else:
conn.execute("DELETE FROM settings WHERE user_id=0 AND key='user_agent'")
for name in created:
conn.execute("DELETE FROM users WHERE username=?", (name,))
conn.execute("DELETE FROM settings WHERE user_id NOT IN (SELECT id FROM users)")
conn.execute("DELETE FROM usage_records WHERE user_id NOT IN (SELECT id FROM users)")
# 注意 `user_id<>0` 不能省:user_id=0 是**实例级配置**(调度、采集参数、
# 接口地址、注册策略都在那里),它不属于任何账号,所以
# `NOT IN (SELECT id FROM users)` 会把它当孤儿一起删掉 ——
# 表现成「跑一次 smoke,全实例的配置被重置」,且不报任何错。
conn.execute("DELETE FROM settings WHERE user_id<>0"
" AND user_id NOT IN (SELECT id FROM users)")
conn.execute("DELETE FROM usage_records WHERE user_id<>0"
" AND user_id NOT IN (SELECT id FROM users)")
# 确认清理干净
left = conn.execute("SELECT COUNT(*) FROM users WHERE username LIKE 'smoke\\_%' ESCAPE '\\'"
@@ -376,7 +417,10 @@ def run() -> None:
chk("3. 临时账号已清理", left == 0, "残留=%d" % left)
# ---------------- 4. 普通账号的权限边界 ----------------
print("== 4. 非管理员:/users 必须 403,导航不出现该入口 ==")
# 规则只有一条(config.writable_by):普通账号只能写本人的 cookie / user_agent,
# 其余(调度、采集参数、接口地址、注册策略)一律 400。页面隐藏 / disabled
# 只是「不给出误导性按钮」,真正的闸门在服务端,所以这里全部走真实请求。
print("== 4. 非管理员:越权面必须全部关死 ==")
viewer2 = "smoke_w_%s" % _rand()
try:
V2 = _mk_user(viewer2)
@@ -386,18 +430,112 @@ def run() -> None:
chk("GET /users 非管理员=403", st == 403, "status=%s" % st)
st, _ = page(cli, "/api/users")
chk("GET /api/users 非管理员=403", st == 403, "status=%s" % st)
st, html = page(cli, "/")
chk("概览导航不含「用户管理」", "用户管理" not in html)
chk("普通账号导航含「个人中心」入口", 'class="who"' in html)
for p in ("/", "/records", "/tasks", "/logs", "/config", "/profile"):
st, _ = page(cli, p)
chk("GET %-10s 非管理员=200" % p, st == 200, "status=%s" % st)
# 日志尾部是管理员专属
# ④ 日志是**实例级**运行信息(含数据库路径 / 账号名 / 来源 IP),
# 普通账号整页 403 —— 不是「只看到自己那份」。
st, _ = page(cli, "/logs")
chk("GET /logs 非管理员=403", st == 403, "status=%s" % st)
st, _ = page(cli, "/logs/tail?lines=10")
chk("GET /logs/tail 非管理员=403", st == 403, "status=%s" % st)
# ⑤ 其余页面(都只渲染本人数据)必须照常能开
for p in ("/", "/dashboard", "/records", "/tasks", "/config", "/profile"):
st, _ = page(cli, p)
chk("GET %-11s 非管理员=200" % p, st == 200, "status=%s" % st)
st, html = page(cli, "/")
chk("概览导航不含「用户管理」", "用户管理" not in html)
chk("概览导航不含「日志管理」", "日志管理" not in html)
chk("普通账号导航含「个人中心」入口", 'class="who"' in html)
# ⑥ 越权写:调度 / 采集参数 / 实例级键,逐个试,全部必须 400
keep_times = db.get_setting(conn, "schedule_times", "", 0)
for key, val in (("schedule_times", "23:59"),
("schedule_enabled", "0"),
("catch_up", "0"),
("page_size", "1000"),
("timeout", "300"),
("ssl_verify", "0"),
("max_prompt", "0"),
("api_base", "http://evil.invalid"),
("allow_register", "0")):
st, body = page(cli, "/api/settings", method="POST", json={key: val},
headers={"X-CSRF-Token": "smoke-csrf-token"})
chk("⑥ 越权写 %-16s =400" % key, st == 400, "status=%s" % st)
chk(" └ 报错里点名 %s" % key, key in body)
chk("⑥ 越权尝试确实没落库(schedule_times 未变)",
db.get_setting(conn, "schedule_times", "", 0) == keep_times)
chk("⑥ 实例级 api_base 未被改写",
"evil" not in db.get_setting(conn, "api_base", "", 0))
# ⑦ 但本人凭证必须写得进去(否则普通账号根本没法采集)
st, _ = page(cli, "/api/settings", method="POST",
json={"user_agent": "SMOKE-VIEWER-UA/1.0"},
headers={"X-CSRF-Token": "smoke-csrf-token"})
chk("⑦ 普通账号写本人 user_agent=200", st == 200, "status=%s" % st)
chk("⑦ 且只写进了自己名下",
db.get_setting(conn, "user_agent", "", V2) == "SMOKE-VIEWER-UA/1.0")
# ⑧ 任务页给普通账号渲染的是只读表,且没有「保存调度配置」按钮
st, tk = page(cli, "/tasks")
chk("⑧ 任务页标注调度只读", "仅管理员可改" in tk)
chk("⑧ 任务页无调度保存按钮", "保存调度配置" not in tk)
# ⑨ 配置页对普通账号只给凭证表单,采集参数渲染成只读表
st, cf = page(cli, "/config")
chk("⑨ 配置页有凭证表单", 'id="formCred"' in cf)
chk("⑨ 配置页无采集参数表单", 'id="formCollect"' not in cf)
chk("⑨ 配置页无实例级设置表单", 'id="formGlobal"' not in cf)
chk("⑨ 配置页说明范围", "唯一可以修改" in cf)
# ⑩ /api/status 的角色字段(大屏与前端靠它显隐管理员入口)
st, sj = page(cli, "/api/status")
chk("⑩ GET /api/status 普通账号=200", st == 200, "status=%s" % st)
if st == 200:
j = json.loads(sj)
chk("⑩ is_admin=False", j.get("is_admin") is False, "%s" % j.get("is_admin"))
chk("⑩ can_edit_schedule=False", j.get("can_edit_schedule") is False)
chk("⑩ can_view_logs=False", j.get("can_view_logs") is False)
finally:
conn.execute("DELETE FROM users WHERE username=?", (viewer2,))
conn.execute("DELETE FROM settings WHERE user_id NOT IN (SELECT id FROM users)")
# `user_id<>0` 是必须的:0 是实例级配置,不能当孤儿清理(见第 3 节的说明)
conn.execute("DELETE FROM settings WHERE user_id<>0"
" AND user_id NOT IN (SELECT id FROM users)")
# ---------------- 4b. 全局键的落库位置 ----------------
# 这一节盯的是「管理员改了但只有自己生效」这类**静默** bug:
# 全局键若被写进管理员的 user_id,其它账号读取时会回落到 DEFAULTS,
# 表现成「设置莫名其妙不生效」,而且不报任何错。
print("== 4b. 全局键必须落在实例级 user_id=0 ==")
chk("schedule_times 是全局键", config.is_global_key("schedule_times"))
chk("page_size 是全局键", config.is_global_key("page_size"))
chk("cookie 不是全局键(本人凭证)", not config.is_global_key("cookie"))
chk("slot:* 仍是个人级(每人各自记今天跑过没)",
not config.is_global_key("slot:09:00"))
chk("普通账号只被允许写 cookie/user_agent",
config.writable_by("cookie", False) and config.writable_by("user_agent", False)
and not config.writable_by("schedule_times", False)
and not config.writable_by("page_size", False))
with app.test_client() as cli:
login(cli, ADMIN)
same = db.get_setting(conn, "schedule_times", "", 0)
st, _ = page(cli, "/api/settings", method="POST",
json={"schedule_times": same or "09:00"},
headers={"X-CSRF-Token": "smoke-csrf-token"})
chk("管理员写 schedule_times=200", st == 200, "status=%s" % st)
chk("只存在实例级那一份",
conn.execute("SELECT COUNT(*) FROM settings WHERE user_id=0"
" AND key='schedule_times'").fetchone()[0] == 1)
chk("管理员名下不留个人级副本(否则别人读不到)",
conn.execute("SELECT COUNT(*) FROM settings WHERE user_id<>0"
" AND key='schedule_times'").fetchone()[0] == 0)
# /api/status 是大屏与前端判断角色用的接口,必须真的能开且角色正确
# (它曾经因为改字段时引用了未定义的变量而 500,两层测试都没覆盖到)
st, sj = page(cli, "/api/status")
chk("GET /api/status 管理员=200", st == 200, "status=%s" % st)
if st == 200:
j = json.loads(sj)
chk("└ is_admin=True", j.get("is_admin") is True, "%s" % j.get("is_admin"))
chk("└ can_edit_schedule=True", j.get("can_edit_schedule") is True)
chk("└ 凭证只回「有没有 / 多少字符」",
all(k in j for k in ("cookie_set", "cookie_chars", "cookie_broken"))
and "cookie" not in j)
# 收尾自检:实例级配置必须还在(对照开头那份快照)
inst_after = {r["key"] for r in conn.execute("SELECT key FROM settings WHERE user_id=0")}
chk("跑完整轮 smoke 后,实例级配置一条不少", inst_before <= inst_after,
"丢失=%s" % sorted(inst_before - inst_after))
conn.close()
# ---------------- 5. 历史缺陷防回归 ----------------