76 lines
3.1 KiB
Python
76 lines
3.1 KiB
Python
"""auth.py —— 鉴权依赖(管理员 + 用户级)+ 鉴权异常。
|
|
|
|
- `get_current_user`:受 auth_enabled 开关控制;关=匿名(None),开=校验 token(401 失败)。
|
|
- `require_user`:强制登录(不受开关控制),用于 /api/auth/me、/logout 等必须登录的端点。
|
|
- `require_admin`:管理员;复用 require_user + role 校验。
|
|
- `require_authorization(unit)`:功能门禁;关=放行,开=admin 或 unit 在 authorizations 才放行。
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
from fastapi import Depends, Request
|
|
|
|
from .config import get_settings
|
|
from .db import Database, get_db
|
|
from .logwriter import log_event
|
|
from .services.auth import resolve_token
|
|
|
|
|
|
class AuthError(Exception):
|
|
def __init__(self, status: int, code: str, message: str) -> None:
|
|
super().__init__(message)
|
|
self.status = status
|
|
self.code = code
|
|
self.message = message
|
|
|
|
|
|
def _bearer(request: Request) -> str:
|
|
header = request.headers.get("Authorization", "")
|
|
return header[7:] if header.startswith("Bearer ") else ""
|
|
|
|
|
|
async def get_current_user(request: Request, db: Database = Depends(get_db)) -> dict | None:
|
|
"""用户级鉴权;auth_enabled 为假时返回 None(匿名放行)。"""
|
|
if not get_settings().auth_enabled:
|
|
return None
|
|
token = _bearer(request)
|
|
user = await resolve_token(db, token) if token else None
|
|
if user is None:
|
|
log_event("call_denied", method=request.method, path=request.url.path, detail={"reason": "unauthorized"})
|
|
raise AuthError(401, "UNAUTHORIZED", "未登录或登录已过期")
|
|
request.state.user = user
|
|
return user
|
|
|
|
|
|
async def require_user(request: Request, db: Database = Depends(get_db)) -> dict:
|
|
"""强制登录(不受 auth_enabled 开关控制)。"""
|
|
token = _bearer(request)
|
|
user = await resolve_token(db, token) if token else None
|
|
if user is None:
|
|
log_event("call_denied", method=request.method, path=request.url.path, detail={"reason": "unauthorized"})
|
|
raise AuthError(401, "UNAUTHORIZED", "未登录或登录已过期")
|
|
request.state.user = user
|
|
return user
|
|
|
|
|
|
async def require_admin(request: Request, db: Database = Depends(get_db)) -> dict:
|
|
user = await require_user(request, db)
|
|
if user["role"] != "admin":
|
|
log_event("call_denied", user_id=user["id"], method=request.method, path=request.url.path, detail={"reason": "not_admin"})
|
|
raise AuthError(403, "FORBIDDEN", "非管理员账号")
|
|
return user
|
|
|
|
|
|
def require_authorization(unit: str):
|
|
"""功能门禁依赖工厂;auth_enabled 关=放行,开=admin 或 unit 在 authorizations 放行。"""
|
|
|
|
async def _dep(request: Request, db: Database = Depends(get_db)) -> dict | None:
|
|
if not get_settings().auth_enabled:
|
|
return None
|
|
user = await require_user(request, db)
|
|
if user["role"] != "admin" and unit not in (user.get("authorizations") or []):
|
|
log_event("call_denied", user_id=user["id"], method=request.method, path=request.url.path, detail={"reason": "forbidden", "unit": unit})
|
|
raise AuthError(403, "FORBIDDEN", "无权访问该功能")
|
|
return user
|
|
|
|
return _dep
|