# -*- coding: utf-8 -*- """接口层依赖:数据库句柄与当前登录用户(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 ..infrastructure.repositories import Database from ..jwt import decode_access_token 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 "" async def get_db(): """每请求一个独立 Database/AsyncSession(绑定共享引擎)。 之前直接返回 app.state.db(单例 AsyncSession)被并发请求复用,会间歇报 ``sqlalchemy.exc.InvalidRequestError: concurrent operations are not permitted``。 改为每请求各建一个 Database 并在请求结束关闭,根除该竞态。 """ db = Database() try: yield db finally: await db.close() async def _resolve_identity(user: dict, db: Database) -> dict: """单账号角色(多身份已收敛):role 由 users.role 提供,仅计算权限与数据范围。""" user["permissions"] = await db.roles.permissions_for( user.get("role", "opc_member"), None, ) user["scope_region_ids"] = await db.regions.visible_region_ids(user.get("region_id")) user["scope_level"] = await db.regions.level(user.get("region_id")) return user async def get_current_user( request: Request, db: Database = Depends(get_db), ) -> dict: """校验调用方 JWT,返回当前用户记录(含 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 await db.tokens.session_valid( payload.get("jti", ""), user_id, payload.get("ver", 0), ): raise HTTPException(status_code=401, detail="Invalid or expired token") user = await 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 await _resolve_identity(user, db) async 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 await db.tokens.session_valid( payload.get("jti", ""), payload.get("sub", ""), payload.get("ver", 0), ): return None user = await db.users.get_by_id(payload.get("sub")) if user is None or user.get("status") != "active": return None return await _resolve_identity(user, db) def require_port(user: dict = Depends(get_current_user)) -> dict: """按端口隔离的资源(如智能体)—— 单角色后不再按端口隔离,仅要求已登录。 账号唯一角色:智能体不再按端口隔离,统一归到该账号(port="")。为兼容 既有 AgentRepository 的 port 过滤,这里补一个恒定空串。 """ user.setdefault("port", "") return user