foms/scripts/init_rbac.py

#!/usr/bin/env python3
"""
FOMS RBAC 初始化 — 创建机构类型、机构、角色、用户

用法: cd /d/ymq/repos/foms && source py3/bin/activate && python scripts/init_rbac.py
"""
import os, sys, asyncio

# Add project root
root = os.path.dirname(os.path.dirname(os.path.abspath(__file__)))
sys.path.insert(0, root)

from appPublic.uniqueID import getID
from appPublic.timeUtils import curDateString
from sqlor.dbpools import DBPools
from ahserver.serverenv import ServerEnv

DBNAME = 'foms'

ORG_ID = 'foms_org_001'
ROLES = {
    'foms_superadmin': '超级管理员',
    'foms_operator':   '运维操作员',
    'foms_viewer':     '只读观察者',
}
USERS = {
    'admin':    {'name': '系统管理员', 'password': 'Admin@123',   'role': 'foms_superadmin'},
    'operator': {'name': '运维操作员', 'password': 'Ops@123456', 'role': 'foms_operator'},
}


async def main():
    env = ServerEnv()
    dbname = env.get_module_dbname('rbac') if hasattr(env, 'get_module_dbname') else DBNAME
    db = DBPools()

    async with db.sqlorContext(dbname) as sor:
        # 1. Create organization
        existing = await sor.R('organization', {'id': ORG_ID})
        if existing:
            print(f"[SKIP] organization {ORG_ID} already exists")
        else:
            await sor.C('organization', {
                'id': ORG_ID,
                'orgname': 'FOMS运维中心',
                'orgabbr': 'FOMS',
                'org_type': 'foms_org',
            })
            print(f"[OK] organization created: FOMS运维中心")

        # 2. Create roles
        for role_id, role_name in ROLES.items():
            existing = await sor.R('role', {'id': role_id})
            if existing:
                print(f"[SKIP] role {role_id} exists")
                continue
            await sor.C('role', {
                'id': role_id,
                'orgtypeid': 'foms_org',
                'name': role_name,
            })
            print(f"[OK] role created: {role_id} ({role_name})")

        # 3. Create users
        for username, info in USERS.items():
            uid = f'user_{username}'
            existing = await sor.R('users', {'username': username})
            if existing:
                print(f"[SKIP] user {username} exists")
            else:
                encoded_pw = env.password_encode(info['password'])
                await sor.C('users', {
                    'id': uid,
                    'username': username,
                    'name': info['name'],
                    'nick_name': info['name'],
                    'password': encoded_pw,
                    'orgid': ORG_ID,
                    'user_status': '0',
                    'created_at': curDateString(),
                })
                print(f"[OK] user created: {username}")

            # 4. Assign role to user
            ur_id = f'ur_{username}_{info["role"]}'
            existing_ur = await sor.R('userrole', {'userid': uid, 'roleid': info['role']})
            if existing_ur:
                print(f"[SKIP] userrole {username}→{info['role']} exists")
                continue
            await sor.C('userrole', {
                'id': ur_id,
                'userid': uid,
                'roleid': info['role'],
            })
            print(f"[OK] userrole: {username} → {info['role']}")

        # 5. Assign permissions to roles
        # Get all FOMS-specific permissions (paths not from rbac/appbase)
        all_perms = await sor.sqlExe(
            "SELECT id, path FROM permission WHERE path IN ("
            "SELECT path FROM permission WHERE path LIKE '/api/%' OR path LIKE '/%_list%'"
            ") ORDER BY path", {}
        )
        # Most paths start with / for FOMS root-mapped module
        foms_perms = [p for p in all_perms if hasattr(p, 'path')]
        if not foms_perms:
            print("[WARN] No FOMS permissions found — run scripts/load_path.py first")
        else:
            # superadmin: all FOMS paths
            # operator: all paths (can trigger switchover)
            # viewer: read-only (no create/update/delete)
            for perm in foms_perms:
                pid = getattr(perm, 'id', '')
                path = getattr(perm, 'path', '')
                is_write = any(x in path for x in ['create', 'update', 'delete', 'trigger'])
                is_agent = 'agent_' in path

                # superadmin gets everything
                await _add_roleperm(sor, 'foms_superadmin', pid, path)

                # operator gets everything except agent paths (internal)
                if not is_agent:
                    await _add_roleperm(sor, 'foms_operator', pid, path)

                # viewer gets read-only
                if not is_write and not is_agent:
                    await _add_roleperm(sor, 'foms_viewer', pid, path)

    print(f"\n=== FOMS RBAC 初始化完成 ===")
    print(f"  机构: FOMS运维中心 ({ORG_ID})")
    print(f"  角色: {', '.join(f'{k}({v})' for k, v in ROLES.items())}")
    print(f"  用户:")
    for u, i in USERS.items():
        print(f"    {u} / {i['password']} (角色: {i['role']})")
    print(f"\n  角色权限分配:")
    print(f"    superadmin → 全部路径")
    print(f"    operator   → 全部路径 (不含agent内部端点)")
    print(f"    viewer     → 只读路径 (查看dashboard/列表)")


async def _add_roleperm(sor, role_id, perm_id, path):
    """Add rolepermission if not exists"""
    rp_id = f'rp_{role_id}_{perm_id}'[:32]
    existing = await sor.R('rolepermission', {'roleid': role_id, 'permid': perm_id})
    if existing:
        return
    await sor.C('rolepermission', {
        'id': rp_id,
        'roleid': role_id,
        'permid': perm_id,
    })


if __name__ == '__main__':
    asyncio.run(main())