Hermes CI Fix
40d522cd34
- Login.vue: no prefilled default creds; no anonymous entity enumeration on mount - deploy.sh: readiness polling (0.5s/30s cap) replaces fixed sleep 3; login smoke reads same-source 600 secret file, missing file = explicit abort - 19 verify/check/gen_report scripts: shared _admin_creds() loader (CMA_ADMIN_ENV -> ops .secrets -> profile copy); no hardcoded creds - tests: single TEST_USER_PASSWORD fixture constant; docs: pointer to 600 secret file; removed 4 legacy .bak + pycache residue
122 lines
4.8 KiB
Python
122 lines
4.8 KiB
Python
#!/usr/bin/env python3
|
||
# -*- coding: utf-8 -*-
|
||
"""my-dashboard 隔离补漏验证: 登录 entity1 -> 工作台 -> 回查DB归属全为 entity1"""
|
||
import json
|
||
import urllib.request
|
||
import pymysql
|
||
|
||
# 2026-09-12 库访问治理（ops）：凭据从 backend/.env 读取，无明文兜底
|
||
import os
|
||
# ---- 管理员凭据同源加载（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}（口令不回显）")
|
||
|
||
try:
|
||
from dotenv import load_dotenv as _ld_gov
|
||
_ld_gov('/root/cma-management/backend/.env')
|
||
except Exception:
|
||
pass
|
||
if not os.getenv('CMA_DB_PASS'):
|
||
raise SystemExit('ERROR: 缺少 CMA_DB_PASS（2026-09-12 凭据治理已移除明文兜底）')
|
||
|
||
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_USER, "password": ADMIN_PASS, "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)} 条")
|
||
|
||
# 账号与口令同源（backend/.env）——硬编码 cma_user + 应用口令会账号/口令错配
|
||
conn = pymysql.connect(host="127.0.0.1", user=os.getenv("CMA_DB_USER", "cma_user"),
|
||
password=os.environ.get("CMA_DB_PASS", ""),
|
||
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 工作台数据无跨账套泄漏")
|