crm-backend-matt/.scratch/crm-project/audit/probe2_api.py

# -*- coding: utf-8 -*-
"""票16 审计 · API 探针第2轮：权限面 / 状态机边界 / 动态 actionName / 卡重复绑定的落库形态"""
import io, json, sys, time, urllib.request, urllib.parse, urllib.error

sys.stdout = io.TextIOWrapper(sys.stdout.buffer, encoding='utf-8')
BASE = "http://localhost:8080"
AUD = r"E:\code\crm-backend-matt\.scratch\crm-project\audit"
DIRECTOR, SALES = "99021002", "99021003"
CUSTOMER, CARD = "99021001", "99021008"

def token(uid):
    j = _req("GET", "/api/auth/debug/token?userId=" + uid)
    d = j["data"]
    return d if isinstance(d, str) else d["token"]

def _req(method, path, body=None, tok=None):
    data = urllib.parse.urlencode(body).encode() if body is not None else None
    headers = {"Authorization": "Bearer " + tok} if tok else {}
    if data: headers["Content-Type"] = "application/x-www-form-urlencoded;charset=UTF-8"
    r = urllib.request.Request(BASE + path, data=data, headers=headers, method=method)
    try:
        with urllib.request.urlopen(r, timeout=20) as resp:
            return json.loads(resp.read().decode("utf-8", "replace"))
    except urllib.error.HTTPError as e:
        try: return json.loads(e.read().decode("utf-8", "replace"))
        except Exception: return {"__status": e.code}
    except Exception as e:
        return {"__error": str(e)}

def P(case, cond, note=True):
    print(("PASS " if cond else "!!   ") + case + ((" | " + str(note)[:260]) if note is not True else ""))
    return cond

TOKD, TOKS = token(DIRECTOR), token(SALES)

# ---------- A. 跨用户权限面（总监建，销售操作） ----------
nm = "A5AUD-权限-" + time.strftime("%H%M%S")
j = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)
pid = j["data"]
P("A1 总监建档成功", j.get("code") == 0, str(j)[:120])

