#!/usr/bin/env python3 """多用户真实验证：主机侧以 codexsvc 身份运行，直接对生产库 + 本地 HTTP API 验证。 用法（主机上）： python3 /opt/codex-app-server/deploy/verify-multiuser.py [--base URL] [--db PATH] 默认 --base http://127.0.0.1:8765，--db /var/lib/codex-app-server/control.db （Postgres 后端时 db 参数只作占位，连接参数走 CLOUD_DB_* 环境变量， 与 server/worker 一致；SQLite 后端时就是库文件路径）。 安全与防误伤（写进逻辑里，不是只写在注释里）： - 测试用户、项目、任务全部用 multiuser-test- 前缀（项目 id 固定为 multiuser-test-p1 / multiuser-test-p2，用户按用户名 multiuser-test-* 查找），清理时只删这些前缀的行，绝不碰其它生产数据。 - 测试任务创建时都带一个永不存在的依赖 multiuser-test-never-completes： worker 只派依赖全部 completed 的 queued 任务（见 cloud/queue.py 的 ready_tasks），所以这些测试任务永远不会被生产 worker 认领去跑模型—— 不花模型钱、不改任何仓库。验证完任务会被删掉，依赖本身只是一个 字符串，不对应任何真实任务行。 - API key 只存在内存里：创建用户时 AuthStore 返回的明文 key 只保存在 本进程的变量里，用完即弃；脚本全程只打印每项 PASS/FAIL 与总数， 绝不打印任何 API key；任务 prompt、用户 id 等内容也不打印，只打印 检查名与结果。 - 审计日志是 append-only 的：验证过程产生的审计条目（建项目、设成员、 建任务、merge 等）会留在 audit_log 里，脚本不删审计表——这是故意的， 审计就该留痕；README 里也写清了这一点。 - 清理失败时脚本退出非零并列出残留测试 id（不含 key），不静默假装干净。 - 本脚本不跑真实模型、不推代码、不改生产项目与任务，只动 multiuser-test-* 的行；跑之前不需要停 worker（见上，测试任务不会被认领）。 检查项（预期值写死在断言里，实际不符即 FAIL）： 1. owner 建 P1、P2 成功（201） 2. owner 建任务成功（201，contributor 建任务也成功） 3. viewer 建任务 403；非成员（viewer 对 P2）建任务 403 4. viewer 读 P1 任务成功（200） 5. viewer 读 P2 任务 403（跨项目隔离：P1 成员不是 P2 成员） 6. maintainer 对 completed 测试任务调 merge 成功（200，状态变 merged） 7. contributor 对 completed 测试任务调 merge 403 8. maintainer 对 queued 测试任务调 merge 409（状态不对） 9. 无 key 调 API 401 """ import argparse import json import os import sys import urllib.error import urllib.request sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) from cloud.auth import AuthStore # noqa: E402 from cloud.db import connect as db_connect # noqa: E402 PREFIX = "multiuser-test-" P1 = PREFIX + "p1" P2 = PREFIX + "p2" NEVER_DEP = PREFIX + "never-completes" RESULTS = [] def check(name, ok): RESULTS.append((name, bool(ok))) print("%s %s" % ("PASS" if ok else "FAIL", name)) def req(base, method, path, key=None, body=None): data = json.dumps(body).encode("utf-8") if body is not None else None r = urllib.request.Request(base + path, data=data, method=method) if key is not None: r.add_header("X-API-Key", key) if body is not None: r.add_header("Content-Type", "application/json") try: with urllib.request.urlopen(r, timeout=15) as resp: raw = resp.read() return resp.status, (json.loads(raw) if raw else {}) except urllib.error.HTTPError as e: raw = e.read() try: payload = json.loads(raw) if raw else {} except ValueError: payload = {} return e.code, payload except Exception as e: return None, {"error": str(e)} def main(): ap = argparse.ArgumentParser(description="multiuser verification") ap.add_argument("--base", default="http://127.0.0.1:8765") ap.add_argument("--db", default="/var/lib/codex-app-server/control.db") args = ap.parse_args() base = args.base.rstrip("/") auth = AuthStore(args.db) created_task_ids = [] created_user_ids = [] try: # -- 预清理：上一轮残留的测试行（可重复跑）------------------- conn = db_connect(args.db) try: conn.execute( "DELETE FROM tasks WHERE project_id LIKE ?", (PREFIX + "%",)) conn.execute( "DELETE FROM project_members WHERE project_id LIKE ?", (PREFIX + "%",)) conn.execute( "DELETE FROM projects WHERE id LIKE ?", (PREFIX + "%",)) conn.execute( "DELETE FROM users WHERE username LIKE ?", (PREFIX + "%",)) conn.commit() finally: conn.close() # -- 建 4 个测试用户（key 只在内存里）------------------------ users = {} for role in ("owner", "maintainer", "contributor", "viewer"): uid, key = auth.create_user(PREFIX + role) users[role] = (uid, key) created_user_ids.append(uid) # -- 建 2 个测试项目（owner 建，P2 只有 owner 是成员）-------- s, p1 = req(base, "POST", "/projects", users["owner"][1], {"id": P1, "name": P1}) check("owner 创建项目 P1（201）", s == 201) s, p2 = req(base, "POST", "/projects", users["owner"][1], {"id": P2, "name": P2}) check("owner 创建项目 P2（201）", s == 201) if s != 201 or p1.get("id") != P1: print("项目创建失败，无法继续验证") return 1 for role in ("maintainer", "contributor", "viewer"): auth.set_member(P1, users["owner"][0], users[role][0], role) # -- 建任务权限 ---------------------------------------------- # 全部测试任务带永不满足的依赖，生产 worker 绝不会认领（见模块注释）。 s, t_owner = req(base, "POST", "/tasks", users["owner"][1], { "project_id": P1, "prompt": "multiuser test task", "depends_on": [NEVER_DEP]}) check("owner 在 P1 建任务（201）", s == 201) if s == 201: created_task_ids.append(t_owner["id"]) s, t_contrib = req(base, "POST", "/tasks", users["contributor"][1], { "project_id": P1, "prompt": "multiuser test task", "depends_on": [NEVER_DEP]}) check("contributor 在 P1 建任务（201）", s == 201) if s == 201: created_task_ids.append(t_contrib["id"]) s, _ = req(base, "POST", "/tasks", users["viewer"][1], { "project_id": P1, "prompt": "multiuser test task"}) check("viewer 在 P1 建任务（403）", s == 403) s, _ = req(base, "POST", "/tasks", users["viewer"][1], { "project_id": P2, "prompt": "multiuser test task"}) check("非成员 viewer 在 P2 建任务（403）", s == 403) # -- 读任务权限与跨项目隔离 ---------------------------------- if created_task_ids: tid = created_task_ids[0] s, _ = req(base, "GET", "/tasks/" + tid, users["viewer"][1]) check("viewer 读 P1 任务（200）", s == 200) s, t_p2 = req(base, "POST", "/tasks", users["owner"][1], { "project_id": P2, "prompt": "multiuser test task", "depends_on": [NEVER_DEP]}) check("owner 在 P2 建任务（201）", s == 201) if s == 201: created_task_ids.append(t_p2["id"]) s, _ = req(base, "GET", "/tasks/" + t_p2["id"], users["viewer"][1]) check("P1 成员 viewer 读 P2 任务（403，跨项目隔离）", s == 403) # -- merge 权限（需先把测试任务直接置 completed，不跑模型）-- conn = db_connect(args.db) try: for tid in created_task_ids: conn.execute( "UPDATE tasks SET status = 'completed' " "WHERE id = ?", (tid,)) conn.commit() finally: conn.close() if len(created_task_ids) >= 1: s, merged = req(base, "POST", "/tasks/" + created_task_ids[0] + "/merge", users["maintainer"][1]) check("maintainer 对 completed 任务调 merge（200 且变 merged）", s == 200 and merged.get("status") == "merged") if len(created_task_ids) >= 2: s, _ = req(base, "POST", "/tasks/" + created_task_ids[1] + "/merge", users["contributor"][1]) check("contributor 对 completed 任务调 merge（403）", s == 403) # queued 任务调 merge 应为 409：再建一个（不置 completed）。 s, t_queued = req(base, "POST", "/tasks", users["owner"][1], { "project_id": P1, "prompt": "multiuser test task", "depends_on": [NEVER_DEP]}) if s == 201: created_task_ids.append(t_queued["id"]) s, _ = req(base, "POST", "/tasks/" + t_queued["id"] + "/merge", users["maintainer"][1]) check("maintainer 对 queued 任务调 merge（409）", s == 409) # -- 无 key --------------------------------------------------- s, _ = req(base, "GET", "/tasks/" + created_task_ids[0] if created_task_ids else "/tasks/none") check("无 key 调 API（401）", s == 401) finally: # -- 清理：只删 multiuser-test-* 的行，审计表不动 ------------ leftover = [] try: conn = db_connect(args.db) try: for tid in created_task_ids: conn.execute("DELETE FROM tasks WHERE id = ?", (tid,)) conn.execute( "DELETE FROM tasks WHERE project_id LIKE ?", (PREFIX + "%",)) conn.execute( "DELETE FROM project_members WHERE project_id LIKE ?", (PREFIX + "%",)) conn.execute( "DELETE FROM projects WHERE id LIKE ?", (PREFIX + "%",)) conn.execute( "DELETE FROM users WHERE username LIKE ?", (PREFIX + "%",)) conn.commit() # 校验残留（只查测试前缀的行，不打印 key 或内容）。 for table, col in (("tasks", "project_id"), ("projects", "id"), ("users", "username")): row = conn.execute( "SELECT COUNT(*) FROM %s WHERE %s LIKE ?" % (table, col), (PREFIX + "%",)).fetchone() if row and row[0]: leftover.append("%s:%d" % (table, row[0])) finally: conn.close() except Exception as e: leftover.append("cleanup-error:%s" % e) try: auth.conn.close() except Exception: pass if leftover: print("清理未完成，残留测试行：%s" % ", ".join(leftover)) print("注意：审计日志里的测试条目是故意留下的，不算残留。") passed = sum(1 for _, ok in RESULTS if ok) total = len(RESULTS) print("多用户验证：%d/%d 通过" % (passed, total)) if leftover: return 1 return 0 if passed == total else 1 if __name__ == "__main__": sys.exit(main())