138 lines
5.7 KiB
Python
138 lines
5.7 KiB
Python
"""CMA 企业记忆分区——生产 E2E 越权验证（2026-09-03）
|
||
|
||
流程：DB插标记文章(全局/E1专属/E2专属) → 登E1/E2互查 → 断言 → 清理回查
|
||
标记: 【系统验证临时数据-验证后删除】
|
||
"""
|
||
import sys
|
||
|
||
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("'")
|
||
|
||
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', 'cma_pass_2026'),
|
||
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)
|
||
# 存量全局文章（entity_id=0 系统共享）在加列后保持 38 篇
|
||
cur.execute("SELECT COUNT(*) FROM knowledge_articles WHERE entity_id = 0")
|
||
g = cur.fetchone()[0]
|
||
print(f" 系统共享文章(entity_id=0): {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)",
|
||
(0, 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", "password": "admin123", "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)
|