"""CMA 企业记忆分区——生产 E2E 越权验证（2026-09-03） 流程：DB插标记文章(全局/E1专属/E2专属) → 登E1/E2互查 → 断言 → 清理回查 标记: 【系统验证临时数据-验证后删除】 """ import sys import os # noqa: E402 (凭据加载器需要) # ---- 管理员凭据同源加载（2026-09-14 cma-defaultcred-code-fix）---- # 生产凭据只存 600 密钥文件；禁止在脚本/文档/日志里硬编码口令。 # 顺序：$CMA_ADMIN_ENV → wecom-ops 主副本（ops 轮换落点，权威） → 本 profile 副本。 def _admin_creds(): """返回 (user, password, 来源文件)；未找到 → 显式报错退出（不静默退回默认口令）""" for p in (os.environ.get("CMA_ADMIN_ENV"), "/root/.hermes/profiles/wecom-ops/.secrets/cma-admin.env", "/root/.hermes/profiles/wecom-fullstack/.secrets/cma-admin.env"): if p and os.path.isfile(p): kv = {} for line in open(p, encoding="utf-8"): line = line.strip() if "=" in line and not line.startswith("#"): k, v = line.split("=", 1) kv[k.strip()] = v.strip().strip('"').strip("'") if kv.get("CMA_ADMIN_USER") and kv.get("CMA_ADMIN_PASS"): return kv["CMA_ADMIN_USER"], kv["CMA_ADMIN_PASS"], p raise SystemExit("[错误] 未找到 CMA 管理员凭据：请设 $CMA_ADMIN_ENV，或准备 600 权限密钥文件 " ".secrets/cma-admin.env（键名 CMA_ADMIN_USER / CMA_ADMIN_PASS）") ADMIN_USER, ADMIN_PASS, CRED_SRC = _admin_creds() print(f"[creds] 来源={CRED_SRC} 账号={ADMIN_USER}（口令不回显）") env = {} with open('/root/cma-management/backend/.env') as f: for line in f: line = line.strip() if line and not line.startswith('#') and '=' in line: k, v = line.split('=', 1) env[k.strip()] = v.strip().strip('"').strip("'") if not env.get('CMA_DB_PASS'): raise SystemExit('ERROR: 缺少 CMA_DB_PASS（2026-09-12 凭据治理已移除明文兜底）') import pymysql conn = pymysql.connect( host=env.get('CMA_DB_HOST', '127.0.0.1'), port=int(env.get('CMA_DB_PORT', '3306')), user=env.get('CMA_DB_USER', 'cma_user'), password=env.get('CMA_DB_PASS'), database=env.get('CMA_DB_NAME', 'cma'), charset='utf8mb4', ) cur = conn.cursor() MARK = "【系统验证临时数据-验证后删除】" ok = True def check(name, cond, extra=""): global ok print((" ✅" if cond else " ❌"), name, extra) if not cond: ok = False # 0. 三表列+索引已存在（DB 回查） for t in ['knowledge_events', 'knowledge_summaries', 'knowledge_articles']: cur.execute(f"SHOW COLUMNS FROM {t} LIKE 'entity_id'") has_col = cur.fetchone() is not None cur.execute(f"SHOW INDEX FROM {t} WHERE Column_name='entity_id'") has_idx = cur.fetchone() is not None check(f"{t}.entity_id 列+索引", has_col and has_idx) # 存量全局文章（NULL）在加列后保持 38 篇 cur.execute("SELECT COUNT(*) FROM knowledge_articles WHERE entity_id IS NULL") g = cur.fetchone()[0] print(f" 存量全局文章(NULL): {g} 篇") check("存量全局教学文章保持全局共享", g == 38, f"(期望38)") # 1. 插临时验证文章（标题带标记，source 专属 entity） cur.execute("DELETE FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",)) conn.commit() cur.execute( "INSERT INTO knowledge_articles (entity_id, title, summary, content, category) VALUES (%s,%s,%s,%s,%s)", (1, f"{MARK}E1专属越权测试", "s", "c", "term")) cur.execute( "INSERT INTO knowledge_articles (entity_id, title, summary, content, category) VALUES (%s,%s,%s,%s,%s)", (2, f"{MARK}E2专属越权测试", "s", "c", "term")) cur.execute( "INSERT INTO knowledge_articles (entity_id, title, summary, content, category) VALUES (%s,%s,%s,%s,%s)", (None, f"{MARK}全局共享测试", "s", "c", "term")) conn.commit() cur.execute("SELECT id, entity_id, title FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",)) rows = cur.fetchall() by_eid = {r[1]: (r[0], r[2]) for r in rows} ids = {r[1]: r[0] for r in rows} print(f" 插入临时文章: {rows}") # 2. 登录 E1/E2 调 API import urllib.request, json as _json def api(method, path, token=None, body=None): url = f"http://127.0.0.1:8010{path}" req = urllib.request.Request(url, method=method) req.add_header("Content-Type", "application/json") if token: req.add_header("Authorization", f"Bearer {token}") if body is not None: req.data = _json.dumps(body).encode() try: with urllib.request.urlopen(req, timeout=30) as resp: return resp.status, resp.read().decode() except urllib.error.HTTPError as e: return e.code, e.read().decode() def login(eid): st, body = api("POST", "/api/cma/auth/login", body={"username": ADMIN_USER, "password": ADMIN_PASS, "entity_id": eid}) if st != 200: print(f" ⚠️ 登录 entity{eid} 失败 {st}: {body[:200]}") return None return _json.loads(body)["token"] t1 = login(1) t2 = login(2) if not t1 or not t2: print("登录失败，跳过 API 断言") sys.exit(1) # E1 视角 st, body = api("GET", "/api/cma/knowledge-articles", token=t1) d1 = _json.loads(body) titles1 = {i["title"] for i in d1.get("data", [])} check("E1 见全局共享文章", f"{MARK}全局共享测试" in titles1) check("E1 见自己专属文章", f"{MARK}E1专属越权测试" in titles1) check("E1 不见 E2 专属文章", f"{MARK}E2专属越权测试" not in titles1, "(核心越权1)") # E2 视角 st, body = api("GET", "/api/cma/knowledge-articles", token=t2) titles2 = {i["title"] for i in _json.loads(body).get("data", [])} check("E2 见全局共享文章", f"{MARK}全局共享测试" in titles2) check("E2 见自己专属文章", f"{MARK}E2专属越权测试" in titles2) check("E2 不见 E1 专属文章", f"{MARK}E1专属越权测试" not in titles2, "(核心越权2)") # E1 直取 E2 专属文章详情 → 404 st, body = api("GET", f"/api/cma/knowledge-articles/{ids[2]}", token=t1) check("E1 直取 E2 专属文章 → 404", st == 404, f"(st={st})") # E2 直取 E1 专属文章详情 → 404 st, body = api("GET", f"/api/cma/knowledge-articles/{ids[1]}", token=t2) check("E2 直取 E1 专属文章 → 404", st == 404, f"(st={st})") # 摘要接口（生产空表）两账套 200 不 500 for eid, tok in ((1, t1), (2, t2)): st, body = api("GET", "/api/cma/knowledge/summaries", token=tok) d = _json.loads(body) check(f"E{eid} /knowledge/summaries 200 + 空列表", st == 200 and d.get("total", -1) == 0, f"(st={st} total={d.get('total')})") st, body = api("GET", "/api/cma/knowledge/events", token=tok) check(f"E{eid} /knowledge/events 200", st == 200, f"(st={st})") # 3. 清理 + 回查 cur.execute("DELETE FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",)) conn.commit() cur.execute("SELECT COUNT(*) FROM knowledge_articles WHERE title LIKE %s", (f"%{MARK}%",)) left = cur.fetchone()[0] check("临时数据清理回查 = 0", left == 0, f"(剩{left})") cur.execute("SELECT COUNT(*) FROM knowledge_articles") print(f" knowledge_articles 最终行数: {cur.fetchone()[0]}") conn.close() print("\n" + ("全部通过 ✅" if ok else "存在失败 ❌")) sys.exit(0 if ok else 1)