Pine
26fa546f67
- 七端口 RBAC、select-identity、JWT、审计 - FastAPI + SQLAlchemy + SQLite，/auth /opc /admin /agents 等路由
118 lines
4.2 KiB
Python
118 lines
4.2 KiB
Python
# -*- coding: utf-8 -*-
|
||
"""FastAPI 依赖：数据库句柄与当前登录用户（JWT + RBAC）。
|
||
|
||
``Database`` 实例由 ``main.py`` 在启动时创建并挂在 ``app.state.db`` 上，
|
||
路由通过 ``Depends(get_db)`` 取用；测试时可替换为临时目录实例。
|
||
"""
|
||
from __future__ import annotations
|
||
|
||
from fastapi import Depends, HTTPException, Request
|
||
|
||
from . import config
|
||
from .jwt import decode_access_token
|
||
from .repositories import Database
|
||
|
||
BEARER_PREFIX = "Bearer "
|
||
|
||
|
||
def extract_bearer_token(request: Request) -> str:
|
||
"""从 Authorization 头提取 Bearer token，没有则返回空串。"""
|
||
auth_header = request.headers.get("Authorization", "")
|
||
if auth_header.startswith(BEARER_PREFIX):
|
||
return auth_header[len(BEARER_PREFIX):].strip()
|
||
return ""
|
||
|
||
|
||
def get_db(request: Request) -> Database:
|
||
return request.app.state.db
|
||
|
||
|
||
def _resolve_identity(user: dict, identity_id: str | None, db: Database) -> dict:
|
||
"""按令牌携带的 identity_id 装配当前身份；缺失则回退到 users.role 单角色。
|
||
|
||
身份即权限来源：从 user_identities 取 role/sub_role/org/region，
|
||
并据此计算 permissions 与数据范围。
|
||
"""
|
||
if identity_id:
|
||
ident = db.identities.get_for_user(identity_id, user["id"])
|
||
if ident and ident.get("status") == "active":
|
||
user["identity_id"] = ident["id"]
|
||
user["port"] = ident["port"]
|
||
user["role"] = ident["role"]
|
||
user["sub_role"] = ident.get("sub_role")
|
||
user["org_id"] = ident.get("org_id")
|
||
user["region_id"] = ident.get("region_id")
|
||
user["identity_name"] = ident.get("name", "")
|
||
|
||
# 动态装配权限与数据范围（每次从库计算，角色变更即时生效）
|
||
user["permissions"] = db.roles.permissions_for(
|
||
user.get("role", "opc_member"), user.get("sub_role"),
|
||
)
|
||
user["scope_region_ids"] = db.regions.visible_region_ids(user.get("region_id"))
|
||
user["scope_level"] = db.regions.level(user.get("region_id"))
|
||
return user
|
||
|
||
|
||
def get_current_user(
|
||
request: Request,
|
||
db: Database = Depends(get_db),
|
||
) -> dict:
|
||
"""校验调用方 JWT，返回当前用户记录（含 role/权限/数据范围）。
|
||
|
||
无效/过期/已吊销/禁用用户一律 401。令牌若绑定 identity_id 则按该身份
|
||
解析；否则回退到 users.role 单角色（兼容旧令牌）。
|
||
"""
|
||
token = extract_bearer_token(request)
|
||
if not token:
|
||
raise HTTPException(status_code=401, detail="No token provided")
|
||
|
||
payload = decode_access_token(token)
|
||
if payload is None:
|
||
raise HTTPException(status_code=401, detail="Invalid or expired token")
|
||
|
||
user_id = payload.get("sub")
|
||
if not db.tokens.session_valid(
|
||
payload.get("jti", ""),
|
||
user_id,
|
||
payload.get("ver", 0),
|
||
):
|
||
raise HTTPException(status_code=401, detail="Invalid or expired token")
|
||
|
||
user = db.users.get_by_id(user_id)
|
||
if user is None or user.get("status") != "active":
|
||
raise HTTPException(status_code=401, detail="Invalid or expired token")
|
||
|
||
return _resolve_identity(user, payload.get("identity_id"), db)
|
||
|
||
|
||
def optional_current_user(
|
||
request: Request,
|
||
db: Database = Depends(get_db),
|
||
) -> dict | None:
|
||
"""可选登录：有有效 JWT 则返回用户，否则返回 None（不报错）。"""
|
||
token = extract_bearer_token(request)
|
||
if not token:
|
||
return None
|
||
payload = decode_access_token(token)
|
||
if payload is None:
|
||
return None
|
||
if not db.tokens.session_valid(
|
||
payload.get("jti", ""), payload.get("sub", ""), payload.get("ver", 0),
|
||
):
|
||
return None
|
||
user = db.users.get_by_id(payload.get("sub"))
|
||
if user is None or user.get("status") != "active":
|
||
return None
|
||
return _resolve_identity(user, payload.get("identity_id"), db)
|
||
|
||
|
||
def require_port(user: dict = Depends(get_current_user)) -> dict:
|
||
"""按端口隔离的资源（如智能体）：必须已解析出端口身份，否则 403。
|
||
|
||
多身份账号在 select-identity 前持有"中性令牌"（无 identity_id，
|
||
port 未解析），此类请求不允许访问按端口隔离的资源。
|
||
"""
|
||
if not user.get("port"):
|
||
raise HTTPException(status_code=403, detail="请先选择身份/端口")
|
||
return user
|