rlz/verify_roles_v2.py

#!/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()