merge: 安全修复与权限补齐

Author: zhangtao

backend/app/config/db_url.py

@@ -0,0 +1,46 @@
+"""数据库 / Redis 连接串构造。
+
+统一使用百分号编码，避免密码中的 ``@ ! # / :`` 等特殊字符被当作
+URL 分隔符；不要用字符串拼接后再手写 ``quote_plus``，否则用户名、
+库名、host 与密码的编码规则容易不一致。
+"""
+
+from __future__ import annotations
+
+from urllib.parse import quote
+
+from sqlalchemy.engine import URL
+
+
+def build_sqlalchemy_url(
+    drivername: str,
+    *,
+    username: str,
+    password: str,
+    host: str,
+    port: int,
+    database: str | None = None,
+    query: dict[str, str] | None = None,
+) -> str:
+    """构造完整 SQLAlchemy URL（显式保留密码，供 SQLAlchemy 解析）。"""
+    return URL.create(
+        drivername=drivername,
+        username=username or None,
+        password=password or None,
+        host=host or None,
+        port=port or None,
+        database=database or None,
+        query=query or {},
+    ).render_as_string(hide_password=False)
+
+
+def build_redis_url(*, username: str, password: str, host: str, port: int, db: int) -> str:
+    """构造 Redis URL；用户名/密码按 RFC 3986 百分号编码。"""
+    auth = ""
+    if username and password:
+        auth = f"{quote(username, safe='')}:{quote(password, safe='')}@"
+    elif password:
+        auth = f":{quote(password, safe='')}@"
+    elif username:
+        auth = f"{quote(username, safe='')}@"
+    return f"redis://{auth}{host}:{port}/{db}"

backend/app/config/setting.py

@@ -2,12 +2,12 @@
 from functools import lru_cache
 from pathlib import Path
 from typing import Any, Literal
-from urllib.parse import quote_plus
 
 from pydantic import model_validator
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 from app.common.enums import EnvironmentEnum
+from app.config.db_url import build_redis_url, build_sqlalchemy_url
 from app.config.path_conf import ENV_DIR
 
 
@@ -313,12 +313,13 @@ def _warn_insecure_cors(message: str) -> None:
     @property
     def REDIS_URI(self) -> str:
         """构建 Redis 连接 URI（供 slowapi / 其他模块复用）。"""
-        auth_part = ""
-        if self.REDIS_USER and self.REDIS_PASSWORD:
-            auth_part = f"{self.REDIS_USER}:{self.REDIS_PASSWORD}@"
-        elif self.REDIS_PASSWORD:
-            auth_part = f":{self.REDIS_PASSWORD}@"
-        return f"redis://{auth_part}{self.REDIS_HOST}:{self.REDIS_PORT}/{self.REDIS_DB_NAME}"
+        return build_redis_url(
+            username=self.REDIS_USER,
+            password=self.REDIS_PASSWORD,
+            host=self.REDIS_HOST,
+            port=self.REDIS_PORT,
+            db=self.REDIS_DB_NAME,
+        )
 
     @property
     def MIDDLEWARE_LIST(self) -> list[str | None]:
@@ -338,9 +339,24 @@ def ASYNC_DB_URI(self) -> str:
             raise ValueError(f"数据库驱动不支持: {self.DATABASE_TYPE}, 异步数据库请选择 mysql、postgres、sqlite")
         db_connect: str = ""
         if self.DATABASE_TYPE == "mysql":
-            db_connect = f"mysql+aiomysql://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}?charset=utf8mb4"
+            db_connect = build_sqlalchemy_url(
+                "mysql+aiomysql",
+                username=self.DATABASE_USER,
+                password=self.DATABASE_PASSWORD,
+                host=self.DATABASE_HOST,
+                port=self.DATABASE_PORT,
+                database=self.DATABASE_NAME,
+                query={"charset": "utf8mb4"},
+            )
         elif self.DATABASE_TYPE == "postgres":
