Add api key plugin (#1)

Author: wu-clan

README.md

@@ -1,7 +1,63 @@
-请在此处填写插件使用说明和您的联系方式
+# API Key 插件
 
-如果插件需要付费，请提供付费相关说明
+用户自定义 API Key 管理插件，允许用户生成、管理和使用 API Key 进行接口认证。
 
-如有配套前端插件，请添加前端插件仓库链接说明
+## 配置
 
-插件开发文档：[fba plugin dev](https://fastapi-practices.github.io/fastapi_best_architecture_docs/plugin/dev.html)
+在 `backend/core/conf.py` 中添加以下内容：
+
+```
+##################################################
+# [ Plugin ] api_key
+##################################################
+API_KEY_GENERATE_PREFIX: str = 'fba-'
+```
+
+## 使用方法
+
+### 2. 替换 JWT 认证中间件
+
+编辑 `backend/core/registrar.py`：
+
+```python
+# 原来的导入
+# from backend.middleware.jwt_auth_middleware import JwtAuthMiddleware
+
+# 替换为
+from backend.plugin.api_key.middleware import JwtApiKeyAuthMiddleware
+
+# 原来的 JWT auth
+# app.add_middleware(
+#     AuthenticationMiddleware,
+#     backend=JwtAuthMiddleware(),
+#     on_error=JwtAuthMiddleware.auth_exception_handler,
+# )
+
+# 替换为
+app.add_middleware(
+    AuthenticationMiddleware,
+    backend=JwtApiKeyAuthMiddleware(),
+    on_error=JwtApiKeyAuthMiddleware.auth_exception_handler,
+)
+```
+
+## 认证流程
+
+```mermaid
+flowchart TD
+    A[Authorization: Bearer token] --> B[token.startswith 'fba-' ?]
+    B -->|Yes| C[API Key 认证]
+    B -->|No| D[JWT Token 认证]
+    C --> I[RBAC 权限校验]
+    D --> I
+```
+
+## 权限控制
+
+API Key 完全继承用户权限
+
+如需限制权限，只需创建专门的 API 用户：
+
+1. 创建受限角色（如 `API 只读角色`），分配必要权限
+2. 创建 API 用户，分配该角色
+3. 使用该用户创建 API Key

__init__.py

@@ -0,0 +1,98 @@
+from typing import Annotated
+
+from fastapi import APIRouter, Depends, Path, Query, Request
+
+from backend.common.response.response_code import CustomResponse
+from backend.common.response.response_schema import ResponseModel, ResponseSchemaModel, response_base
+from backend.common.security.jwt import DependsJwtAuth
+from backend.common.security.permission import RequestPermission
+from backend.common.security.rbac import DependsRBAC
+from backend.database.db import CurrentSession, CurrentSessionTransaction
+from backend.plugin.api_key.schema.api_key import (
+    CreateApiKeyParam,
+    DeleteApiKeyParam,
+    GetApiKeyDetail,
+    UpdateApiKeyParam,
+)
+from backend.plugin.api_key.service import api_key_service
+
+router = APIRouter()
+
+
+@router.get('/{pk}', summary='获取 API Key 详情', dependencies=[DependsJwtAuth])
+async def get_api_key(
+    db: CurrentSession, request: Request, pk: Annotated[int, Path(description='通知公告 ID')]
+) -> ResponseSchemaModel[GetApiKeyDetail]:
+    data = await api_key_service.get(db=db, user_id=request.user.id, is_superuser=request.user.is_superuser, pk=pk)
+    return response_base.success(data=data)
+
+
+@router.get('', summary='分页获取所有 API Key', dependencies=[DependsJwtAuth])
+async def get_api_keys_paginated(
+    db: CurrentSession,
+    request: Request,
+    name: Annotated[str | None, Query(description='API Key 名称')] = None,
+    status: Annotated[int | None, Query(description='状态')] = None,
+) -> ResponseSchemaModel[list[GetApiKeyDetail]]:
+    data = await api_key_service.get_list(
+        db=db, user_id=request.user.id, is_superuser=request.user.is_superuser, name=name, status=status
+    )
+    return response_base.success(data=data)
+
+
+@router.post(
+    '',
+    summary='创建 API Key',
+    dependencies=[
+        Depends(RequestPermission('sys:apikey:add')),
+        DependsRBAC,
+    ],
+)
+async def create_api_key(
+    db: CurrentSessionTransaction,
+    request: Request,
+    obj: CreateApiKeyParam,
+) -> ResponseSchemaModel[GetApiKeyDetail]:
+    data = await api_key_service.create(db=db, user_id=request.user.id, obj=obj)
+    return response_base.success(
+        res=CustomResponse(
+            code=200,
+            msg='创建成功，请妥善保管此 API Key，仅显示一次',
+        ),
+        data=data,
+    )
+
+
+@router.put(
+    '/{pk}',
+    summary='更新 API Key',
+    dependencies=[
+        Depends(RequestPermission('sys:apikey:edit')),
+        DependsRBAC,
+    ],
+)
+async def update_api_key(
+    db: CurrentSessionTransaction,
+    request: Request,
+    pk: Annotated[int, Path(description='API Key ID')],
+    obj: UpdateApiKeyParam,
+) -> ResponseModel:
+    await api_key_service.update(db=db, user_id=request.user.id, is_superuser=request.user.is_superuser, pk=pk, obj=obj)
+    return response_base.success()
+
+
+@router.delete(
+    '',
+    summary='批量删除 API Key',
+    dependencies=[
+        Depends(RequestPermission('sys:apikey:del')),
+        DependsRBAC,
+    ],
+)
+async def delete_api_keys(
+    db: CurrentSessionTransaction,
+    request: Request,
+    obj: DeleteApiKeyParam,
+) -> ResponseModel:
+    await api_key_service.delete(db=db, user_id=request.user.id, pks=obj.pks)
+    return response_base.success()

api/v1/__init__.py

@@ -0,0 +1 @@
+from backend.plugin.api_key.crud.crud_api_key import api_key_dao as api_key_dao

api/v1/sys/__init__.py

@@ -0,0 +1,112 @@
+from datetime import timedelta
+
+from sqlalchemy import Select
+from sqlalchemy.ext.asyncio import AsyncSession
+from sqlalchemy_crud_plus import CRUDPlus
+
+from backend.plugin.api_key.model import ApiKey
+from backend.plugin.api_key.schema.api_key import CreateApiKeyParam, UpdateApiKeyParam
+from backend.utils.timezone import timezone
+
+
+class CRUDApiKey(CRUDPlus[ApiKey]):
+    """API Key 数据库操作类"""
+
+    async def get(self, db: AsyncSession, pk: int) -> ApiKey | None:
+        """
+        获取 API Key
+
+        :param db: 数据库会话
+        :param pk: API Key ID
+        :return:
+        """
+        return await self.select_model(db, pk)
+
+    async def get_by_user_id(self, db: AsyncSession, user_id: int, pk: int) -> ApiKey | None:
+        """
+        通过用户 ID 获取 API Key
+
+        :param db: 数据库会话
+        :param user_id: 用户 ID
+        :param pk: API Key ID
+        :return:
+        """
+        return await self.select_model(db, pk, user_id=user_id)
+
+    async def get_by_key(self, db: AsyncSession, key: str) -> ApiKey | None:
+        """
+        通过 key 获取 API Key
+
+        :param db: 数据库会话
+        :param key: API Key
+        :return:
+        """
+        return await self.select_model_by_column(db, key=key)
+
+    async def get_select(self, user_id: int, is_superuser: bool, name: str | None, status: int | None) -> Select:  # noqa: FBT001
+        """
+        获取 API Key 列表查询表达式
+
+        :param user_id: 用户 ID
+        :param is_superuser: 用户超级管理员权限
+        :param name: API Key 名称
+        :param status: 状态
+        :return:
+        """
+        filters = {}
+
+        if not is_superuser:
+            filters['user_id'] = user_id
+        if name is not None:
+            filters['name__like'] = f'%{name}%'
+        if status is not None:
+            filters['status'] = status
+
+        return await self.select_order('id', 'desc', **filters)
+
+    async def create(self, db: AsyncSession, user_id: int, key: str, obj: CreateApiKeyParam) -> ApiKey:
+        """
+        创建 API Key
+
+        :param db: 数据库会话
+        :param user_id: 用户 ID
+        :param key: API Key
+        :param obj: 创建 API Key 参数
+        :return:
+        """
+        dict_obj = obj.model_dump(exclude={'expire_days'})
+        dict_obj['user_id'] = user_id
+        dict_obj['key'] = key
+        if obj.expire_days is not None:
+            dict_obj['expire_time'] = timezone.now() + timedelta(days=obj.expire_days)
+
+        new_api_key = self.model(**dict_obj)
+        db.add(new_api_key)
+        await db.flush()
+
+        return new_api_key
+
+    async def update(self, db: AsyncSession, pk: int, obj: UpdateApiKeyParam) -> int:
+        """
+        更新 API Key
+
+        :param db: 数据库会话
+        :param pk: API Key ID
+        :param obj: 更新 API Key 参数
+        :return:
+        """
+        expire_time = timezone.now() + timedelta(days=obj.expire_days) if obj.expire_days is not None else None
+        return await self.update_model(db, pk, obj, expire_time=expire_time)
+
+    async def delete(self, db: AsyncSession, pks: list[int]) -> int:
+        """
+        批量删除 API Key
+
+        :param db: 数据库会话
+        :param pks: API Key ID 列表
+        :return:
+        """
+        return await self.delete_model_by_column(db, allow_multiple=True, id__in=pks)
+
+
+api_key_dao: CRUDApiKey = CRUDApiKey(ApiKey)

api/v1/sys/api_key.py

@@ -0,0 +1,3 @@
+from backend.plugin.api_key.middleware.jwt_auth_middleware import (
+    JwtApiKeyAuthMiddleware as JwtApiKeyAuthMiddleware,
+)

crud/__init__.py

@@ -0,0 +1,60 @@
+"""
+扩展的 JWT 认证中间件，支持 API Key 认证
+
+使用方式：
+    Authorization: Bearer <jwt_token>  # JWT 认证
+    Authorization: Bearer fba_xxxxx    # API Key 认证（以 fba_ 开头）
+"""
+
+from fastapi import Request
+from starlette.authentication import AuthCredentials
+
+from backend.app.admin.schema.user import GetUserInfoWithRelationDetail
+from backend.common.exception.errors import TokenError
+from backend.common.log import log
+from backend.common.security.jwt import get_jwt_user
+from backend.core.conf import settings
+from backend.database.db import async_db_session
+from backend.middleware.jwt_auth_middleware import AuthenticationError, JwtAuthMiddleware
+from backend.plugin.api_key.utils.key_ops import api_key_verify
+
+
+class JwtApiKeyAuthMiddleware(JwtAuthMiddleware):
+    """JWT 认证中间件（扩展支持 API Key）"""
+
+    async def authenticate(self, request: Request) -> tuple[AuthCredentials, GetUserInfoWithRelationDetail] | None:
+        """
+        认证请求（支持 JWT Token 和 API Key）
+
+        :param request: FastAPI 请求对象
+        :return:
+        """
+        token = self.extract_token(request)
+        if token is None:
+            return None
+
+        if token.startswith(settings.API_KEY_GENERATE_PREFIX):
+            return await self._api_key_authentication(token)
+
+        return await super().authenticate(request)
+
+    @staticmethod
+    async def _api_key_authentication(api_key: str) -> tuple[AuthCredentials, GetUserInfoWithRelationDetail]:
+        """
+        API Key 认证
+
+        :param api_key: API Key
+        :return:
+        """
+        try:
+            async with async_db_session() as db:
+                api_key_obj = await api_key_verify(db=db, key=api_key)
+                user_id = api_key_obj.user_id
+                user = await get_jwt_user(user_id)
+        except TokenError as exc:
+            raise AuthenticationError(code=exc.code, msg=exc.detail, headers=exc.headers)
+        except Exception as e:
+            log.exception(f'API Key 授权异常：{e}')
+            raise AuthenticationError(code=getattr(e, 'code', 500), msg=getattr(e, 'msg', 'Internal Server Error'))
+
+        return AuthCredentials(['authenticated']), user

crud/crud_api_key.py

@@ -0,0 +1 @@
+from backend.plugin.api_key.model.api_key import ApiKey as ApiKey