86 lines
2.9 KiB
Python
86 lines
2.9 KiB
Python
#!/usr/bin/env python3
|
|
# -*- coding: utf-8 -*-
|
|
"""my-dashboard 隔离补漏验证: 登录 entity1 -> 工作台 -> 回查DB归属全为 entity1"""
|
|
import json
|
|
import urllib.request
|
|
import pymysql
|
|
|
|
BASE = "http://127.0.0.1:8010"
|
|
|
|
|
|
def post(path, data):
|
|
req = urllib.request.Request(
|
|
BASE + path,
|
|
data=json.dumps(data).encode(),
|
|
headers={"Content-Type": "application/json"},
|
|
)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def get(path, token):
|
|
req = urllib.request.Request(
|
|
BASE + path, headers={"Authorization": "Bearer " + token}
|
|
)
|
|
with urllib.request.urlopen(req, timeout=30) as resp:
|
|
return json.loads(resp.read().decode())
|
|
|
|
|
|
def main():
|
|
# 1. 登录 entity1
|
|
r = post("/api/cma/auth/login", {"username": "admin", "password": "admin123", "entity_id": 1})
|
|
token = r.get("access_token") or r.get("token")
|
|
assert token, f"登录失败: {r}"
|
|
print("1. 登录 entity1 OK")
|
|
|
|
# 2. 工作台
|
|
d = get("/api/cma/dashboard/my-dashboard", token)
|
|
plans = d.get("action_plans", [])
|
|
reminders = d.get("reminders", [])
|
|
print(f"2. my-dashboard OK: action_plans={len(plans)} 条, reminders={len(reminders)} 条")
|
|
|
|
conn = pymysql.connect(host="127.0.0.1", user="cma_user", password="cma_pass_2026",
|
|
database="cma", charset="utf8mb4")
|
|
cur = conn.cursor()
|
|
|
|
# 3. action_plans 全部归属 entity1
|
|
ids = [p["id"] for p in plans]
|
|
if ids:
|
|
fmt = ",".join(["%s"] * len(ids))
|
|
cur.execute(f"SELECT id, entity_id FROM action_plans WHERE id IN ({fmt})", ids)
|
|
rows = cur.fetchall()
|
|
bad = [r for r in rows if r[1] != 1]
|
|
print(f"3. action_plans 回查DB归属: 非entity1 = {bad if bad else '无'}")
|
|
else:
|
|
print("3. action_plans 返回 0 条 (跳过)")
|
|
|
|
# 4. reminders 中 action_plan 的 related_id 归属
|
|
plan_rids = [r["related_id"] for r in reminders if r.get("related_type") == "action_plan"]
|
|
isolation_bad = False
|
|
if plan_rids:
|
|
fmt = ",".join(["%s"] * len(plan_rids))
|
|
cur.execute(f"SELECT id, entity_id FROM action_plans WHERE id IN ({fmt})", plan_rids)
|
|
rows = cur.fetchall()
|
|
bad = [r for r in rows if r[1] != 1]
|
|
isolation_bad = bool(bad)
|
|
print(f"4. reminders.action_plan related_id 共 {len(plan_rids)} 个, 非entity1 = {bad if bad else '无'}")
|
|
else:
|
|
print("4. reminders 无 action_plan 类型 (跳过)")
|
|
|
|
# 5. reminders 类型/严重度分布 (前端标签/排序数据源)
|
|
dist = {}
|
|
for r in reminders:
|
|
key = (r.get("related_type"), r.get("type"), r.get("severity"))
|
|
dist[key] = dist.get(key, 0) + 1
|
|
print("5. reminders 类型分布:", dist)
|
|
|
|
cur.close()
|
|
conn.close()
|
|
|
|
assert not isolation_bad, "发现跨账套数据泄漏"
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|
|
print("PASS: entity1 工作台数据无跨账套泄漏")
|