-            db_connect = f"postgresql+asyncpg://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}"
+            db_connect = build_sqlalchemy_url(
+                "postgresql+asyncpg",
+                username=self.DATABASE_USER,
+                password=self.DATABASE_PASSWORD,
+                host=self.DATABASE_HOST,
+                port=self.DATABASE_PORT,
+                database=self.DATABASE_NAME,
+            )
         else:
             name = self.DATABASE_NAME if self.DATABASE_NAME.endswith(".db") else f"{self.DATABASE_NAME}.db"
             db_connect = f"sqlite+aiosqlite:///{name}"
@@ -352,9 +368,24 @@ def DB_URI(self) -> str:
             raise ValueError(f"数据库驱动不支持: {self.DATABASE_TYPE}, 同步数据库请选择 mysql、postgres、sqlite")
         db_connect: str = ""
         if self.DATABASE_TYPE == "mysql":
-            db_connect = f"mysql+pymysql://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}?charset=utf8mb4"
+            db_connect = build_sqlalchemy_url(
+                "mysql+pymysql",
+                username=self.DATABASE_USER,
+                password=self.DATABASE_PASSWORD,
+                host=self.DATABASE_HOST,
+                port=self.DATABASE_PORT,
+                database=self.DATABASE_NAME,
+                query={"charset": "utf8mb4"},
+            )
         elif self.DATABASE_TYPE == "postgres":
-            db_connect = f"postgresql+psycopg://{self.DATABASE_USER}:{quote_plus(self.DATABASE_PASSWORD)}@{self.DATABASE_HOST}:{self.DATABASE_PORT}/{self.DATABASE_NAME}"
+            db_connect = build_sqlalchemy_url(
+                "postgresql+psycopg",
+                username=self.DATABASE_USER,
+                password=self.DATABASE_PASSWORD,
+                host=self.DATABASE_HOST,
+                port=self.DATABASE_PORT,
+                database=self.DATABASE_NAME,
+            )
         else:
             name = self.DATABASE_NAME if self.DATABASE_NAME.endswith(".db") else f"{self.DATABASE_NAME}.db"
             db_connect = f"sqlite:///{name}"
@@ -379,7 +410,7 @@ def FASTAPI_CONFIG(self) -> dict[str, Any]:
                 404: {"description": "资源不存在"},
                 422: {"description": "请求参数验证错误"},
                 500: {"description": "服务器内部错误"},
-            }
+            },
         }
 
 

backend/app/modules/dbconsole/service.py

@@ -11,11 +11,11 @@
 import asyncio
 import time
 from typing import Any
-from urllib.parse import quote_plus
 
 from sqlalchemy import text
 from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
 
+from app.config.db_url import build_sqlalchemy_url
 from app.core.base_schema import AuthSchema
 from app.core.exceptions import CustomException
 from app.utils.crypto_util import CryptoUtil
@@ -43,9 +43,14 @@ def build_url(db_type: str, host: str, port: int, username: str, password: str,
     driver = _DRIVERS.get(db_type)
     if driver is None:
         raise CustomException(msg=f"不支持的数据库类型: {db_type}（仅 mysql/postgres）")
-    auth = f"{quote_plus(username)}:{quote_plus(password)}"
-    db_part = f"/{quote_plus(database)}" if database else "/"
-    return f"{driver}://{auth}@{host}:{port}{db_part}"
+    return build_sqlalchemy_url(
+        driver,
+        username=username,
+        password=password,
+        host=host,
+        port=port,
+        database=database,
+    )
 
 
 class DbConsoleService:

backend/app/modules/monitor/cache/controller.py

@@ -41,7 +41,7 @@ async def get_monitor_cache_key_controller(cache_name: Annotated[str, Path(descr
     return SuccessResponse(data=result, msg=f"获取缓存{cache_name}的键名列表成功")
 
 
-@CacheRouter.get("/get/value/{cache_name}/{cache_key}", summary="获取缓存值", response_model=ResponseSchema[CacheInfoSchema], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))])
+@CacheRouter.get("/get/value/{cache_name}/{cache_key}", summary="获取缓存值", response_model=ResponseSchema[CacheInfoSchema], dependencies=[Security(AuthPermission(["module_monitor:cache:detail"]))])
 async def get_monitor_cache_value_controller(
     cache_name: Annotated[str, Path(description="缓存名称")],
     cache_key: Annotated[str, Path(description="缓存键名")],
@@ -74,7 +74,9 @@ async def clear_monitor_cache_all_controller(
 # ── Redis 通用键空间浏览（只读）───────────────────────────────────────
 
 
-@CacheRouter.get("/redis/databases", summary="Redis 逻辑库列表（只读）", response_model=ResponseSchema[list[RedisDatabaseSchema]], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))])
+@CacheRouter.get(
+    "/redis/databases", summary="Redis 逻辑库列表（只读）", response_model=ResponseSchema[list[RedisDatabaseSchema]], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))]
+)
 async def list_redis_databases_controller(
     redis: Annotated[Redis, Depends(redis_getter)],
 ) -> JSONResponse:
