Hermes CI Fix
b31f9b80c4
- auth_middleware: create_token存JSON{user_id,entity_id}，旧int格式token强制下线
- models: 新增UserEntity授权表(user_id↔entity_id多对多,唯一约束)
- database: init_db自动建表+存量用户×active企业默认授权(平滑迁移)
- deps: get_entity_id解析链倒置 token优先 → Bot白名单(query/header校验entity active) → 默认1
- auth: login加entity_id+授权校验; 新增switch-entity/my-entities/登录页entities接口; register自动授权
252 lines
8.6 KiB
Python
252 lines
8.6 KiB
Python
"""
|
||
角色权限中间件 — 管理会计OS
|
||
4角色: ceo(CEO/总览), finance(财务), business(业务), it(IT/运维)
|
||
权限配置支持从数据库动态加载
|
||
"""
|
||
|
||
from fastapi import Request, HTTPException, Depends
|
||
from fastapi.security import HTTPBearer, HTTPAuthorizationCredentials
|
||
from sqlalchemy.orm import Session
|
||
from app.database import get_db
|
||
from app.models import User, RolePermission
|
||
import secrets
|
||
import json
|
||
|
||
# 角色定义（固定）
|
||
ROLES = {
|
||
"ceo": {"name": "CEO", "priority": 1},
|
||
"finance": {"name": "财务", "priority": 2},
|
||
"business": {"name": "业务", "priority": 3},
|
||
"it": {"name": "IT运维", "priority": 4},
|
||
}
|
||
|
||
# 默认权限（数据库没有时的 fallback）
|
||
DEFAULT_ROUTE_PERMISSIONS = {
|
||
"ceo": ["dashboard", "kpis", "kpi_detail", "maps", "alerts", "ai_analysis", "data_source", "user_manage", "system_config"],
|
||
"finance": ["dashboard", "kpis", "kpi_detail", "maps", "alerts", "ai_analysis", "data_source"],
|
||
"business": ["dashboard", "kpis", "kpi_detail", "alerts"],
|
||
"it": ["dashboard", "kpis", "kpi_detail", "alerts", "data_source", "user_manage", "system_config"],
|
||
}
|
||
|
||
DEFAULT_KPI_VISIBILITY = {
|
||
"ceo": ["*"], # CEO看全部维度
|
||
"finance": ["finance_*"], # 财务只看财务
|
||
"business": ["customer_*", "process_*", "learning_*"], # 业务看客户/流程/学习
|
||
"it": ["*"], # IT看全部（运维）
|
||
}
|
||
|
||
DEFAULT_ACTION_PERMISSIONS = {
|
||
"ceo": ["read", "approve"],
|
||
"finance": ["read", "write", "import", "export"],
|
||
"business": ["read", "write"],
|
||
"it": ["read", "write", "delete", "admin"],
|
||
}
|
||
|
||
import logging
|
||
logger = logging.getLogger("cma.auth")
|
||
|
||
# Redis token 存储（跨 worker 共享）
|
||
try:
|
||
import redis as redis_lib
|
||
_redis = redis_lib.Redis(
|
||
host="127.0.0.1", port=6379, db=1,
|
||
decode_responses=True, socket_connect_timeout=2, socket_timeout=3
|
||
)
|
||
_redis.ping()
|
||
_redis_available = True
|
||
except Exception:
|
||
_redis = None
|
||
_redis_available = False
|
||
logger.warning("Redis不可用，token存储降级到内存（不支持多worker）")
|
||
|
||
# 内存 fallback
|
||
_token_store: dict[str, dict] = {}
|
||
|
||
TOKEN_PREFIX = "cma:token:"
|
||
TOKEN_TTL = 86400 # 24小时
|
||
|
||
# 缓存权限配置（每5分钟刷新）
|
||
_permissions_cache = {"route": None, "action": None, "ts": 0}
|
||
_PERM_CACHE_TTL = 300
|
||
|
||
|
||
def _load_permissions(db: Session = None):
|
||
"""从数据库加载权限配置"""
|
||
import time
|
||
now = time.time()
|
||
if db is None:
|
||
if now - _permissions_cache["ts"] < _PERM_CACHE_TTL:
|
||
return _permissions_cache["route"] or DEFAULT_ROUTE_PERMISSIONS, _permissions_cache["action"] or DEFAULT_ACTION_PERMISSIONS
|
||
return DEFAULT_ROUTE_PERMISSIONS, DEFAULT_ACTION_PERMISSIONS
|
||
|
||
try:
|
||
route_perm = db.query(RolePermission).filter(RolePermission.key == "route_permissions").first()
|
||
action_perm = db.query(RolePermission).filter(RolePermission.key == "action_permissions").first()
|
||
|
||
routes = route_perm.value if route_perm else DEFAULT_ROUTE_PERMISSIONS
|
||
actions = action_perm.value if action_perm else DEFAULT_ACTION_PERMISSIONS
|
||
|
||
_permissions_cache["route"] = routes
|
||
_permissions_cache["action"] = actions
|
||
_permissions_cache["ts"] = now
|
||
|
||
return routes, actions
|
||
except Exception:
|
||
return DEFAULT_ROUTE_PERMISSIONS, DEFAULT_ACTION_PERMISSIONS
|
||
|
||
|
||
def create_token(user_id: int, entity_id: int = None) -> str:
|
||
"""签发token：Redis存储 JSON {user_id, entity_id}（账套模式）
|
||
兼容旧调用 create_token(user_id) → entity_id=None（切换器会重新签发）
|
||
"""
|
||
token = secrets.token_hex(32)
|
||
payload = json.dumps({"user_id": user_id, "entity_id": entity_id}, ensure_ascii=False)
|
||
if _redis_available:
|
||
_redis.setex(f"{TOKEN_PREFIX}{token}", TOKEN_TTL, payload)
|
||
else:
|
||
_token_store[token] = payload
|
||
return token
|
||
|
||
|
||
def _resolve_token_data(token: str) -> dict | None:
|
||
"""解析token → {user_id, entity_id}
|
||
- 新格式 JSON → 返回 dict
|
||
- 旧格式 int（改造前）→ 返回 None（强制下线，账套模式需重新登录）
|
||
- 不存在 → None
|
||
"""
|
||
if _redis_available:
|
||
val = _redis.get(f"{TOKEN_PREFIX}{token}")
|
||
if val is None:
|
||
return None
|
||
try:
|
||
data = json.loads(val)
|
||
if isinstance(data, dict) and "user_id" in data:
|
||
return data
|
||
except (json.JSONDecodeError, ValueError, TypeError):
|
||
pass
|
||
# 旧格式纯 int → 强制下线
|
||
return None
|
||
val = _token_store.get(token)
|
||
if val is None:
|
||
return None
|
||
if isinstance(val, dict):
|
||
return val
|
||
try:
|
||
data = json.loads(val)
|
||
if isinstance(data, dict) and "user_id" in data:
|
||
return data
|
||
except (json.JSONDecodeError, ValueError, TypeError):
|
||
pass
|
||
return None
|
||
|
||
|
||
def _resolve_user_id(token: str) -> int | None:
|
||
data = _resolve_token_data(token)
|
||
return int(data["user_id"]) if data else None
|
||
|
||
|
||
def get_token_entity_id(token: str) -> int | None:
|
||
"""从token解析绑定的entity_id（token不存在/旧格式 → None）"""
|
||
data = _resolve_token_data(token)
|
||
if not data:
|
||
return None
|
||
eid = data.get("entity_id")
|
||
return int(eid) if eid else None
|
||
|
||
|
||
def user_has_entity(db: Session, user_id: int, entity_id: int) -> bool:
|
||
"""校验用户是否被授权访问指定企业（账套授权表 user_entities）"""
|
||
from app.models import UserEntity, Entity
|
||
ent = db.query(Entity).filter(Entity.id == entity_id).first()
|
||
if not ent or ent.status != "active":
|
||
return False
|
||
rel = db.query(UserEntity).filter(
|
||
UserEntity.user_id == user_id,
|
||
UserEntity.entity_id == entity_id,
|
||
).first()
|
||
return rel is not None
|
||
|
||
|
||
def extract_bearer_token(request) -> str | None:
|
||
"""从Request提取Bearer token（无/非Bearer格式 → None）"""
|
||
auth = request.headers.get("Authorization", "")
|
||
if auth.startswith("Bearer "):
|
||
return auth[7:].strip()
|
||
return None
|
||
|
||
|
||
def require_auth(
|
||
credentials: HTTPAuthorizationCredentials = Depends(HTTPBearer(auto_error=True)),
|
||
db: Session = Depends(get_db),
|
||
) -> User:
|
||
token = credentials.credentials
|
||
user_id = _resolve_user_id(token)
|
||
if user_id is None:
|
||
raise HTTPException(401, "无效的token，请重新登录")
|
||
user = db.query(User).filter(User.id == user_id).first()
|
||
if not user:
|
||
raise HTTPException(401, "用户不存在")
|
||
return user
|
||
|
||
|
||
def require_role(*roles: str):
|
||
async def role_checker(
|
||
current_user: User = Depends(require_auth),
|
||
) -> User:
|
||
if current_user.role not in roles:
|
||
raise HTTPException(403, f"权限不足: 需要 {', '.join(roles)} 角色")
|
||
return current_user
|
||
return role_checker
|
||
|
||
|
||
def has_permission(user: User, module: str, db: Session = None) -> bool:
|
||
routes, _ = _load_permissions(db)
|
||
return module in routes.get(user.role, [])
|
||
|
||
|
||
def has_action(user: User, action: str, db: Session = None) -> bool:
|
||
_, actions = _load_permissions(db)
|
||
return action in actions.get(user.role, [])
|
||
|
||
|
||
# ─── KPI可见性（按维度/分类过滤） ───
|
||
|
||
def _load_kpi_visibility(db: Session = None):
|
||
"""从RolePermission表加载kpi_visibility配置（独立缓存）"""
|
||
import time
|
||
if not hasattr(_load_kpi_visibility, "_cache"):
|
||
_load_kpi_visibility._cache = {"data": None, "ts": 0}
|
||
cache = _load_kpi_visibility._cache
|
||
now = time.time()
|
||
if db is None or (cache["data"] and now - cache["ts"] < _PERM_CACHE_TTL):
|
||
return cache["data"] or DEFAULT_KPI_VISIBILITY
|
||
try:
|
||
perm = db.query(RolePermission).filter(RolePermission.key == "kpi_visibility").first()
|
||
cache["data"] = perm.value if perm else DEFAULT_KPI_VISIBILITY
|
||
cache["ts"] = now
|
||
return cache["data"]
|
||
except Exception:
|
||
return DEFAULT_KPI_VISIBILITY
|
||
|
||
|
||
def kpi_visible_dims(role: str, db: Session = None) -> list[str]:
|
||
"""返回角色可见的维度列表（空列表=全部可见）"""
|
||
vis = _load_kpi_visibility(db)
|
||
rules = vis.get(role, ["*"])
|
||
if "*" in rules:
|
||
return [] # 空=全部可见
|
||
# 提取维度前缀：finance_* -> finance
|
||
dims = set()
|
||
for r in rules:
|
||
if r.endswith("_*"):
|
||
dims.add(r[:-2])
|
||
return list(dims)
|
||
|
||
|
||
def filter_kpis_by_role(kpis: list, role: str, db: Session = None) -> list:
|
||
"""按角色可见性过滤KPI列表"""
|
||
dims = kpi_visible_dims(role, db)
|
||
if not dims:
|
||
return kpis # 全部可见
|
||
return [k for k in kpis if k.dimension in dims or (hasattr(k, 'dimension') and k.dimension in dims)]
|