From 68b40b62c34e4cf68ec61cd0f1984f359ed8e970 Mon Sep 17 00:00:00 2001 From: zhangtao <9480807882@qq.com> Date: Thu, 22 Oct 2026 22:17:46 +0800 Subject: [PATCH] =?UTF-8?q?feat(ops):=20P2-2=20SSH=20=E8=BF=90=E7=BB=B4?= =?UTF-8?q?=E6=A8=A1=E5=9D=97=EF=BC=88=E4=B8=BB=E6=9C=BA/=E7=BB=88?= =?UTF-8?q?=E7=AB=AF=20POST=20=E8=BD=AE=E8=AF=A2/=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=89=A7=E8=A1=8C/=E7=89=87=E6=AE=B5/=E6=97=A5=E5=BF=97?= =?UTF-8?q?=E8=B7=9F=E8=B8=AA=EF=BC=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 按 FastHub 移植清单落地 P2-2；**不含本机终端/本机文件/run_host_command**（多租户下等于 RCE，按需求排除）。 后端（新模块 app/modules/ops/ssh） - model：ops_ssh_host（密码/私钥/口令 Fernet 加密，出参只回 has_password/has_private_key）、 ops_ssh_snippet（命令片段） - 主机：CRUD + 唯一名 + 认证方式校验（password 必须有密码、key 必须有私钥）+ 按凭据测试连接 （look_for_keys=False/allow_agent=False，只用显式凭据；私钥支持 Ed25519/RSA/ECDSA） - 终端会话（terminal.py）：进程内会话表 + 读取线程 + 队列；**POST 轮询**（不依赖 WebSocket）； 归属校验（跨用户 403）、单用户 3 个/全局 50 个上限、空闲 30 分钟回收、resize/resync/close - 日志跟踪：在交互 shell 内执行 ，复用轮询会话 - 批量执行：多主机同命令（并发 5、单机超时 15s 可调、stdout/stderr 截断、单机失败不影响其它主机）； 执行记录写审计日志（用户/主机/命令截断 200 字） - 路由 /ops/ssh/*（19 条），全部走 AuthPermission + OperationLogRoute 前端 - api/module_ops/ssh.ts + views/module_ops/ssh/index.vue：主机管理/命令片段/批量执行/终端四个 tab； 终端为轮询式（700ms 拉取，base64 传输，自动滚动，可关闭）；日志跟踪入口 - router：/ops/ssh 懒加载静态路由（隐藏菜单，直接访问） 测试：tests/test_ops_ssh_terminal.py（7 项）+ tests/test_ops_ssh_service.py（12 项） 验证：后端 pytest **219 passed**、ruff 通过、basedpyright 0 errors； 前端 vue-tsc 通过、eslint 无错、prettier 通过、vitest **41 passed**。 --- backend/app/api/v1/routers.py | 2 + backend/app/modules/ops/__init__.py | 0 backend/app/modules/ops/ssh/__init__.py | 0 backend/app/modules/ops/ssh/controller.py | 266 +++++++++ backend/app/modules/ops/ssh/crud.py | 21 + backend/app/modules/ops/ssh/model.py | 35 ++ backend/app/modules/ops/ssh/schema.py | 166 ++++++ backend/app/modules/ops/ssh/service.py | 345 +++++++++++ backend/app/modules/ops/ssh/terminal.py | 193 ++++++ backend/tests/test_ops_ssh_service.py | 218 +++++++ backend/tests/test_ops_ssh_terminal.py | 151 +++++ frontend/web/src/api/module_ops/ssh.ts | 173 ++++++ frontend/web/src/router/routes.ts | 9 + .../web/src/views/module_ops/ssh/index.vue | 559 ++++++++++++++++++ 14 files changed, 2138 insertions(+) create mode 100644 backend/app/modules/ops/__init__.py create mode 100644 backend/app/modules/ops/ssh/__init__.py create mode 100644 backend/app/modules/ops/ssh/controller.py create mode 100644 backend/app/modules/ops/ssh/crud.py create mode 100644 backend/app/modules/ops/ssh/model.py create mode 100644 backend/app/modules/ops/ssh/schema.py create mode 100644 backend/app/modules/ops/ssh/service.py create mode 100644 backend/app/modules/ops/ssh/terminal.py create mode 100644 backend/tests/test_ops_ssh_service.py create mode 100644 backend/tests/test_ops_ssh_terminal.py create mode 100644 frontend/web/src/api/module_ops/ssh.ts create mode 100644 frontend/web/src/views/module_ops/ssh/index.vue diff --git a/backend/app/api/v1/routers.py b/backend/app/api/v1/routers.py index f0c592cab..112984ee6 100644 --- a/backend/app/api/v1/routers.py +++ b/backend/app/api/v1/routers.py @@ -10,6 +10,7 @@ from app.modules.monitor.health.controller import HealthRouter from app.modules.monitor.online.controller import OnlineRouter from app.modules.monitor.server.controller import ServerRouter +from app.modules.ops.ssh.controller import SshRouter from app.modules.system.auth.controller import AuthRouter from app.modules.system.dept.controller import DeptRouter from app.modules.system.dict.controller import DictRouter @@ -63,6 +64,7 @@ "/generator": [GenRouter], "/common": [FileRouter], "/dbconsole": [DbConsoleRouter], + "/ops": [SshRouter], } api_v1 = APIRouter() diff --git a/backend/app/modules/ops/__init__.py b/backend/app/modules/ops/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/modules/ops/ssh/__init__.py b/backend/app/modules/ops/ssh/__init__.py new file mode 100644 index 000000000..e69de29bb diff --git a/backend/app/modules/ops/ssh/controller.py b/backend/app/modules/ops/ssh/controller.py new file mode 100644 index 000000000..f518d08f1 --- /dev/null +++ b/backend/app/modules/ops/ssh/controller.py @@ -0,0 +1,266 @@ +"""SSH 运维接口：主机/片段管理、终端会话（POST 轮询）、批量执行、日志跟踪。""" + +import base64 +from datetime import UTC, datetime +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 ( + BatchExecResultSchema, + BatchExecSchema, + SshHostCreateSchema, + SshHostOutSchema, + SshHostTestSchema, + SshHostUpdateSchema, + SshSnippetCreateSchema, + SshSnippetOutSchema, + SshSnippetUpdateSchema, + TerminalInputSchema, + TerminalLogOpenSchema, + TerminalOpenSchema, + TerminalPollSchema, + TerminalResizeSchema, + TerminalSessionSchema, +) +from .service import SshHostService, SshSnippetService +from .terminal import TerminalManager + +SshRouter = APIRouter(route_class=OperationLogRoute, prefix="/ssh", tags=["SSH运维"]) + +_QUERY = "module_ops:ssh:query" +_TERMINAL = "module_ops:ssh:terminal" +_EXEC = "module_ops:ssh:exec" + + +def _session_schema(session) -> TerminalSessionSchema: + return TerminalSessionSchema( + session_id=session.session_id, + host_id=session.host_id, + host_name=session.host_name, + cols=session.cols, + rows=session.rows, + created_at=datetime.fromtimestamp(session.created_at, tz=UTC).isoformat(timespec="seconds"), + ) + + +# ── 主机管理 ────────────────────────────────────────────────────────── + + +@SshRouter.get("/host/list", summary="查询 SSH 主机列表", response_model=ResponseSchema[list[SshHostOutSchema]]) +async def list_ssh_host_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_QUERY]))], + db: Annotated[AsyncSession, Depends(db_getter)], +) -> JSONResponse: + result = await SshHostService(auth, db).get_list() + return SuccessResponse(data=result, msg="查询主机列表成功") + + +@SshRouter.get("/host/detail/{id}", summary="查询 SSH 主机详情", response_model=ResponseSchema[SshHostOutSchema]) +async def detail_ssh_host_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_QUERY]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="主机ID", ge=1)], +) -> JSONResponse: + result = await SshHostService(auth, db).detail(id=id) + return SuccessResponse(data=result, msg="查询主机详情成功") + + +@SshRouter.post("/host/create", status_code=status.HTTP_201_CREATED, summary="新增 SSH 主机", response_model=ResponseSchema[SshHostOutSchema]) +async def create_ssh_host_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:ssh:create"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[SshHostCreateSchema, Body(description="主机配置（密码/私钥加密存储）")], +) -> JSONResponse: + result = await SshHostService(auth, db).create(data=data) + return SuccessResponse(data=result, msg="新增主机成功") + + +@SshRouter.put("/host/update/{id}", summary="修改 SSH 主机", response_model=ResponseSchema[SshHostOutSchema]) +async def update_ssh_host_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:ssh:update"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="主机ID", ge=1)], + data: Annotated[SshHostUpdateSchema, Body(description="主机配置（凭据不传表示不修改）")], +) -> JSONResponse: + result = await SshHostService(auth, db).update(id=id, data=data) + return SuccessResponse(data=result, msg="修改主机成功") + + +@SshRouter.delete("/host/delete", summary="删除 SSH 主机", response_model=ResponseSchema[None]) +async def delete_ssh_host_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:ssh:delete"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + ids: Annotated[list[int], Body(description="主机ID列表")], +) -> JSONResponse: + await SshHostService(auth, db).delete(ids=ids) + return SuccessResponse(msg="删除主机成功") + + +@SshRouter.post("/host/test", summary="按填写中的凭据测试 SSH 连接（不落库）", response_model=ResponseSchema[bool]) +async def test_ssh_host_config_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:ssh:update"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[SshHostTestSchema, Body(description="连接配置")], +) -> JSONResponse: + result = await SshHostService(auth, db).test_connection(data) + return SuccessResponse(data=result, msg="连接成功") + + +@SshRouter.post("/host/test/{id}", summary="测试已保存的 SSH 主机", response_model=ResponseSchema[bool]) +async def test_ssh_host_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_QUERY]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="主机ID", ge=1)], +) -> JSONResponse: + result = await SshHostService(auth, db).test_by_id(id=id) + return SuccessResponse(data=result, msg="连接成功") + + +# ── 命令片段 ────────────────────────────────────────────────────────── + + +@SshRouter.get("/snippet/list", summary="查询命令片段列表", response_model=ResponseSchema[list[SshSnippetOutSchema]]) +async def list_ssh_snippet_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:snippet:query"]))], + db: Annotated[AsyncSession, Depends(db_getter)], +) -> JSONResponse: + result = await SshSnippetService(auth, db).get_list() + return SuccessResponse(data=result, msg="查询片段列表成功") + + +@SshRouter.post("/snippet/create", status_code=status.HTTP_201_CREATED, summary="新增命令片段", response_model=ResponseSchema[SshSnippetOutSchema]) +async def create_ssh_snippet_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:snippet:create"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[SshSnippetCreateSchema, Body(description="片段内容")], +) -> JSONResponse: + result = await SshSnippetService(auth, db).create(data=data) + return SuccessResponse(data=result, msg="新增片段成功") + + +@SshRouter.put("/snippet/update/{id}", summary="修改命令片段", response_model=ResponseSchema[SshSnippetOutSchema]) +async def update_ssh_snippet_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:snippet:update"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + id: Annotated[int, Path(description="片段ID", ge=1)], + data: Annotated[SshSnippetUpdateSchema, Body(description="片段内容")], +) -> JSONResponse: + result = await SshSnippetService(auth, db).update(id=id, data=data) + return SuccessResponse(data=result, msg="修改片段成功") + + +@SshRouter.delete("/snippet/delete", summary="删除命令片段", response_model=ResponseSchema[None]) +async def delete_ssh_snippet_controller( + auth: Annotated[AuthSchema, Security(AuthPermission(["module_ops:snippet:delete"]))], + db: Annotated[AsyncSession, Depends(db_getter)], + ids: Annotated[list[int], Body(description="片段ID列表")], +) -> JSONResponse: + await SshSnippetService(auth, db).delete(ids=ids) + return SuccessResponse(msg="删除片段成功") + + +# ── 终端会话（POST 轮询）───────────────────────────────────────────── + + +@SshRouter.post("/terminal/open", summary="打开终端会话", response_model=ResponseSchema[TerminalSessionSchema]) +async def open_terminal_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[TerminalOpenSchema, Body(description="主机与终端尺寸")], +) -> JSONResponse: + session = await SshHostService(auth, db).open_terminal(host_id=data.host_id, cols=data.cols, rows=data.rows) + return SuccessResponse(data=_session_schema(session), msg="终端会话已打开") + + +@SshRouter.post("/log/open", summary="打开日志跟踪会话（tail -F）", response_model=ResponseSchema[TerminalSessionSchema]) +async def open_log_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[TerminalLogOpenSchema, Body(description="主机与日志路径")], +) -> JSONResponse: + session = await SshHostService(auth, db).open_log_tail(host_id=data.host_id, path=data.path, lines=data.lines) + return SuccessResponse(data=_session_schema(session), msg="日志跟踪会话已打开") + + +@SshRouter.get("/terminal/{session_id}/poll", summary="轮询终端输出（长轮询）", response_model=ResponseSchema[TerminalPollSchema]) +async def poll_terminal_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + session_id: Annotated[str, Path(min_length=8, max_length=64, description="会话ID")], + timeout: Annotated[float, Query(ge=0, le=5, description="长轮询等待秒数")] = 2.0, +) -> JSONResponse: + data, closed, exit_status = TerminalManager.poll(session_id, auth.user.id or 0, timeout) + return SuccessResponse( + data=TerminalPollSchema( + data=base64.b64encode(data).decode() if data else "", + closed=closed, + exit_status=exit_status, + ), + msg="轮询成功", + ) + + +@SshRouter.post("/terminal/{session_id}/input", summary="终端输入（base64）", response_model=ResponseSchema[None]) +async def input_terminal_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + session_id: Annotated[str, Path(min_length=8, max_length=64)], + data: Annotated[TerminalInputSchema, Body(description="base64 输入")], +) -> JSONResponse: + try: + raw = base64.b64decode(data.data, validate=True) + except ValueError as exc: # binascii.Error 是 ValueError 子类：base64 非法 + from app.core.exceptions import CustomException + + raise CustomException(msg="输入内容不是合法的 base64") from exc + TerminalManager.write(session_id, auth.user.id or 0, raw) + return SuccessResponse(msg="输入成功") + + +@SshRouter.post("/terminal/{session_id}/resync", summary="重同步会话（清空积压输出）", response_model=ResponseSchema[None]) +async def resync_terminal_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + session_id: Annotated[str, Path(min_length=8, max_length=64)], +) -> JSONResponse: + TerminalManager.poll(session_id, auth.user.id or 0, 0) + return SuccessResponse(msg="已重同步") + + +@SshRouter.post("/terminal/{session_id}/resize", summary="同步终端尺寸", response_model=ResponseSchema[None]) +async def resize_terminal_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + session_id: Annotated[str, Path(min_length=8, max_length=64)], + data: Annotated[TerminalResizeSchema, Body(description="列/行")], +) -> JSONResponse: + TerminalManager.resize(session_id, auth.user.id or 0, data.cols, data.rows) + return SuccessResponse(msg="尺寸已同步") + + +@SshRouter.delete("/terminal/{session_id}", summary="关闭终端会话", response_model=ResponseSchema[None]) +async def close_terminal_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_TERMINAL]))], + session_id: Annotated[str, Path(min_length=8, max_length=64)], +) -> JSONResponse: + TerminalManager.close(session_id, auth.user.id or 0) + return SuccessResponse(msg="终端会话已关闭") + + +# ── 批量执行 ────────────────────────────────────────────────────────── + + +@SshRouter.post("/batch-exec", summary="多主机批量执行命令", response_model=ResponseSchema[list[BatchExecResultSchema]]) +async def batch_exec_controller( + auth: Annotated[AuthSchema, Security(AuthPermission([_EXEC]))], + db: Annotated[AsyncSession, Depends(db_getter)], + data: Annotated[BatchExecSchema, Body(description="主机列表与命令")], +) -> JSONResponse: + results = await SshHostService(auth, db).batch_execute( + host_ids=data.host_ids, command=data.command, timeout=data.timeout + ) + return SuccessResponse(data=results, msg="批量执行完成") diff --git a/backend/app/modules/ops/ssh/crud.py b/backend/app/modules/ops/ssh/crud.py new file mode 100644 index 000000000..34ce7e874 --- /dev/null +++ b/backend/app/modules/ops/ssh/crud.py @@ -0,0 +1,21 @@ +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.base_crud import CRUDBase +from app.core.base_schema import AuthSchema + +from .model import SshHostModel, SshSnippetModel +from .schema import SshHostCreateSchema, SshHostUpdateSchema, SshSnippetCreateSchema, SshSnippetUpdateSchema + + +class SshHostCRUD(CRUDBase[SshHostModel, SshHostCreateSchema, SshHostUpdateSchema]): + """SSH 主机数据层""" + + def __init__(self, auth: AuthSchema, db: AsyncSession) -> None: + super().__init__(model=SshHostModel, auth=auth, db=db) + + +class SshSnippetCRUD(CRUDBase[SshSnippetModel, SshSnippetCreateSchema, SshSnippetUpdateSchema]): + """命令片段数据层""" + + def __init__(self, auth: AuthSchema, db: AsyncSession) -> None: + super().__init__(model=SshSnippetModel, auth=auth, db=db) diff --git a/backend/app/modules/ops/ssh/model.py b/backend/app/modules/ops/ssh/model.py new file mode 100644 index 000000000..7696e7497 --- /dev/null +++ b/backend/app/modules/ops/ssh/model.py @@ -0,0 +1,35 @@ +"""SSH 运维：主机与命令片段（凭据加密存储）。""" + +from sqlalchemy import Integer, String, Text +from sqlalchemy.orm import Mapped, mapped_column + +from app.core.base_model import ModelMixin, UserMixin + + +class SshHostModel(ModelMixin, UserMixin): + """SSH 主机配置（密码/私钥 Fernet 加密；出参不含任何凭据）""" + + __tablename__: str = "ops_ssh_host" + __table_args__: dict[str, str] = {"comment": "SSH运维主机表"} + + name: Mapped[str] = mapped_column(String(64), unique=True, nullable=False, index=True, comment="主机名称") + host: Mapped[str] = mapped_column(String(255), nullable=False, comment="主机地址") + port: Mapped[int] = mapped_column(Integer, default=22, nullable=False, comment="SSH端口") + username: Mapped[str] = mapped_column(String(64), nullable=False, comment="登录用户名") + auth_type: Mapped[str] = mapped_column(String(16), default="password", nullable=False, comment="认证方式(password/key)") + password: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="密码(Fernet加密)") + private_key: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="私钥(Fernet加密)") + passphrase: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="私钥口令(Fernet加密)") + 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="备注") + + +class SshSnippetModel(ModelMixin, UserMixin): + """常用命令片段""" + + __tablename__: str = "ops_ssh_snippet" + __table_args__: dict[str, str] = {"comment": "SSH命令片段表"} + + name: Mapped[str] = mapped_column(String(64), nullable=False, comment="片段名称") + command: Mapped[str] = mapped_column(Text, nullable=False, comment="命令内容") + description: Mapped[str | None] = mapped_column(Text, default=None, nullable=True, comment="备注") diff --git a/backend/app/modules/ops/ssh/schema.py b/backend/app/modules/ops/ssh/schema.py new file mode 100644 index 000000000..d7c097e16 --- /dev/null +++ b/backend/app/modules/ops/ssh/schema.py @@ -0,0 +1,166 @@ +"""SSH 运维数据契约。""" + +from typing import Literal + +from pydantic import BaseModel, ConfigDict, Field, field_validator + + +class SshHostBaseSchema(BaseModel): + """主机公共字段（不含凭据）""" + + name: str = Field(..., min_length=1, max_length=64, description="主机名称（唯一）") + host: str = Field(..., min_length=1, max_length=255, description="主机地址") + port: int = Field(22, ge=1, le=65535, description="SSH 端口") + username: str = Field(..., min_length=1, max_length=64, description="登录用户名") + auth_type: Literal["password", "key"] = Field("password", description="认证方式") + status: int = Field(0, ge=0, le=1, description="状态(0:启用 1:停用)") + description: str | None = Field(None, max_length=500, description="备注") + + +class SshHostCreateSchema(SshHostBaseSchema): + """新建主机（按认证方式提供密码或私钥）""" + + password: str | None = Field(None, max_length=500, description="密码（auth_type=password）") + private_key: str | None = Field(None, max_length=20000, description="私钥内容（auth_type=key）") + passphrase: str | None = Field(None, max_length=500, description="私钥口令（可选）") + + @field_validator("name") + @classmethod + def validate_name(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("主机名称不能为空") + return value + + +class SshHostUpdateSchema(SshHostBaseSchema): + """修改主机（凭据字段不传表示不修改）""" + + password: str | None = Field(None, max_length=500) + private_key: str | None = Field(None, max_length=20000) + passphrase: str | None = Field(None, max_length=500) + + +class SshHostOutSchema(SshHostBaseSchema): + """主机出参（不含任何凭据）""" + + id: int = Field(..., description="主机ID") + has_password: bool = Field(False, description="是否已设置密码") + has_private_key: bool = Field(False, description="是否已设置私钥") + + model_config = ConfigDict(from_attributes=True) + + +class SshHostTestSchema(BaseModel): + """按填写中的凭据测试（不落库）""" + + host: str = Field(..., min_length=1, max_length=255) + port: int = Field(22, ge=1, le=65535) + username: str = Field(..., min_length=1, max_length=64) + auth_type: Literal["password", "key"] = "password" + password: str | None = Field(None, max_length=500) + private_key: str | None = Field(None, max_length=20000) + passphrase: str | None = Field(None, max_length=500) + + +class SshSnippetBaseSchema(BaseModel): + """命令片段公共字段""" + + name: str = Field(..., min_length=1, max_length=64, description="片段名称") + command: str = Field(..., min_length=1, max_length=4000, description="命令内容") + description: str | None = Field(None, max_length=500, description="备注") + + +class SshSnippetCreateSchema(SshSnippetBaseSchema): + """新建片段""" + + +class SshSnippetUpdateSchema(SshSnippetBaseSchema): + """修改片段""" + + +class SshSnippetOutSchema(SshSnippetBaseSchema): + """片段出参""" + + id: int = Field(..., description="片段ID") + + model_config = ConfigDict(from_attributes=True) + + +class TerminalOpenSchema(BaseModel): + """打开终端会话""" + + host_id: int = Field(..., ge=1, description="SSH 主机ID") + cols: int = Field(120, ge=20, le=400, description="终端列数") + rows: int = Field(30, ge=5, le=200, description="终端行数") + + +class TerminalLogOpenSchema(BaseModel): + """打开日志跟踪会话（tail -F）""" + + host_id: int = Field(..., ge=1, description="SSH 主机ID") + path: str = Field(..., min_length=1, max_length=512, description="日志文件路径") + lines: int = Field(200, ge=1, le=2000, description="初始行数") + + +class TerminalInputSchema(BaseModel): + """终端输入（base64 编码的原始字节）""" + + data: str = Field(..., max_length=20000, description="base64 编码的输入内容") + + +class TerminalResizeSchema(BaseModel): + """终端尺寸同步""" + + cols: int = Field(..., ge=20, le=400) + rows: int = Field(..., ge=5, le=200) + + +class TerminalSessionSchema(BaseModel): + """终端会话信息""" + + session_id: str = Field(..., description="会话ID") + host_id: int = Field(..., description="主机ID") + host_name: str = Field(..., description="主机名称") + cols: int = Field(..., description="列数") + rows: int = Field(..., description="行数") + created_at: str = Field(..., description="创建时间（ISO）") + + +class TerminalPollSchema(BaseModel): + """终端轮询结果""" + + data: str = Field("", description="base64 编码的输出（可能为空）") + closed: bool = Field(False, description="会话/通道是否已关闭") + exit_status: int | None = Field(None, description="退出码（会话结束时可得）") + + +class BatchExecSchema(BaseModel): + """多主机批量执行""" + + host_ids: list[int] = Field(..., min_length=1, max_length=50, description="主机ID列表（最多 50）") + command: str = Field(..., min_length=1, max_length=2000, description="要执行的命令") + timeout: int = Field(15, ge=1, le=120, description="每台主机执行超时（秒）") + + @field_validator("command") + @classmethod + def validate_command(cls, value: str) -> str: + value = value.strip() + if not value: + raise ValueError("命令不能为空") + if "\x00" in value: + raise ValueError("命令不能包含空字节") + return value + + +class BatchExecResultSchema(BaseModel): + """单台主机的批量执行结果""" + + host_id: int = Field(..., description="主机ID") + host_name: str = Field(..., description="主机名称") + ok: bool = Field(False, description="是否执行成功") + exit_status: int | None = Field(None, description="退出码") + stdout: str = Field("", description="标准输出（截断）") + stderr: str = Field("", description="标准错误（截断）") + error: str = Field("", description="连接/执行错误") + elapsed_ms: int = Field(0, description="耗时（毫秒）") diff --git a/backend/app/modules/ops/ssh/service.py b/backend/app/modules/ops/ssh/service.py new file mode 100644 index 000000000..613b7bba4 --- /dev/null +++ b/backend/app/modules/ops/ssh/service.py @@ -0,0 +1,345 @@ +"""SSH 运维服务：主机/片段 CRUD、连接探测、批量执行、终端建连。 + +安全： +- 凭据（密码/私钥/口令）Fernet 加密存储，出参永不返回； +- 批量执行与终端只允许「已启用」主机； +- 执行命令会记录审计日志（用户/主机/命令截断 200 字），不记录凭据。 +""" + +import asyncio +import io +import shlex +import time + +import paramiko +from sqlalchemy.ext.asyncio import AsyncSession + +from app.core.base_schema import AuthSchema +from app.core.exceptions import CustomException +from app.core.logger import logger +from app.utils.crypto_util import CryptoUtil + +from .crud import SshHostCRUD, SshSnippetCRUD +from .model import SshHostModel +from .schema import ( + BatchExecResultSchema, + SshHostCreateSchema, + SshHostOutSchema, + SshHostTestSchema, + SshHostUpdateSchema, + SshSnippetCreateSchema, + SshSnippetOutSchema, + SshSnippetUpdateSchema, +) +from .terminal import TerminalManager + +_CONNECT_TIMEOUT = 8 +_MAX_CAPTURE = 4000 +_BATCH_CONCURRENCY = 5 + + +def _load_private_key(text: str | None, passphrase: str | None) -> paramiko.PKey: + """解析私钥（支持 Ed25519 / RSA / ECDSA）。""" + if not text: + raise CustomException(msg="该主机未配置私钥") + for key_cls in (paramiko.Ed25519Key, paramiko.RSAKey, paramiko.ECDSAKey): + try: + return key_cls.from_private_key(io.StringIO(text), password=passphrase or None) + except paramiko.SSHException: + continue + except Exception: # noqa: BLE001 - 口令错误等统一落到最终异常 + continue + raise CustomException(msg="无法解析私钥（支持 Ed25519 / RSA / ECDSA）") + + +def connect_client( + *, + host: str, + port: int, + username: str, + auth_type: str, + password: str | None, + private_key: str | None, + passphrase: str | None, +) -> paramiko.SSHClient: + """建立 SSH 连接（禁用 agent/默认密钥，只用显式凭据）。""" + client = paramiko.SSHClient() + client.set_missing_host_key_policy(paramiko.AutoAddPolicy()) + pkey = _load_private_key(private_key, passphrase) if auth_type == "key" else None + try: + client.connect( + hostname=host, + port=port, + username=username, + password=None if pkey is not None else (password or ""), + pkey=pkey, + timeout=_CONNECT_TIMEOUT, + banner_timeout=_CONNECT_TIMEOUT, + auth_timeout=_CONNECT_TIMEOUT, + look_for_keys=False, + allow_agent=False, + ) + except Exception: + client.close() + raise + return client + + +class SshHostService: + """SSH 主机管理""" + + def __init__(self, auth: AuthSchema, db: AsyncSession) -> None: + self.auth = auth + self.db = db + + def _crud(self) -> SshHostCRUD: + return SshHostCRUD(self.auth, self.db) + + @staticmethod + def _to_out(row: SshHostModel) -> SshHostOutSchema: + out = SshHostOutSchema.model_validate(row) + out.has_password = bool(row.password) + out.has_private_key = bool(row.private_key) + return out + + async def get_list(self) -> list[SshHostOutSchema]: + return [self._to_out(row) for row in await self._crud().get_list()] + + async def detail(self, id: int) -> SshHostOutSchema: + return self._to_out(await self._crud().get_or_404(id=id, msg="SSH 主机不存在")) + + async def create(self, data: SshHostCreateSchema) -> SshHostOutSchema: + if await self._crud().get(name=data.name): + raise CustomException(msg="创建失败，主机名称已存在") + if data.auth_type == "password" and not data.password: + raise CustomException(msg="密码认证必须提供密码") + if data.auth_type == "key" and not data.private_key: + raise CustomException(msg="密钥认证必须提供私钥") + payload = data.model_dump() + payload["password"] = CryptoUtil.encrypt(data.password) if data.password else None + payload["private_key"] = CryptoUtil.encrypt(data.private_key) if data.private_key else None + payload["passphrase"] = CryptoUtil.encrypt(data.passphrase) if data.passphrase else None + row = await self._crud().create(data=payload) + return self._to_out(row) + + async def update(self, id: int, data: SshHostUpdateSchema) -> SshHostOutSchema: + await self._crud().get_or_404(id=id, msg="SSH 主机不存在") + exist = await self._crud().get(name=data.name) + if exist and exist.id != id: + raise CustomException(msg="更新失败，主机名称已存在") + payload = data.model_dump(exclude_unset=True) + for field in ("password", "private_key", "passphrase"): + value = payload.get(field) + if value: + payload[field] = CryptoUtil.encrypt(value) + else: + payload.pop(field, 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: SshHostTestSchema | SshHostModel) -> bool: + """按给定凭据或已保存主机测试连接（失败抛 CustomException）。""" + if isinstance(data, SshHostModel): + password = CryptoUtil.decrypt(data.password) + private_key = CryptoUtil.decrypt(data.private_key) + passphrase = CryptoUtil.decrypt(data.passphrase) + host, port, username, auth_type = data.host, data.port, data.username, data.auth_type + else: + password, private_key, passphrase = data.password, data.private_key, data.passphrase + host, port, username, auth_type = data.host, data.port, data.username, data.auth_type + + def _probe() -> None: + client = connect_client( + host=host, + port=port, + username=username, + auth_type=auth_type, + password=password, + private_key=private_key, + passphrase=passphrase, + ) + client.close() + + try: + await asyncio.wait_for(asyncio.to_thread(_probe), timeout=_CONNECT_TIMEOUT + 5) + except TimeoutError as exc: + raise CustomException(msg="连接超时") 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 + return True + + async def test_by_id(self, id: int) -> bool: + row = await self._crud().get_or_404(id=id, msg="SSH 主机不存在") + return await self.test_connection(row) + + async def _enabled_host(self, id: int) -> SshHostModel: + row = await self._crud().get_or_404(id=id, msg="SSH 主机不存在") + if row.status != 0: + raise CustomException(msg="该主机已停用") + return row + + async def open_terminal(self, host_id: int, cols: int, rows: int): + """打开一个交互终端会话（paramiko invoke_shell）。""" + row = await self._enabled_host(host_id) + password = CryptoUtil.decrypt(row.password) + private_key = CryptoUtil.decrypt(row.private_key) + passphrase = CryptoUtil.decrypt(row.passphrase) + + def _open() -> tuple[paramiko.SSHClient, paramiko.Channel]: + client = connect_client( + host=row.host, + port=row.port, + username=row.username, + auth_type=row.auth_type, + password=password, + private_key=private_key, + passphrase=passphrase, + ) + channel = client.invoke_shell(width=cols, height=rows) + return client, channel + + try: + client, channel = await asyncio.wait_for(asyncio.to_thread(_open), timeout=_CONNECT_TIMEOUT + 5) + except TimeoutError as exc: + raise CustomException(msg="终端连接超时") 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 + return TerminalManager.open( + user_id=self.auth.user.id or 0, + host_id=row.id, + host_name=row.name, + client=client, + channel=channel, + cols=cols, + rows=rows, + ) + + async def open_log_tail(self, host_id: int, path: str, lines: int): + """打开日志跟踪会话（在交互 shell 里执行 tail -n N -F）。""" + session = await self.open_terminal(host_id, cols=160, rows=40) + logger.info(f"SSH 日志跟踪: user={self.auth.user.username} host={session.host_name} path={path}") + command = f"tail -n {int(lines)} -F -- {shlex.quote(path)}\r" + TerminalManager.write(session.session_id, self.auth.user.id or 0, command.encode()) + return session + + async def batch_execute(self, host_ids: list[int], command: str, timeout: int) -> list[BatchExecResultSchema]: + """在多台主机上执行同一条命令（并发上限 5，单机超时）。""" + rows = {row.id: row for row in await self._crud().get_list()} + logger.info( + f"SSH 批量执行: user={self.auth.user.username} hosts={host_ids} cmd={command[:200]}" + ) + semaphore = asyncio.Semaphore(_BATCH_CONCURRENCY) + + async def _one(host_id: int) -> BatchExecResultSchema: + row = rows.get(host_id) + if row is None: + return BatchExecResultSchema( + host_id=host_id, host_name="", ok=False, exit_status=None, + stdout="", stderr="", error="主机不存在", elapsed_ms=0, + ) + if row.status != 0: + return BatchExecResultSchema( + host_id=host_id, host_name=row.name, ok=False, exit_status=None, + stdout="", stderr="", error="主机已停用", elapsed_ms=0, + ) + async with semaphore: + started = time.perf_counter() + try: + result = await asyncio.wait_for( + asyncio.to_thread(self._exec_on_host, row, command, timeout), + timeout=timeout + 10, + ) + except TimeoutError: + return BatchExecResultSchema( + host_id=host_id, + host_name=row.name, + ok=False, + exit_status=None, + stdout="", + stderr="", + error=f"执行超时（>{timeout + 10}s）", + elapsed_ms=int((time.perf_counter() - started) * 1000), + ) + except Exception as exc: # noqa: BLE001 - 单机失败不影响其它主机 + reason = str(exc).strip().splitlines()[0][:200] if str(exc).strip() else type(exc).__name__ + return BatchExecResultSchema( + host_id=host_id, + host_name=row.name, + ok=False, + exit_status=None, + stdout="", + stderr="", + error=f"执行失败: {reason}", + elapsed_ms=int((time.perf_counter() - started) * 1000), + ) + result.elapsed_ms = int((time.perf_counter() - started) * 1000) + return result + + return list(await asyncio.gather(*(_one(host_id) for host_id in host_ids))) + + def _exec_on_host(self, row: SshHostModel, command: str, timeout: int) -> BatchExecResultSchema: + """在单台主机执行命令（同步，跑在线程里）。""" + password = CryptoUtil.decrypt(row.password) + private_key = CryptoUtil.decrypt(row.private_key) + passphrase = CryptoUtil.decrypt(row.passphrase) + client = connect_client( + host=row.host, + port=row.port, + username=row.username, + auth_type=row.auth_type, + password=password, + private_key=private_key, + passphrase=passphrase, + ) + try: + _, stdout, stderr = client.exec_command(command, timeout=timeout) + out = stdout.read().decode("utf-8", errors="replace")[:_MAX_CAPTURE] + err = stderr.read().decode("utf-8", errors="replace")[:_MAX_CAPTURE] + exit_status = stdout.channel.recv_exit_status() + finally: + client.close() + return BatchExecResultSchema( + host_id=row.id, + host_name=row.name, + ok=exit_status == 0, + exit_status=exit_status, + stdout=out, + stderr=err, + error="", + elapsed_ms=0, + ) + + +class SshSnippetService: + """命令片段管理""" + + def __init__(self, auth: AuthSchema, db: AsyncSession) -> None: + self.auth = auth + self.db = db + + def _crud(self) -> SshSnippetCRUD: + return SshSnippetCRUD(self.auth, self.db) + + async def get_list(self) -> list[SshSnippetOutSchema]: + return [SshSnippetOutSchema.model_validate(row) for row in await self._crud().get_list()] + + async def create(self, data: SshSnippetCreateSchema) -> SshSnippetOutSchema: + row = await self._crud().create(data=data.model_dump()) + return SshSnippetOutSchema.model_validate(row) + + async def update(self, id: int, data: SshSnippetUpdateSchema) -> SshSnippetOutSchema: + await self._crud().get_or_404(id=id, msg="命令片段不存在") + row = await self._crud().update(id=id, data=data.model_dump()) + return SshSnippetOutSchema.model_validate(row) + + async def delete(self, ids: list[int]) -> None: + await self._crud().delete(ids=ids) diff --git a/backend/app/modules/ops/ssh/terminal.py b/backend/app/modules/ops/ssh/terminal.py new file mode 100644 index 000000000..2c011ea93 --- /dev/null +++ b/backend/app/modules/ops/ssh/terminal.py @@ -0,0 +1,193 @@ +"""SSH 终端会话管理（进程内、**POST 轮询**、不依赖 WebSocket）。 + +为什么用 POST 轮询：Web 端已有一条 WebSocket 聊天连接，终端再开一条 WS 会多一套握手/鉴权分支； +轮询在局域网/内网管理场景足够（延迟 ~200ms），实现与排障都更简单（FastHub 也采用同样取舍）。 + +资源与安全： +- 会话归创建者所有（跨用户访问返回 403）； +- 单用户最多 3 个、全局最多 50 个；空闲 30 分钟自动回收； +- 主机凭据只在建连时解密使用，不存入会话对象； +- 读取线程只做一件事：把通道输出塞进无界队列（单会话输出量由终端自身流量决定）。 +""" + +import queue +import threading +import time +import uuid +from dataclasses import dataclass, field + +import paramiko + +from app.common.enums import RET +from app.core.exceptions import CustomException +from app.core.logger import logger + +MAX_SESSIONS_PER_USER = 3 +MAX_SESSIONS_TOTAL = 50 +IDLE_TIMEOUT = 30 * 60 +_READ_CHUNK = 4096 +_POLL_INTERVAL = 0.15 + + +@dataclass +class TerminalSession: + """一个交互终端会话""" + + session_id: str + user_id: int + host_id: int + host_name: str + client: paramiko.SSHClient + channel: paramiko.Channel + cols: int + rows: int + created_at: float + last_active: float + output: "queue.Queue[bytes]" = field(default_factory=queue.Queue) + closed: bool = False + exit_status: int | None = None + + +class TerminalManager: + """进程内终端会话表（单进程假设：多 worker 部署时会话仅落在处理请求的进程）。""" + + _sessions: dict[str, TerminalSession] = {} + + @classmethod + def _close_impl(cls, session_id: str) -> None: + sess = cls._sessions.pop(session_id, None) + if sess is None: + return + sess.closed = True + for closer in (sess.channel.close, sess.client.close): + try: + closer() + except Exception: # noqa: BLE001 - 关闭失败不影响后续清理 + continue + + @classmethod + def _reap(cls) -> None: + now = time.monotonic() + for session_id, sess in list(cls._sessions.items()): + if sess.closed or now - sess.last_active > IDLE_TIMEOUT: + logger.info(f"回收终端会话: {session_id}（closed={sess.closed}）") + cls._close_impl(session_id) + + @classmethod + def open( + cls, + *, + user_id: int, + host_id: int, + host_name: str, + client: paramiko.SSHClient, + channel: paramiko.Channel, + cols: int, + rows: int, + ) -> TerminalSession: + """登记一个已建连的终端会话并启动读取线程。""" + cls._reap() + owned = [item for item in cls._sessions.values() if item.user_id == user_id] + if len(owned) >= MAX_SESSIONS_PER_USER: + raise CustomException(msg=f"单用户最多同时打开 {MAX_SESSIONS_PER_USER} 个终端会话，请先关闭空闲会话") + if len(cls._sessions) >= MAX_SESSIONS_TOTAL: + raise CustomException(msg="终端会话数已达上限，请稍后再试") + + session_id = uuid.uuid4().hex + now = time.monotonic() + sess = TerminalSession( + session_id=session_id, + user_id=user_id, + host_id=host_id, + host_name=host_name, + client=client, + channel=channel, + cols=cols, + rows=rows, + created_at=now, + last_active=now, + ) + try: + channel.settimeout(0.0) + except Exception: # noqa: BLE001 - 部分通道不支持 settimeout + pass + threading.Thread(target=cls._reader, args=(sess,), daemon=True).start() + cls._sessions[session_id] = sess + return sess + + @staticmethod + def _reader(sess: TerminalSession) -> None: + """读取线程：把通道输出塞进队列；通道结束即标记 closed。""" + while not sess.closed: + try: + data = sess.channel.recv(_READ_CHUNK) + except Exception: # noqa: BLE001 - 通道关闭/超时都视为结束 + break + if not data: + break + sess.output.put(data) + sess.closed = True + try: + sess.exit_status = sess.channel.recv_exit_status() + except Exception: # noqa: BLE001 - 已关闭时拿不到退出码 + sess.exit_status = None + + @classmethod + def get(cls, session_id: str, user_id: int) -> TerminalSession: + """取会话并做归属校验（非本人 403）。""" + sess = cls._sessions.get(session_id) + if sess is None or sess.closed: + raise CustomException(msg="终端会话不存在或已关闭") + if sess.user_id != user_id: + raise CustomException(msg="无权访问该终端会话", code=RET.FORBIDDEN.code) + sess.last_active = time.monotonic() + return sess + + @classmethod + def poll(cls, session_id: str, user_id: int, timeout: float = 2.0) -> tuple[bytes, bool, int | None]: + """长轮询：最多等 timeout 秒，返回累计输出。""" + sess = cls.get(session_id, user_id) + chunks: list[bytes] = [] + deadline = time.monotonic() + max(0.0, min(timeout, 5.0)) + while True: + while True: + try: + chunks.append(sess.output.get_nowait()) + except queue.Empty: + break + if chunks or sess.closed or time.monotonic() >= deadline: + break + time.sleep(_POLL_INTERVAL) + return b"".join(chunks), sess.closed, sess.exit_status + + @classmethod + def write(cls, session_id: str, user_id: int, data: bytes) -> None: + sess = cls.get(session_id, user_id) + try: + sess.channel.sendall(data) + except CustomException: + raise + except Exception as exc: # noqa: BLE001 - 通道关闭时统一转业务异常 + raise CustomException(msg=f"终端写入失败: {type(exc).__name__}") from exc + + @classmethod + def resize(cls, session_id: str, user_id: int, cols: int, rows: int) -> None: + sess = cls.get(session_id, user_id) + try: + sess.channel.resize_pty(width=cols, height=rows) + sess.cols, sess.rows = cols, rows + except CustomException: + raise + except Exception as exc: # noqa: BLE001 + raise CustomException(msg=f"终端尺寸同步失败: {type(exc).__name__}") from exc + + @classmethod + def close(cls, session_id: str, user_id: int | None = None) -> None: + """关闭会话；传 user_id 时校验归属（内部回收不传）。""" + if user_id is not None: + cls.get(session_id, user_id) + cls._close_impl(session_id) + + @classmethod + def count(cls) -> int: + return len(cls._sessions) diff --git a/backend/tests/test_ops_ssh_service.py b/backend/tests/test_ops_ssh_service.py new file mode 100644 index 000000000..fda0e7fe1 --- /dev/null +++ b/backend/tests/test_ops_ssh_service.py @@ -0,0 +1,218 @@ +"""SSH 运维服务测试：凭据加密、认证方式校验、私钥解析、批量执行容错。""" + +import io +from types import SimpleNamespace +from typing import Any + +import paramiko +import pytest + +from app.core.exceptions import CustomException +from app.modules.ops.ssh import service as service_module +from app.modules.ops.ssh.schema import ( + BatchExecResultSchema, + SshHostCreateSchema, + SshHostOutSchema, + SshHostUpdateSchema, + SshSnippetCreateSchema, +) +from app.modules.ops.ssh.service import SshHostService, SshSnippetService, _load_private_key +from app.utils.crypto_util import CryptoUtil + + +class _FakeCrud: + def __init__(self, rows: list[SimpleNamespace] | None = None) -> None: + self.rows = rows or [] + self.created: list[dict[str, Any]] = [] + self.updated: list[dict[str, Any]] = [] + self.deleted: list[int] = [] + + async def get(self, **kwargs: Any) -> SimpleNamespace | None: + name = kwargs.get("name") + return next((row for row in self.rows if row.name == name), None) + + async def get_or_404(self, id: int, msg: str) -> SimpleNamespace: + for row in self.rows: + if row.id == id: + return row + raise CustomException(msg=msg) + + async def get_list(self) -> list[SimpleNamespace]: + return self.rows + + async def create(self, data: dict[str, Any]) -> SimpleNamespace: + self.created.append(data) + return _row_ns(**data) + + async def update(self, id: int, data: dict[str, Any]) -> SimpleNamespace: + self.updated.append(data) + return _row_ns(id=id, **data) + + async def delete(self, ids: list[int]) -> None: + self.deleted.extend(ids) + + +def _host_service(crud: _FakeCrud) -> SshHostService: + service = SshHostService.__new__(SshHostService) + setattr(service, "auth", SimpleNamespace(user=SimpleNamespace(id=1, username="admin"))) + setattr(service, "db", None) + setattr(service, "_crud", lambda: crud) + return service + + +def _row_ns(**overrides: Any) -> SimpleNamespace: + """构造带全部凭据字段的行（避免 _to_out 访问缺失字段）。""" + base: dict[str, Any] = { + "id": 1, + "name": "", + "host": "", + "port": 22, + "username": "", + "auth_type": "password", + "password": None, + "private_key": None, + "passphrase": None, + "status": 0, + "description": "", + } + base.update(overrides) + return SimpleNamespace(**base) + + +def _host_row(**overrides: Any) -> SimpleNamespace: + data: dict[str, Any] = { + "id": 1, + "name": "web-1", + "host": "127.0.0.1", + "port": 22, + "username": "root", + "auth_type": "password", + "password": CryptoUtil.encrypt("secret"), + "private_key": None, + "passphrase": None, + "status": 0, + "description": "", + } + data.update(overrides) + return SimpleNamespace(**data) + + +def test_host_out_schema_has_no_credentials() -> None: + fields = set(SshHostOutSchema.model_fields) + assert "password" not in fields + assert "private_key" not in fields + assert "passphrase" not in fields + + +async def test_create_password_host_encrypts_password() -> None: + crud = _FakeCrud() + service = _host_service(crud) + data = SshHostCreateSchema(name="web-1", host="10.0.0.1", port=22, username="root", auth_type="password", 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 out.has_password is True and out.has_private_key is False + + +async def test_create_password_host_requires_password() -> None: + service = _host_service(_FakeCrud()) + data = SshHostCreateSchema(name="w", host="h", username="root", auth_type="password") + with pytest.raises(CustomException, match="必须提供密码"): + await service.create(data) + + +async def test_create_key_host_requires_private_key() -> None: + service = _host_service(_FakeCrud()) + data = SshHostCreateSchema(name="w", host="h", username="root", auth_type="key") + with pytest.raises(CustomException, match="必须提供私钥"): + await service.create(data) + + +async def test_update_without_credentials_keeps_them() -> None: + crud = _FakeCrud([_host_row()]) + service = _host_service(crud) + data = SshHostUpdateSchema(name="web-1", host="10.0.0.2", port=22, username="root", auth_type="password") + await service.update(1, data) + assert "password" not in crud.updated[0] and "private_key" not in crud.updated[0] + + +async def test_duplicate_name_rejected() -> None: + crud = _FakeCrud([_host_row()]) + service = _host_service(crud) + data = SshHostCreateSchema(name="web-1", host="h", username="u", password="p") + with pytest.raises(CustomException, match="名称已存在"): + await service.create(data) + + +def test_load_private_key_parses_generated_rsa() -> None: + key = paramiko.RSAKey.generate(2048) + buffer = io.StringIO() + key.write_private_key(buffer) + parsed = _load_private_key(buffer.getvalue(), None) + assert parsed is not None + + +def test_load_private_key_rejects_garbage() -> None: + with pytest.raises(CustomException, match="无法解析私钥"): + _load_private_key("not-a-key", None) + + +async def test_test_connection_uses_injected_connect(monkeypatch: pytest.MonkeyPatch) -> None: + calls: list[dict[str, Any]] = [] + + class _Client: + def close(self) -> None: + calls.append({"closed": True}) + + def _fake_connect(**kwargs: Any) -> _Client: + calls.append(kwargs) + return _Client() + + monkeypatch.setattr(service_module, "connect_client", _fake_connect) + service = _host_service(_FakeCrud()) + from app.modules.ops.ssh.schema import SshHostTestSchema + + ok = await service.test_connection( + SshHostTestSchema(host="h", port=22, username="u", auth_type="password", password="p") + ) + assert ok is True + assert calls and calls[0]["password"] == "p" and calls[-1] == {"closed": True} + + +async def test_test_connection_failure_is_wrapped(monkeypatch: pytest.MonkeyPatch) -> None: + def _boom(**kwargs: Any) -> Any: + raise OSError("connection refused") + + monkeypatch.setattr(service_module, "connect_client", _boom) + service = _host_service(_FakeCrud()) + from app.modules.ops.ssh.schema import SshHostTestSchema + + with pytest.raises(CustomException, match="连接失败"): + await service.test_connection( + SshHostTestSchema(host="h", port=22, username="u", auth_type="password", password="p") + ) + + +async def test_batch_execute_reports_missing_and_disabled_hosts(monkeypatch: pytest.MonkeyPatch) -> None: + crud = _FakeCrud([_host_row(id=1), _host_row(id=2, name="disabled", status=1)]) + service = _host_service(crud) + + def _fake_exec(row: SimpleNamespace, command: str, timeout: int) -> BatchExecResultSchema: + return BatchExecResultSchema(host_id=row.id, host_name=row.name, ok=True, exit_status=0, stdout="ok") + + monkeypatch.setattr(service, "_exec_on_host", _fake_exec) + results = await service.batch_execute([1, 2, 999], "uptime", 5) + by_id = {item.host_id: item for item in results} + assert by_id[1].ok is True and by_id[1].stdout == "ok" + assert by_id[2].error == "主机已停用" + assert by_id[999].error == "主机不存在" + + +async def test_snippet_create_returns_validated_row() -> None: + crud = _FakeCrud() + service = SshSnippetService.__new__(SshSnippetService) + setattr(service, "auth", SimpleNamespace(user=SimpleNamespace(id=1, username="admin"))) + setattr(service, "db", None) + setattr(service, "_crud", lambda: crud) + row = await service.create(SshSnippetCreateSchema(name="disk", command="df -h")) + assert row.name == "disk" and row.command == "df -h" diff --git a/backend/tests/test_ops_ssh_terminal.py b/backend/tests/test_ops_ssh_terminal.py new file mode 100644 index 000000000..263bc643b --- /dev/null +++ b/backend/tests/test_ops_ssh_terminal.py @@ -0,0 +1,151 @@ +"""SSH 终端会话管理测试（假 paramiko 通道，不联网）。""" + +import queue +import threading +import time +from typing import Any + +import pytest + +from app.core.exceptions import CustomException +from app.modules.ops.ssh.terminal import ( + MAX_SESSIONS_PER_USER, + TerminalManager, +) + + +class _FakeChannel: + """阻塞式假通道：recv 最多等 5 秒（close 时立即返回 b""）。""" + + def __init__(self) -> None: + self.sent: list[bytes] = [] + self.size: tuple[int, int] | None = None + self.closed = False + self._event = threading.Event() + + def settimeout(self, timeout: float) -> None: + return None + + def recv(self, size: int) -> bytes: + self._event.wait(5) + return b"" + + def sendall(self, data: bytes) -> None: + self.sent.append(data) + + def resize_pty(self, width: int, height: int) -> None: + self.size = (width, height) + + def close(self) -> None: + self.closed = True + self._event.set() + + def recv_exit_status(self) -> int: + return 0 + + +class _FakeClient: + def __init__(self) -> None: + self.closed = False + + def close(self) -> None: + self.closed = True + + +def _open(user_id: int = 1) -> Any: + channel = _FakeChannel() + client = _FakeClient() + session = TerminalManager.open( + user_id=user_id, + host_id=1, + host_name="web-1", + client=client, # type: ignore[arg-type] + channel=channel, # type: ignore[arg-type] + cols=120, + rows=30, + ) + return session, channel, client + + +@pytest.fixture(autouse=True) +def _cleanup() -> Any: + yield + for session_id in list(TerminalManager._sessions): + TerminalManager.close(session_id) + + +async def test_open_poll_write_resize_close() -> None: + session, channel, client = _open() + assert session.session_id + # 模拟读取线程塞入的输出 + session.output.put(b"hello ") + session.output.put(b"world") + data, closed, _ = TerminalManager.poll(session.session_id, 1, timeout=0.3) + assert data == b"hello world" and closed is False + + TerminalManager.write(session.session_id, 1, b"ls\r") + assert channel.sent == [b"ls\r"] + + TerminalManager.resize(session.session_id, 1, 100, 40) + assert channel.size == (100, 40) + + TerminalManager.close(session.session_id, 1) + assert channel.closed is True and client.closed is True + + +async def test_poll_after_close_raises() -> None: + session, _, _ = _open() + TerminalManager.close(session.session_id, 1) + with pytest.raises(CustomException): + TerminalManager.poll(session.session_id, 1, timeout=0) + + +async def test_other_user_cannot_access_session() -> None: + session, _, _ = _open(user_id=1) + with pytest.raises(CustomException): + TerminalManager.poll(session.session_id, 2, timeout=0) + with pytest.raises(CustomException): + TerminalManager.write(session.session_id, 2, b"x") + + +async def test_per_user_session_limit() -> None: + for _ in range(MAX_SESSIONS_PER_USER): + _open(user_id=7) + with pytest.raises(CustomException, match="单用户最多"): + _open(user_id=7) + + +async def test_resync_drains_buffered_output() -> None: + session, _, _ = _open() + session.output.put(b"stale") + TerminalManager.poll(session.session_id, 1, timeout=0) # 模拟 resync + data, _, _ = TerminalManager.poll(session.session_id, 1, timeout=0) + assert data == b"" + + +async def test_reader_thread_marks_closed_on_channel_eof() -> None: + """通道 EOF（recv 返回 b""）后读取线程应把会话标记为 closed。""" + channel = _FakeChannel() + channel._event.set() # 让 recv 立刻返回 b"" + TerminalManager.open( + user_id=9, + host_id=1, + host_name="h", + client=_FakeClient(), # type: ignore[arg-type] + channel=channel, # type: ignore[arg-type] + cols=80, + rows=24, + ) + deadline = time.monotonic() + 2 + while TerminalManager.count() and time.monotonic() < deadline: + session = next(iter(TerminalManager._sessions.values())) + if session.closed: + break + time.sleep(0.02) + assert session.closed is True or TerminalManager.count() == 0 + + +def test_queue_type_is_thread_safe() -> None: + q: queue.Queue[bytes] = queue.Queue() + q.put(b"a") + assert q.get_nowait() == b"a" diff --git a/frontend/web/src/api/module_ops/ssh.ts b/frontend/web/src/api/module_ops/ssh.ts new file mode 100644 index 000000000..ec47f6bb1 --- /dev/null +++ b/frontend/web/src/api/module_ops/ssh.ts @@ -0,0 +1,173 @@ +import { request } from "@utils"; + +const API_PATH = "/ops/ssh"; + +export interface SshHost { + id: number; + name: string; + host: string; + port: number; + username: string; + auth_type: "password" | "key"; + status: number; + description?: string | null; + has_password: boolean; + has_private_key: boolean; +} + +export interface SshHostForm { + name: string; + host: string; + port: number; + username: string; + auth_type: "password" | "key"; + password?: string; + private_key?: string; + passphrase?: string; + status?: number; + description?: string | null; +} + +export interface SshSnippet { + id: number; + name: string; + command: string; + description?: string | null; +} + +export interface TerminalSession { + session_id: string; + host_id: number; + host_name: string; + cols: number; + rows: number; + created_at: string; +} + +export interface TerminalPoll { + data: string; + closed: boolean; + exit_status: number | null; +} + +export interface BatchExecResult { + host_id: number; + host_name: string; + ok: boolean; + exit_status: number | null; + stdout: string; + stderr: string; + error: string; + elapsed_ms: number; +} + +const SshAPI = { + listHosts() { + return request>({ url: `${API_PATH}/host/list`, method: "get" }); + }, + createHost(body: SshHostForm) { + return request>({ + url: `${API_PATH}/host/create`, + method: "post", + data: body, + }); + }, + updateHost(id: number, body: SshHostForm) { + return request>({ + url: `${API_PATH}/host/update/${id}`, + method: "put", + data: body, + }); + }, + deleteHosts(ids: number[]) { + return request>({ + url: `${API_PATH}/host/delete`, + method: "delete", + data: ids, + }); + }, + testHostConfig(body: SshHostForm) { + return request>({ + url: `${API_PATH}/host/test`, + method: "post", + data: body, + }); + }, + testHost(id: number) { + return request>({ url: `${API_PATH}/host/test/${id}`, method: "post" }); + }, + listSnippets() { + return request>({ url: `${API_PATH}/snippet/list`, method: "get" }); + }, + createSnippet(body: { name: string; command: string; description?: string | null }) { + return request>({ + url: `${API_PATH}/snippet/create`, + method: "post", + data: body, + }); + }, + updateSnippet(id: number, body: { name: string; command: string; description?: string | null }) { + return request>({ + url: `${API_PATH}/snippet/update/${id}`, + method: "put", + data: body, + }); + }, + deleteSnippets(ids: number[]) { + return request>({ + url: `${API_PATH}/snippet/delete`, + method: "delete", + data: ids, + }); + }, + openTerminal(body: { host_id: number; cols: number; rows: number }) { + return request>({ + url: `${API_PATH}/terminal/open`, + method: "post", + data: body, + }); + }, + openLog(body: { host_id: number; path: string; lines: number }) { + return request>({ + url: `${API_PATH}/log/open`, + method: "post", + data: body, + }); + }, + pollTerminal(sessionId: string, timeout = 1.5) { + return request>({ + url: `${API_PATH}/terminal/${sessionId}/poll`, + method: "get", + params: { timeout }, + }); + }, + inputTerminal(sessionId: string, data: string) { + return request>({ + url: `${API_PATH}/terminal/${sessionId}/input`, + method: "post", + data: { data }, + }); + }, + resizeTerminal(sessionId: string, cols: number, rows: number) { + return request>({ + url: `${API_PATH}/terminal/${sessionId}/resize`, + method: "post", + data: { cols, rows }, + }); + }, + closeTerminal(sessionId: string) { + return request>({ + url: `${API_PATH}/terminal/${sessionId}`, + method: "delete", + }); + }, + batchExec(body: { host_ids: number[]; command: string; timeout?: number }) { + return request>({ + url: `${API_PATH}/batch-exec`, + method: "post", + data: body, + }); + }, +}; + +export default SshAPI; diff --git a/frontend/web/src/router/routes.ts b/frontend/web/src/router/routes.ts index 6307d21a8..80e71635b 100644 --- a/frontend/web/src/router/routes.ts +++ b/frontend/web/src/router/routes.ts @@ -99,6 +99,8 @@ export class IframeRouteManager { /** 首页菜单配置（图标、缓存、固定标签） */ // SQL 控制台页面懒加载：既避免静态路由测试在导入期拉全页面依赖，也减小首屏体积 const DbConsoleView = defineAsyncComponent(() => import("@views/module_dbconsole/index.vue")); +// SSH 运维页面懒加载（同 SQL 控制台） +const SshOpsView = defineAsyncComponent(() => import("@views/module_ops/ssh/index.vue")); export const HOME_MENU_META: RouteMeta = { title: "menus.home.title", @@ -248,6 +250,13 @@ export const staticRoutes: AppRouteRecordRaw[] = [ component: DashboardHome, meta: HOME_MENU_META, }, + { + // SSH 运维：隐藏菜单，直接访问 /ops/ssh + path: "ops/ssh", + name: "SshOps", + component: SshOpsView, + meta: { hidden: true, title: "SSH 运维" }, + }, { // SQL 控制台（仅外部连接）：隐藏菜单，直接访问 /dbconsole path: "dbconsole", diff --git a/frontend/web/src/views/module_ops/ssh/index.vue b/frontend/web/src/views/module_ops/ssh/index.vue new file mode 100644 index 000000000..81d03cf5d --- /dev/null +++ b/frontend/web/src/views/module_ops/ssh/index.vue @@ -0,0 +1,559 @@ + + + + + +