@@ -90,13 +92,13 @@ async def scan_redis_keys_controller(
     cursor: Annotated[int, Query(ge=0, description="SCAN 游标，首页传 0")] = 0,
     count: Annotated[int, Query(ge=1, le=500, description="单页数量提示")] = 100,
 ) -> JSONResponse:
-    result: RedisKeyPageSchema = await RedisBrowserService.scan_keys(
-        redis=redis, db=db, pattern=pattern, cursor=cursor, count=count
-    )
+    result: RedisKeyPageSchema = await RedisBrowserService.scan_keys(redis=redis, db=db, pattern=pattern, cursor=cursor, count=count)
     return SuccessResponse(data=result, msg="扫描键成功")
 
 
-@CacheRouter.get("/redis/value", summary="读取 Redis 单键详情（只读，敏感键脱敏）", response_model=ResponseSchema[RedisValueSchema], dependencies=[Security(AuthPermission(["module_monitor:cache:query"]))])
+@CacheRouter.get(
+    "/redis/value", summary="读取 Redis 单键详情（只读，敏感键脱敏）", response_model=ResponseSchema[RedisValueSchema], dependencies=[Security(AuthPermission(["module_monitor:cache:detail"]))]
+)
 async def get_redis_value_controller(
     redis: Annotated[Redis, Depends(redis_getter)],
     key: Annotated[str, Query(min_length=1, max_length=512, description="键名")],

backend/app/modules/system/log/controller.py

@@ -59,7 +59,7 @@ async def delete_log_controller(
     return SuccessResponse(msg="删除登录日志成功")
 
 
-@LogRouter.get("/operation/detail/{id}", summary="获取操作日志详情", response_model=ResponseSchema[OperationLogDetailOutSchema], dependencies=[Security(AuthPermission(["module_system:log:query"]))])
+@LogRouter.get("/operation/detail/{id}", summary="获取操作日志详情", response_model=ResponseSchema[OperationLogDetailOutSchema], dependencies=[Security(AuthPermission(["module_system:log:detail"]))])
 async def get_operation_log_detail_controller(
     auth: Annotated[AuthSchema, Security(get_current_user)],
     db: Annotated[AsyncSession, Depends(db_getter)],
@@ -70,7 +70,10 @@ async def get_operation_log_detail_controller(
 
 
 @LogRouter.get(
-    "/operation/list", summary="获取操作日志列表", response_model=ResponseSchema[PageResultSchema[OperationLogOutSchema]], dependencies=[Security(AuthPermission(["module_system:log:query"]))],
+    "/operation/list",
+    summary="获取操作日志列表",
+    response_model=ResponseSchema[PageResultSchema[OperationLogOutSchema]],
+    dependencies=[Security(AuthPermission(["module_system:log:query"]))],
 )
 async def get_operation_log_list_controller(
     auth: Annotated[AuthSchema, Security(get_current_user)],
@@ -104,9 +107,7 @@ async def export_operation_log_list_controller(
     search: Annotated[OperationLogQueryParam, Body()],
 ) -> StreamingResponse:
     operation_log_query_result: list[OperationLogOutSchema] = await OperationLogService(auth, db).get_list(search=search)
-    operation_log_export_result: bytes = await OperationLogService.export_list(
-        operation_log_list=[item.model_dump() for item in operation_log_query_result]
-    )
+    operation_log_export_result: bytes = await OperationLogService.export_list(operation_log_list=[item.model_dump() for item in operation_log_query_result])
 
     return StreamResponse(
         data=bytes2file_response(operation_log_export_result),