68 lines
2.4 KiB
Python
68 lines
2.4 KiB
Python
"""只读核验场外 Worker 技术账号和最小权限。"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import asyncio
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
from sqlalchemy import text
|
|
|
|
ROOT = Path(__file__).resolve().parents[1]
|
|
sys.path.insert(0, str(ROOT))
|
|
|
|
from app.core.contracts import RequestContext # noqa: E402
|
|
from app.infrastructure.db import SessionFactory, engine # noqa: E402
|
|
from app.service.identity_service import IdentityService # noqa: E402
|
|
|
|
WORKER_USER_NO = "OFFSITE-WORKER"
|
|
WORKER_USERNAME = "offsite_worker"
|
|
REQUIRED_ROLE = "operator"
|
|
REQUIRED_PERMISSION = "offsite:write"
|
|
|
|
|
|
async def main() -> None:
|
|
async with SessionFactory() as session:
|
|
row = (
|
|
await session.execute(
|
|
text(
|
|
"""
|
|
SELECT id, user_no, username, user_type, employee_role, status
|
|
FROM sys_user
|
|
WHERE user_no=:user_no AND username=:username
|
|
"""
|
|
),
|
|
{"user_no": WORKER_USER_NO, "username": WORKER_USERNAME},
|
|
)
|
|
).mappings().first()
|
|
if row is None:
|
|
raise SystemExit("未找到场外 Worker 技术账号，请先执行 Alembic 迁移")
|
|
if row["user_type"] != "员工" or row["employee_role"] != REQUIRED_ROLE:
|
|
raise SystemExit("场外 Worker 技术账号身份属性不符合最小权限方案")
|
|
if row["status"] != "正常":
|
|
raise SystemExit("场外 Worker 技术账号未启用")
|
|
user_id = str(row["id"])
|
|
|
|
context = await IdentityService().resolve(
|
|
RequestContext(user_id=user_id, trace_id="verify-offsite-worker-identity")
|
|
)
|
|
roles = set(context.roles)
|
|
permissions = set(context.permissions)
|
|
if REQUIRED_ROLE not in roles:
|
|
raise SystemExit("场外 Worker 技术账号缺少 operator 角色")
|
|
if REQUIRED_PERMISSION not in permissions:
|
|
raise SystemExit("场外 Worker 技术账号缺少 offsite:write 权限")
|
|
extra_permissions = permissions - {REQUIRED_PERMISSION}
|
|
if extra_permissions:
|
|
raise SystemExit(f"场外 Worker 技术账号存在额外权限：{sorted(extra_permissions)}")
|
|
|
|
print("场外 Worker 技术账号核验通过")
|
|
print(f"OFFSITE_WORKER_USER_ID={user_id}")
|
|
print(f"roles={sorted(roles)}")
|
|
print(f"permissions={sorted(permissions)}")
|
|
await engine.dispose()
|
|
|
|
|
|
if __name__ == "__main__":
|
|
asyncio.run(main())
|