j = _req("POST", "/api/project/filing/review", {"id": pid, "approved": "true"}, tok=TOKS)
P("A2 无关销售可报备审核（原型：仅报备专员+管理员）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("POST", "/api/project/filing/review", {"id": pid, "approved": "true"}, tok=TOKD)
P("A3 总监报备审核通过", j.get("code") == 0)

j = _req("POST", "/api/project/stage/advance", {"id": pid}, tok=TOKS)
P("A4 无关销售可推进他人项目（原型：确认通过限负责人/销售管理员）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")

j = _req("GET", "/api/project/detail?id=" + pid, tok=TOKS)
P("A5 无关销售可读他人项目详情（数据权限）", j.get("code") == 0, f"code={j.get('code')}")
j = _req("GET", "/api/project/list?scope=manage&current=1&size=50", tok=TOKS)
rows = (j.get("data") or {}).get("content") or []
P("A6 普通销售 manage 可见他人项目（DataScope 应限部门）", all(str(r.get("ownerUserId")) == SALES for r in rows), f"rows={len(rows)} leak={[r.get('ownerUserId') for r in rows][:5]}")
j = _req("GET", "/api/project/board?scope=manage", tok=TOKS)
cols = (j.get("data") or {}).get("columns") or []
leak = [c["id"] for c in (cols[1].get("cards") or []) if str(c.get("ownerUserId")) == str(DIRECTOR)] if len(cols) > 1 else []
P("A7 普通销售 manage 看板可见总监项目", not leak, f"stage1 leak={leak}")

j = _req("POST", "/api/project/close", {"id": pid, "closeType": "MANUAL"}, tok=TOKS)
P("A8 无关销售可关闭他人项目", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")

# ---------- B. 状态机边界 ----------
nm = "A5AUD-边界-" + time.strftime("%H%M%S")
j = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)
pidB = j["data"]
_req("POST", "/api/project/filing/review", {"id": pidB, "approved": "true"}, tok=TOKD)

j = _req("POST", "/api/project/stage/advance", {"id": pidB, "skipBidding": "true"}, tok=TOKD)
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
P("B1 stage1 带 skipBidding 不应连跳（唯一连跳在 stage3）", d.get("projectStage") == 2, f"stage={d.get('projectStage')} code={j.get('code')}")

_req("POST", "/api/project/stage/advance", {"id": pidB}, tok=TOKD)  # 2→3
j = _req("POST", "/api/project/stage/advance", {"id": pidB, "skipBidding": "false"}, tok=TOKD)
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
P("B2 stage3 skipBidding=false 走 3→4", d.get("projectStage") == 4, f"stage={d.get('projectStage')}")

j = _req("POST", "/api/project/stage/advance", {"id": pidB}, tok=TOKD)  # 4→5
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
P("B3 4→5 推进", d.get("projectStage") == 5, f"stage={d.get('projectStage')} code={j.get('code')}")

j = _req("POST", "/api/project/stage/advance", {"id": pidB}, tok=TOKD)  # stage5 未标记不允许推进→6
P("B4 stage5 未标记时推进被拒（不许线性 5→6）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
d = (_req("GET", "/api/project/detail?id=" + pidB, tok=TOKD).get("data") or {})
P("B5 stage5 未标记可回退（spec 一致性）", True, "")
j = _req("GET", "/api/project/stage/rollback-targets?id=" + pidB, tok=TOKD)
tg = sorted(t["stage"] for t in (j.get("data") or []))
P("B6 走满 1-4 后可选节点={1,2,3,4}", tg == [1, 2, 3, 4], str(tg))

_req("POST", "/api/project/bid-result/mark", {"id": pidB, "bidResult": "2"}, tok=TOKD)  # 输单
j = _req("POST", "/api/project/filing/review", {"id": pidB, "approved": "true"}, tok=TOKD)
P("B7 结项后报备审核被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("POST", "/api/project/follow/add", {"projectId": pidB, "followContent": "终态后跟进", "followTime": "2026-09-10 11:00:00"}, tok=TOKD)
P("B8 结项终态后跟进新增（原型已关闭全部只读）", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")

# 关闭后再关闭 / 非法 closeType
nm = "A5AUD-关闭-" + time.strftime("%H%M%S")
pidC = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)["data"]
j = _req("POST", "/api/project/close", {"id": pidC, "closeType": "NOT_A_TYPE"}, tok=TOKD)
P("B9 非法 closeType 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
_req("POST", "/api/project/close", {"id": pidC, "closeType": "MANUAL", "closeReason": "r", "closeRemark": "m"}, tok=TOKD)
j = _req("POST", "/api/project/close", {"id": pidC, "closeType": "MANUAL"}, tok=TOKD)
P("B10 已关闭不可重复关闭", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("POST", "/api/project/stage/advance", {"id": pidC}, tok=TOKD)
P("B11 已关闭不可推进", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("POST", "/api/project/bid-result/mark", {"id": pidC, "bidResult": "1"}, tok=TOKD)
P("B12 已关闭不可标记赢单", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")

# ---------- C. 参数健壮性 ----------
j = _req("GET", "/api/project/board", tok=TOKD)
P("C1 board 缺 scope 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("GET", "/api/project/board?scope=all", tok=TOKD)
P("C2 board 非法 scope 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("GET", "/api/project/detail", tok=TOKD)
P("C3 detail 缺 id 被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("GET", "/api/project/detail?id=999", tok=TOKD)
P("C4 detail 不存在 id → 68002", j.get("code") == 68002, str(j.get("message")))
j = _req("GET", "/api/project/list?scope=manage&current=0&size=0", tok=TOKD)
P("C5 list 分页参数退化不炸", j.get("code") == 0, f"code={j.get('code')} msg={j.get('message')}")
j = _req("POST", "/api/project/follow/add", {"projectId": pidC, "followContent": "x", "followTime": "垃圾时间"}, tok=TOKD)
P("C6 跟进非法时间格式被拒", j.get("code") != 0, f"code={j.get('code')} msg={j.get('message')}")

# ---------- D. 动态 actionName 全集 + 跟进是否留痕 ----------
j = _req("GET", "/api/project/timeline/list?projectId=" + pidB + "&current=1&size=50", tok=TOKD)
ev = [(t.get("action"), t.get("actionName"), t.get("fromStage"), t.get("toStage"), t.get("opDesc")) for t in (j.get("data") or {}).get("content") or []]
print("TIMELINE pidB:")
for e in ev: print("   ", e)
acts = [e[0] for e in ev]
P("D1 输单留痕 BID_LOSE", "BID_LOSE" in acts, str(acts))
j = _req("GET", "/api/project/timeline/list?projectId=" + pidC + "&current=1&size=50", tok=TOKD)
evC = [(t.get("action"), t.get("actionName")) for t in (j.get("data") or {}).get("content") or []]
print("TIMELINE pidC:", evC)

# 跟进新增是否写动态（用一个新项目验证）
nm = "A5AUD-动态-" + time.strftime("%H%M%S")
pidD = _req("POST", "/api/project/create", {"projectName": nm, "customerId": CUSTOMER, "schemeCardId": CARD}, tok=TOKD)["data"]
_req("POST", "/api/project/follow/add", {"projectId": pidD, "followContent": "留痕验证", "followTime": "2026-09-10 12:00:00"}, tok=TOKD)
j = _req("GET", "/api/project/timeline/list?projectId=" + pidD + "&current=1&size=50", tok=TOKD)
actsD = [t.get("action") for t in (j.get("data") or {}).get("content") or []]
P("D2 跟进新增写项目动态", any("FOLLOW" in str(a) for a in actsD), str(actsD))

# ---------- E. 卡重复绑定后的卡归属形态（读 DB） ----------
try:
    import pymysql
    conn = pymysql.connect(host="8.129.84.155", port=3306, user="root", password="Itc@123456", database="itc_crm", charset="utf8mb4")
    cur = conn.cursor()
    cur.execute("SELECT id, owner_type, owner_id, customer_id, delete_key FROM opportunity_scheme_card WHERE id=%s", (CARD,))
    print("CARD ROW:", cur.fetchall())
    cur.execute("SELECT table_schema FROM information_schema.schemata WHERE table_schema LIKE '%%crm%%' LIMIT 5")
    print("SCHEMAS:", cur.fetchall())
    conn.close()
except Exception as e:
    print("DB CHECK SKIP:", e)