feat: 新增多角色系统（派单/分润/推广/区域）
- 后端：RlzDispatch/Profit/Promotion/Region 控制器、服务、Mapper、领域模型 - 权限：MethodSecurityConfig + PermissionInterceptor - 管理后台：派单/分润/推广/区域 页面与 API - Android 陪护端：派单/运营/退款 Fragment 及布局 - 小程序：推广员子包、派单/运营/退款/检索子包、自定义 tabBar - 文档：多角色系统方案、角色权限矩阵、用户操作手册等 Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
renjianbo
174
verify_roles_v2.py
Normal file
174
verify_roles_v2.py
Normal file
@@ -0,0 +1,174 @@
|
||||
#!/usr/bin/env python3
|
||||
# -*- coding: utf-8 -*-
|
||||
"""
|
||||
Role permission verification script.
|
||||
Tests each role's access to critical endpoints via the PermissionInterceptor.
|
||||
"""
|
||||
|
||||
import requests
|
||||
import json
|
||||
import sys
|
||||
|
||||
BASE = "http://101.43.95.130:8039"
|
||||
|
||||
# Test users: login uses user_name, not phone
|
||||
# Role assignments from DB:
|
||||
# admin(1): admin role -> *:*:* (from code)
|
||||
# dispatcher01(112): dispatcher -> system:dispatch:assign, system:dispatch:list, system:profit:list
|
||||
# service01(113): service -> system:order:list, system:profit:list
|
||||
# ops01(114): operations -> system:dispatch:list, system:order:list, system:platformConfig:list, system:profit:list, system:region:list
|
||||
# promoter_a(115): promoter -> system:profit:list, system:promotion:list, system:promotion:query
|
||||
# caregiver01(118): NO ROLE -> no permissions
|
||||
# patient01(119): NO ROLE -> no permissions
|
||||
TEST_USERS = [
|
||||
{"name": "admin", "username": "admin", "password": "admin123", "role": "admin"},
|
||||
{"name": "dispatcher01", "username": "dispatcher01", "password": "123456", "role": "dispatcher"},
|
||||
{"name": "service01", "username": "service01", "password": "123456", "role": "service"},
|
||||
{"name": "ops01", "username": "ops01", "password": "123456", "role": "operations"},
|
||||
{"name": "promoter_a", "username": "promoter_a", "password": "123456", "role": "promoter"},
|
||||
{"name": "caregiver01", "username": "caregiver01", "password": "123456", "role": "none"},
|
||||
{"name": "patient01", "username": "patient01", "password": "123456", "role": "none"},
|
||||
]
|
||||
|
||||
# Endpoints to test
|
||||
# (method, url, desc, should_work_if_role_has_this_permission)
|
||||
# For "none" role users, most should be blocked except unprotected endpoints
|
||||
ENDPOINTS = [
|
||||
# User management - requires system:user:list
|
||||
("GET", "/system/user/list", "userList", "system:user:list"),
|
||||
# Promotion - requires system:promotion:query
|
||||
("GET", "/system/promotion/myStats", "promoStats", "system:promotion:query"),
|
||||
("GET", "/system/promotion/team", "promoTeam", "system:promotion:query"),
|
||||
("GET", "/system/promotion/myCommission", "promoCommission", "system:promotion:query"),
|
||||
# Dispatch - requires system:dispatch:list
|
||||
("GET", "/system/dispatch/pending", "dispatchPending", "system:dispatch:list"),
|
||||
("GET", "/system/dispatch/caregivers", "dispatchCaregivers", "system:dispatch:list"),
|
||||
# Region - requires system:region:list
|
||||
("GET", "/system/region/list", "regionList", "system:region:list"),
|
||||
# Order myBalance - may not have @PreAuthorize (checking)
|
||||
("GET", "/system/profit/myBalance", "myBalance", None),
|
||||
# Order list - requires system:order:list
|
||||
("GET", "/system/order/list", "orderList", "system:order:list"),
|
||||
]
|
||||
|
||||
|
||||
def login(username, password):
|
||||
"""Login via appLogin, return (token, userId)"""
|
||||
try:
|
||||
r = requests.post(f"{BASE}/appLogin", data={
|
||||
"username": username,
|
||||
"password": password
|
||||
}, timeout=10)
|
||||
data = r.json()
|
||||
if data.get("code") == 200:
|
||||
return data["data"]["token"], data["data"]["userId"]
|
||||
return None, data.get("msg", "unknown error")
|
||||
except Exception as e:
|
||||
return None, str(e)
|
||||
|
||||
|
||||
def get_user_info(token):
|
||||
"""Get user permissions from /getInfo"""
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
try:
|
||||
r = requests.get(f"{BASE}/getInfo", headers=headers, timeout=10)
|
||||
data = r.json()
|
||||
if data.get("code") == 200:
|
||||
return data.get("permissions", [])
|
||||
return []
|
||||
except:
|
||||
return []
|
||||
|
||||
|
||||
def test_endpoint(method, url, token):
|
||||
"""Test an endpoint and return (http_code, response_code, msg)"""
|
||||
headers = {"Authorization": f"Bearer {token}"}
|
||||
try:
|
||||
if method == "GET":
|
||||
r = requests.get(f"{BASE}{url}", headers=headers, timeout=10)
|
||||
else:
|
||||
r = requests.post(f"{BASE}{url}", headers=headers, timeout=10)
|
||||
data = r.json()
|
||||
return r.status_code, data.get("code"), data.get("msg", "")
|
||||
except Exception as e:
|
||||
return 0, None, str(e)
|
||||
|
||||
|
||||
def main():
|
||||
total_ok = 0
|
||||
total_bad = 0
|
||||
issues = []
|
||||
|
||||
for user in TEST_USERS:
|
||||
print(f"\n{'='*60}")
|
||||
print(f"User: {user['name']} | Role: {user['role']}")
|
||||
print(f"{'='*60}")
|
||||
|
||||
token, info = login(user["username"], user["password"])
|
||||
if not token:
|
||||
print(f" !! Login FAILED: {info}")
|
||||
continue
|
||||
|
||||
# Get actual permissions
|
||||
perms = get_user_info(token)
|
||||
perm_set = set(perms)
|
||||
print(f" Login OK (userId={info}), permissions={list(perm_set)[:5]}...")
|
||||
print()
|
||||
|
||||
for method, url, desc, required_perm in ENDPOINTS:
|
||||
http_code, resp_code, msg = test_endpoint(method, url, token)
|
||||
|
||||
# Determine access
|
||||
if resp_code == 200:
|
||||
access = "ALLOW"
|
||||
elif resp_code == 401:
|
||||
access = "DENY_401"
|
||||
elif resp_code == 403:
|
||||
access = "DENY_403"
|
||||
elif resp_code == 500 and "denied" in str(msg).lower():
|
||||
access = "DENY_500"
|
||||
elif http_code == 200 and resp_code != 200:
|
||||
access = f"BIZ_ERR({resp_code})"
|
||||
else:
|
||||
access = f"HTTP_{http_code}"
|
||||
|
||||
# Determine if this is correct
|
||||
has_perm = required_perm is None or required_perm in perm_set
|
||||
# admin always has *:*:*
|
||||
if user["role"] == "admin":
|
||||
has_perm = True
|
||||
|
||||
if access == "ALLOW":
|
||||
if has_perm:
|
||||
verdict = "OK"
|
||||
total_ok += 1
|
||||
else:
|
||||
verdict = "LEAK"
|
||||
total_bad += 1
|
||||
issues.append(f"{user['name']} can access {desc} ({url}) but lacks {required_perm}")
|
||||
elif access.startswith("DENY"):
|
||||
if has_perm:
|
||||
verdict = "BLOCK_ERR"
|
||||
total_bad += 1
|
||||
issues.append(f"{user['name']} DENIED {desc} ({url}) but has {required_perm}")
|
||||
else:
|
||||
verdict = "OK_BLOCK"
|
||||
total_ok += 1
|
||||
else:
|
||||
verdict = f"?{access}"
|
||||
total_bad += 1
|
||||
issues.append(f"{user['name']} UNEXPECTED {access} on {desc} ({url})")
|
||||
|
||||
print(f" [{verdict:12s}] {desc:20s} ({url}) | access={access} | need={required_perm} | has={has_perm}")
|
||||
|
||||
print(f"\n{'='*60}")
|
||||
print(f"SUMMARY: {total_ok} OK, {total_bad} PROBLEMS")
|
||||
print(f"{'='*60}")
|
||||
if issues:
|
||||
print("\nISSUES FOUND:")
|
||||
for i in issues:
|
||||
print(f" * {i}")
|
||||
|
||||
|
||||
if __name__ == "__main__":
|
||||
main()
|
||||
Reference in New Issue
Block a user