From fc22de7f4d1c50753485ae9356ff6169ddcd25f2 Mon Sep 17 00:00:00 2001 From: Simon Date: Thu, 2 Jul 2026 19:06:12 +0800 Subject: [PATCH] =?UTF-8?q?A=E7=BB=84=E8=AE=A4=E8=AF=81=E5=8A=A0=E5=9B=BA:?= =?UTF-8?q?=20P0=E5=85=9C=E5=BA=95+Token=E5=88=B7=E6=96=B0+=E7=8E=AF?= =?UTF-8?q?=E5=A2=83=E6=A3=80=E6=B5=8B+OTP+RBRAC=E8=90=BD=E5=9C=B0+P1?= =?UTF-8?q?=E6=97=A5=E5=BF=97=E5=AE=A1=E8=AE=A1+Token=E6=92=A4=E9=94=80=20?= =?UTF-8?q?-=20=E5=85=A8=E5=B1=80=E4=B8=80=E8=87=B4=E6=80=A7=E5=AE=A1?= =?UTF-8?q?=E6=9F=A5=E9=80=9A=E8=BF=87?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- backend/app/api/admin_api.py | 68 +++ backend/app/api/auth_qrcode.py | 102 +++- backend/app/api/auth_wecom_sso.py | 431 ++++++++++++--- backend/app/api/conversations.py | 75 +++ backend/app/api/messages.py | 154 +++++- backend/app/dependencies.py | 65 ++- backend/app/services/rbac_service.py | 1 + frontend-agent/src/api/index.ts | 148 ++++- frontend-portal/src/router/index.ts | 18 + frontend-portal/src/views/ErrorPage.vue | 156 ++++++ jumpserver-webcli/requirements.txt | 3 + ops-tools/jms_ops.py | 698 ++++++++++++++++++++++++ 12 files changed, 1813 insertions(+), 106 deletions(-) create mode 100644 frontend-portal/src/views/ErrorPage.vue create mode 100644 jumpserver-webcli/requirements.txt create mode 100644 ops-tools/jms_ops.py diff --git a/backend/app/api/admin_api.py b/backend/app/api/admin_api.py index 9f8e64c..1dc4407 100644 --- a/backend/app/api/admin_api.py +++ b/backend/app/api/admin_api.py @@ -1014,3 +1014,71 @@ async def ragflow_retrieval( return success_response(data={"error": e.message, "error_code": "config_missing"}) except RagflowError as e: return success_response(data={"error": e.message, "error_code": "api_error"}) + + +# ---------- POST /api/admin/users/{employee_id}/revoke-token ---------- +@router.post("/users/{employee_id}/revoke-token") +async def revoke_user_token( + employee_id: str, + admin: Agent = Depends(require_admin), + db: AsyncSession = Depends(get_db), +): + """强制撤销指定用户的登录Token（管理员操作）。 + + 清除该用户在 Redis 中的所有 Token，使其被迫下线。 + 同时记录审计日志。 + + Args: + employee_id: 要撤销 Token 的用户 ID（企微 UserID） + + Returns: + 撤销结果 + """ + from app.dependencies import get_redis + from app.services.audit_log_service import record_audit_log + + redis_client = await get_redis() + + # 搜索可能的 Token key 模式 + # 1. user:token:* - 统一格式 + # 2. agent:token:* - 坐席端 + # 3. employee:token:* - 员工端 + revoked_count = 0 + patterns = ["user:token:*", "agent:token:*", "employee:token:*"] + + for pattern in patterns: + cursor = 0 + while True: + cursor, keys = await redis_client.scan(cursor, match=pattern, count=100) + for key in keys: + token_data = await redis_client.get(key) + if token_data: + try: + import json + data = json.loads(token_data) + if data.get("employee_id") == employee_id: + await redis_client.delete(key) + revoked_count += 1 + logger.info(f"撤销 Token: key={key}, employee_id={employee_id}") + except (json.JSONDecodeError, Exception): + pass + if cursor == 0: + break + + # 记录审计日志 + await record_audit_log( + db=db, + employee_id=admin.employee_id, + action="revoke_token", + resource="user", + resource_id=employee_id, + details={"revoked_count": revoked_count, "operator": admin.employee_id}, + result="success", + ) + await db.commit() + + return success_response(data={ + "employee_id": employee_id, + "revoked_count": revoked_count, + "message": f"已撤销 {revoked_count} 个 Token" if revoked_count > 0 else "未找到该用户的有效 Token", + }) diff --git a/backend/app/api/auth_qrcode.py b/backend/app/api/auth_qrcode.py index b0fe538..771056f 100644 --- a/backend/app/api/auth_qrcode.py +++ b/backend/app/api/auth_qrcode.py @@ -39,9 +39,11 @@ import logging from typing import Optional import redis.asyncio as aioredis -from fastapi import APIRouter, Depends, Path +from fastapi import APIRouter, Depends, Path, Query +from sqlalchemy.ext.asyncio import AsyncSession from app.config import settings +from app.database import get_db from app.dependencies import dep_redis, get_current_user, UserInfo from app.schemas.qrcode import ( QrcodeConfirmRequest, @@ -139,30 +141,85 @@ async def poll_qrcode( # -------------------------------------------------------------------------- -# POST /api/auth_qrcode/scan — 企微 OAuth code 回调 +# GET|POST /api/auth_qrcode/scan — 企微 OAuth code 回调 # -------------------------------------------------------------------------- -@router.post("/scan", response_model=None) +@router.api_route("/scan", methods=["GET", "POST"], response_model=None) async def scan_qrcode( - body: QrcodeScanRequest, + body: Optional[QrcodeScanRequest] = None, + ticket: Optional[str] = Query(None, description="扫码登录票据(兼容旧参数名)"), + state: Optional[str] = Query(None, description="扫码登录票据(企微 OAuth state 标准参数名)"), + code: Optional[str] = Query(None, description="企微 OAuth 授权码"), redis_client: aioredis.Redis = Depends(dep_redis), ): """处理企微 OAuth2 扫码回调。 + 企微 OAuth2 标准回调走 **GET** 带 query 参数 `?code=xxx&state=`, + 本端点同时支持 GET 和 POST(POST 兼容内部调用 / 旧前端代码)。 + + GET 模式 (企微 OAuth2 标准回调): + - ticket ← query.state + - code ← query.code + - 自动 302 跳转到 /itdesk/ 或 /itadmin/ 或 /itagent/(按角色) + POST 模式 (内部调用): + - ticket ← body.ticket + - code ← body.code + 无需鉴权(此端点被企微服务器回调,带 code + ticket)。 用 code 换取企微 userid,然后写 Redis scan:{ticket} 等待 confirm 端点。 dev 模式: code 形如 "dev:dev-user-001",跳过企微 API 调用。 - - Args: - body: 包含 ticket 和 code - - Returns: - Dict: 统一响应格式,data 字段是 QrcodeScanResponse """ try: - service = _get_qrcode_service(redis_client) - result = await service.process_scan(ticket=body.ticket, code=body.code) + # 1. 解析参数:POST 用 body,GET 用 query + if body is not None: + final_ticket = body.ticket + final_code = body.code + else: + # 优先用 state(企微 OAuth 标准),回退到 ticket(兼容旧调用) + final_ticket = state or ticket + final_code = code + if not final_ticket or not final_code: + logger.warning(f"扫码参数缺失: ticket={final_ticket!r}, code={final_code!r}") + raise AppException(1000, "缺少 ticket 或 code 参数") + + service = _get_qrcode_service(redis_client) + result = await service.process_scan(ticket=final_ticket, code=final_code) + + # GET 请求(企微 OAuth 回调)→ 重定向到前端选择页 + # 因为企微 OAuth 流程不在这个端点完成最终登录,只标记 scanned, + # 等用户在坐席端点 confirm 后才能拿到 token。 + # 但企微 WebView 期望看到跳转后的页面,所以这里给个提示页。 + from fastapi.responses import HTMLResponse + if code is not None and ticket is not None and body is None: + # GET 模式:渲染一个 "扫码成功" 的 HTML 提示页 + 引导用户到登录页 + html = f""" + + + + +扫码成功 + + + +
+
✅
+

扫码成功

+

请在刚才打开登录页的浏览器中

+

点击 「确认登录」 按钮完成登录

+

本页可关闭

+
+ +""" + return HTMLResponse(content=html, status_code=200) + + # POST 模式:返回 JSON return success_response(data={ "success": result["success"], "message": result["message"], @@ -173,7 +230,7 @@ async def scan_qrcode( logger.warning(f"扫码业务错误: {ve}") raise AppException(1003, str(ve)) except Exception as e: - logger.error(f"扫码处理异常: ticket={body.ticket[:8]}..., error={e}", exc_info=True) + logger.error(f"扫码处理异常: error={e}", exc_info=True) raise AppException(1005, f"扫码处理失败: {str(e)}") @@ -185,6 +242,7 @@ async def confirm_qrcode( body: QrcodeConfirmRequest, current_user: UserInfo = Depends(get_current_user), redis_client: aioredis.Redis = Depends(dep_redis), + db: AsyncSession = Depends(get_db), ): """处理当前已登录坐席的扫码确认授权。 @@ -213,6 +271,24 @@ async def confirm_qrcode( otp_code=body.otp_code, ) + # 记录扫码登录日志（成功） + from app.services.audit_log_service import record_audit_log + await record_audit_log( + db=db, + employee_id=result["employee_id"], + action="qrcode_login", + resource="auth", + resource_id=result["employee_id"], + details={ + "name": result["name"], + "roles": result["roles"], + "confirmed_by": current_user.employee_id, + "login_method": "qrcode_confirm", + }, + result="success", + ) + await db.commit() + return success_response(data={ "token": result["token"], "employee_id": result["employee_id"], diff --git a/backend/app/api/auth_wecom_sso.py b/backend/app/api/auth_wecom_sso.py index 8ab2671..bc8b579 100644 --- a/backend/app/api/auth_wecom_sso.py +++ b/backend/app/api/auth_wecom_sso.py @@ -20,6 +20,7 @@ # ============================================================================= import logging +import re import secrets import urllib.parse from datetime import datetime, timedelta @@ -35,6 +36,7 @@ from app.database import get_db from app.models.role import Role from app.models.user_role import UserRole from app.services.wecom_service import WecomService +from app.services.audit_log_service import record_audit_log from app.utils.response import AppException from app.dependencies import get_redis @@ -46,6 +48,15 @@ router = APIRouter(prefix="/auth_wecom", tags=["企微 SSO"]) OAUTH_STATE_TTL = 300 # SSO token 长度 SSO_TOKEN_BYTES = 32 +# Token TTL 常量（8小时） +TOKEN_TTL_SECONDS = 8 * 60 * 60 # 8小时 +# 企微 API 超时设置（秒） +WECOM_API_TIMEOUT = 10 + +# -------------------------------------------------------------------------- +# 企微环境检测（使用统一工具模块） +# -------------------------------------------------------------------------- +from app.utils.wecom_auth import require_wecom_ua as _require_wework_ua def _sso_enabled() -> bool: @@ -101,6 +112,9 @@ async def sso_init( Args: next: 登录成功后跳转路径,如 /itdesk/ /itagent/ /itadmin/ """ + # 后端第二道防线：非企微环境拒绝授权 + _require_wework_ua(request) + if not _sso_enabled(): raise AppException(1001, "企微 SSO 未启用, 请用扫码登录") @@ -116,103 +130,229 @@ async def sso_init( str(state_payload).encode("utf-8"), ) - # 2. 拼企微 OAuth URL + # 2. 拼企微 OAuth URL（回调URL中包含next参数，用于state失效时仍能知道目标路径） callback_url = _get_oauth_callback_url(request) - oauth_url = _build_oauth_url(state, callback_url) + # 在回调URL中添加next参数 + separator = "&" if "?" in callback_url else "?" + callback_url_with_next = f"{callback_url}{separator}next={urllib.parse.quote(next)}" + oauth_url = _build_oauth_url(state, callback_url_with_next) logger.info(f"SSO init: state={state[:8]}..., next={next}") return RedirectResponse(url=oauth_url, status_code=302) +def _get_error_redirect_url(error_code: str, error_msg: str, next_path: str = "/itdesk/") -> str: + """生成 OAuth 错误重定向 URL。 + + 异常时重定向到前端错误页面（ErrorPage），而不是返回 JSON 错误。 + ErrorPage 读取 code 和 message 参数显示友好错误提示。 + + Args: + error_code: 错误码 + error_msg: 错误信息 + next_path: 原始请求的目标路径（保留但不再用于决定重定向） + """ + import os + base = getattr(settings, "wecom_sso_callback_base", None) + if not base: + base = os.getenv("WECOM_SSO_CALLBACK_BASE", "https://itsupport.servyou.com.cn") + + # 重定向到 Portal 的 ErrorPage，带错误参数 + # ErrorPage 期望格式：?code=xxx&message=yyy + return f"{base.rstrip('/')}/itportal/error?code={error_code}&message={urllib.parse.quote(error_msg)}" + + @router.get("/sso/callback") async def sso_callback( - code: str = Query(..., description="企微 OAuth2 授权 code"), - state: str = Query(..., description="防 CSRF state"), + request: Request, + code: Optional[str] = Query(None, description="企微 OAuth2 授权 code"), + state: Optional[str] = Query(None, description="防 CSRF state"), + errcode: Optional[int] = Query(None, description="企微 OAuth 错误码"), + errmsg: Optional[str] = Query(None, description="企微 OAuth 错误信息"), + next: Optional[str] = Query(None, description="原始请求的目标路径（可选，用于错误时重定向）"), redis_client = Depends(get_redis), db: AsyncSession = Depends(get_db), ): - """企微 OAuth 回调: 用 code 换 userid → 查 role → 生成 token → 跳 next。""" - # 1. 校验 state(防 CSRF) - state_key = f"wecom_sso:state:{state}" - state_raw = await redis_client.get(state_key) - if not state_raw: - raise AppException(1002, "SSO state 已过期或无效, 请重新进入") + """企微 OAuth 回调: 用 code 换 userid → 查 role → 生成 token → 跳 next。 - # 删除 state(一次性) - await redis_client.delete(state_key) + 异常时重定向到前端错误页面，避免白屏。 + 所有未处理的异常都会记录详细日志（包含 traceback）。 - import ast - state_data = ast.literal_eval(state_raw.decode("utf-8")) - next_path = state_data.get("next", "/itdesk/") + Args: + next: 原始请求的目标路径，用于错误重定向。如果 state 验证失败，使用此参数决定重定向位置。 + """ + import traceback + + # 默认 next 路径 + next_path = next or "/itdesk/" - # 2. 用 code 换 userid - wecom = WecomService(redis_client) try: - oauth_info = await wecom.get_oauth_user_info(code) - user_id = oauth_info.get("userid", "") - if not user_id: - raise AppException(1003, "企微 OAuth 返回 userid 为空") + # 后端第二道防线：非企微环境拒绝回调 + _require_wework_ua(request) - user_info = await wecom.get_user_info(user_id) - name = user_info.get("name", user_id) - except Exception as e: - logger.error(f"SSO callback 调企微 API 失败: code={code[:8]}..., error={e}") - raise AppException(1004, f"企微身份识别失败: {str(e)}") - finally: + # 0. 处理企微返回的错误（用户拒绝授权等） + if errcode is not None: + msg = errmsg or "用户取消授权或授权失败" + logger.warning(f"SSO callback 企微返回错误: errcode={errcode}, errmsg={errmsg}") + return RedirectResponse(url=_get_error_redirect_url(f"wecom_{errcode}", msg, next_path), status_code=302) + + # 1. 校验必要参数 + if not code or not state: + logger.warning(f"SSO callback 缺少必要参数: code={bool(code)}, state={bool(state)}") + return RedirectResponse(url=_get_error_redirect_url("missing_params", "授权参数不完整，请重试", next_path), status_code=302) + + # 2. 校验 state(防 CSRF) + state_key = f"wecom_sso:state:{state}" try: - await wecom.close() - except Exception: - pass + state_raw = await redis_client.get(state_key) + except Exception as e: + logger.error(f"SSO callback Redis 获取 state 失败: {e}") + return RedirectResponse(url=_get_error_redirect_url("redis_error", "服务暂不可用，请稍后重试", next_path), status_code=302) - # 3. 查 role (user/agent/admin) - role_stmt = ( - select(Role) - .join(UserRole, Role.id == UserRole.role_id) - .where(UserRole.employee_id == user_id) - ) - role_result = await db.execute(role_stmt) - roles = role_result.scalars().all() + if not state_raw: + logger.warning(f"SSO callback state 过期: state={state[:8]}...") + return RedirectResponse(url=_get_error_redirect_url("state_expired", "授权已过期，请重新进入", next_path), status_code=302) - if not roles: - # 没有绑定角色: 跳"无权限"页 - logger.warning(f"SSO: user_id={user_id} 没绑定任何角色") - return RedirectResponse(url=f"/itdesk/no-role?user_id={user_id}", status_code=302) + # 删除 state(一次性) + try: + await redis_client.delete(state_key) + except Exception as e: + logger.warning(f"SSO callback 删除 state 失败: {e}") # 不阻塞流程 - # 4. 选最高权限角色 (admin > agent > user) - role_priority = {"admin": 3, "agent": 2, "user": 1} - best_role = max(roles, key=lambda r: role_priority.get(r.name, 0)) - role_name = best_role.name + # 解析 state 数据（添加异常处理） + import ast + import json + try: + state_data = json.loads(state_raw.decode("utf-8")) + except (json.JSONDecodeError, AttributeError) as e: + # 兼容旧格式（使用 ast.literal_eval） + try: + state_data = ast.literal_eval(state_raw.decode("utf-8")) + except (ValueError, SyntaxError) as e2: + logger.error(f"SSO callback state 解析失败: {e2}") + return RedirectResponse(url=_get_error_redirect_url("state_invalid", "授权信息无效，请重新进入", next_path), status_code=302) - # 5. 生成 SSO token(随机 + Redis 存 8 小时) - sso_token = secrets.token_urlsafe(SSO_TOKEN_BYTES) - sso_payload = { - "user_id": user_id, - "name": name, - "role": role_name, - "created_at": datetime.now().isoformat(), - } - import json - await redis_client.setex( - f"wecom_sso:token:{sso_token}", - 8 * 3600, # 8 小时 - json.dumps(sso_payload, ensure_ascii=False).encode("utf-8"), - ) + # 优先使用 state 中存储的 next，fallback 到 URL 参数 + next_path = state_data.get("next", next_path) - # 6. 跳转到 next + token - separator = "&" if "?" in next_path else "?" - redirect_url = f"{next_path}{separator}sso_token={sso_token}" + # 3. 用 code 换 userid + wecom = WecomService(redis_client) + try: + oauth_info = await wecom.get_oauth_user_info(code) + user_id = oauth_info.get("userid", "") + if not user_id: + logger.warning("SSO callback 企微返回 userid 为空") + return RedirectResponse(url=_get_error_redirect_url("empty_userid", "无法获取您的企业微信身份，请重试", next_path), status_code=302) - logger.info(f"SSO 成功: user_id={user_id}, role={role_name}, next={next_path}") - return RedirectResponse(url=redirect_url, status_code=302) + user_info = await wecom.get_user_info(user_id) + name = user_info.get("name", user_id) + except Exception as e: + logger.error(f"SSO callback 调企微 API 失败: code={code[:8]}..., error={e}") + return RedirectResponse(url=_get_error_redirect_url("api_failed", f"企业微信服务异常: {str(e)}"), status_code=302) + finally: + try: + await wecom.close() + except Exception: + pass + + # 3. 查 role (user/agent/admin) + try: + role_stmt = ( + select(Role) + .join(UserRole, Role.id == UserRole.role_id) + .where(UserRole.employee_id == user_id) + ) + role_result = await db.execute(role_stmt) + roles = role_result.scalars().all() + except Exception as e: + logger.error(f"SSO callback 查询角色失败: {e}") + return RedirectResponse(url=_get_error_redirect_url("db_error", "服务暂不可用，请稍后重试", next_path), status_code=302) + + if not roles: + # 没有绑定角色: 跳"无权限"页 + logger.warning(f"SSO: user_id={user_id} 没绑定任何角色") + return RedirectResponse(url=f"/itdesk/no-role?user_id={user_id}", status_code=302) + + # 4. 选最高权限角色 (admin > agent > user) + role_priority = {"admin": 3, "agent": 2, "user": 1} + best_role = max(roles, key=lambda r: role_priority.get(r.name, 0)) + role_name = best_role.name + + # 5. 生成 SSO token(随机 + Redis 存 8 小时) + sso_token = secrets.token_urlsafe(SSO_TOKEN_BYTES) + sso_payload = { + "user_id": user_id, + "name": name, + "role": role_name, + "created_at": datetime.now().isoformat(), + } + import json + try: + await redis_client.setex( + f"wecom_sso:token:{sso_token}", + TOKEN_TTL_SECONDS, + json.dumps(sso_payload, ensure_ascii=False).encode("utf-8"), + ) + except Exception as e: + logger.error(f"SSO callback 存储 token 失败: {e}") + return RedirectResponse(url=_get_error_redirect_url("redis_error", "服务暂不可用，请稍后重试", next_path), status_code=302) + + # 6. 记录登录日志 + try: + await record_audit_log( + db=db, + employee_id=user_id, + action="sso_login", + resource="auth", + resource_id=user_id, + details={"name": name, "role": role_name, "login_method": "wecom_sso"}, + result="success", + request=None, # callback 请求没有直接可用的 request 对象 + ) + await db.commit() + except Exception as e: + logger.warning(f"SSO callback 记录登录日志失败: {e}") # 不阻塞登录流程 + + # 7. 跳转到 next + token + separator = "&" if "?" in next_path else "?" + redirect_url = f"{next_path}{separator}sso_token={sso_token}" + + logger.info(f"SSO 成功: user_id={user_id}, role={role_name}, next={next_path}") + return RedirectResponse(url=redirect_url, status_code=302) + + except Exception as e: + # 捕获所有未处理的异常，记录详细日志（包含 traceback）并重定向到错误页 + error_details = { + "error": str(e), + "error_type": type(e).__name__, + "code": code[:8] + "..." if code else None, + "state": state[:8] + "..." if state else None, + "next": next_path, + } + logger.error( + f"SSO callback 未处理的异常: {error_details}\n" + f"traceback: {traceback.format_exc()}" + ) + return RedirectResponse( + url=_get_error_redirect_url("oauth_failed", "登录过程出现异常，请重试"), + status_code=302 + ) @router.get("/sso/verify") async def sso_verify( + request: Request, sso_token: str = Query(..., description="SSO token"), redis_client = Depends(get_redis), db: AsyncSession = Depends(get_db), ): - """前端用 SSO token 换用户身份(token 一次性使用,用完删除)。""" + """前端用 SSO token 换用户身份(token 一次性使用,用完删除)。 + + 后端第二道防线：非企微环境拒绝验证。 + """ + # 后端第二道防线：非企微环境拒绝验证 + _require_wework_ua(request) + import json token_raw = await redis_client.get(f"wecom_sso:token:{sso_token}") if not token_raw: @@ -226,3 +366,162 @@ async def sso_verify( "code": 0, "data": payload, } + + +@router.post("/refresh") +async def refresh_token( + token: str = Query(..., description="当前 Bearer token"), + redis_client = Depends(get_redis), +): + """刷新 Token TTL。 + + 前端在 Token 过期前 5 分钟自动调用此接口，实现静默刷新。 + 如果 Token 无效或已过期，返回 401 错误。 + + Returns: + 刷新成功：{ code: 0, data: { token: "新token", expires_in: 28800 } } + """ + import json + + # 1. 尝试统一格式 Token + token_key = f"user:token:{token}" + token_data_raw = await redis_client.get(token_key) + + if token_data_raw: + try: + user_info = json.loads(token_data_raw) + # 更新最后活跃时间 + user_info["last_active"] = datetime.now().isoformat() + + # 延长 TTL（重新设置 8 小时） + await redis_client.setex( + token_key, + TOKEN_TTL_SECONDS, + json.dumps(user_info, ensure_ascii=False), + ) + + logger.info(f"Token 刷新成功: employee_id={user_info.get('employee_id')}") + return { + "code": 0, + "data": { + "token": token, # 复用同一个 token，只延长 TTL + "expires_in": TOKEN_TTL_SECONDS, + }, + } + except json.JSONDecodeError: + pass + + # 2. 尝试旧格式 Token (employee:token) + employee_key = f"employee:token:{token}" + employee_id = await redis_client.get(employee_key) + if employee_id: + # 延长 TTL + await redis_client.expire(employee_key, TOKEN_TTL_SECONDS) + logger.info(f"Token 刷新成功(employee): employee_id={employee_id}") + return { + "code": 0, + "data": { + "token": token, + "expires_in": TOKEN_TTL_SECONDS, + }, + } + + # 3. 尝试旧格式 Token (agent:token) + agent_key = f"agent:token:{token}" + agent_id = await redis_client.get(agent_key) + if agent_id: + await redis_client.expire(agent_key, TOKEN_TTL_SECONDS) + logger.info(f"Token 刷新成功(agent): agent_id={agent_id}") + return { + "code": 0, + "data": { + "token": token, + "expires_in": TOKEN_TTL_SECONDS, + }, + } + + # Token 无效或已过期 + logger.warning(f"Token 刷新失败: token 不存在或已过期") + raise AppException(401, "Token 已过期，请重新登录") + + +# -------------------------------------------------------------------------- +# 别名路由：支持前端 /api/auth/refresh 调用（与 /api/auth_wecom/refresh 等效） +# -------------------------------------------------------------------------- +# 前端 H5/坐席/管理后台调用 /api/auth/refresh，后端响应 /api/auth_wecom/refresh +# 为兼容前端习惯，添加此别名路由 +# -------------------------------------------------------------------------- + +# 创建别名路由器（无 prefix） +alias_router = APIRouter(tags=["认证"]) + + +@alias_router.post("/auth/refresh") +async def refresh_token_alias( + token: str = Query(..., description="当前 Bearer token"), + redis_client = Depends(get_redis), +): + """Token 刷新接口别名。 + + 前端调用 /api/auth/refresh，后端实际处理逻辑与 /api/auth_wecom/refresh 相同。 + 这是为了兼容前端的调用习惯。 + + Returns: + 刷新成功：{ code: 0, data: { token: "新token", expires_in: 28800 } } + """ + import json + + # 1. 尝试统一格式 Token + token_key = f"user:token:{token}" + token_data_raw = await redis_client.get(token_key) + + if token_data_raw: + try: + user_info = json.loads(token_data_raw) + user_info["last_active"] = datetime.now().isoformat() + await redis_client.setex( + token_key, + TOKEN_TTL_SECONDS, + json.dumps(user_info, ensure_ascii=False), + ) + logger.info(f"Token 刷新成功(alias): employee_id={user_info.get('employee_id')}") + return { + "code": 0, + "data": { + "token": token, + "expires_in": TOKEN_TTL_SECONDS, + }, + } + except json.JSONDecodeError: + pass + + # 2. 尝试旧格式 Token + employee_key = f"employee:token:{token}" + employee_id = await redis_client.get(employee_key) + if employee_id: + await redis_client.expire(employee_key, TOKEN_TTL_SECONDS) + logger.info(f"Token 刷新成功(alias employee): employee_id={employee_id}") + return { + "code": 0, + "data": { + "token": token, + "expires_in": TOKEN_TTL_SECONDS, + }, + } + + # 3. 尝试 agent token + agent_key = f"agent:token:{token}" + agent_id = await redis_client.get(agent_key) + if agent_id: + await redis_client.expire(agent_key, TOKEN_TTL_SECONDS) + logger.info(f"Token 刷新成功(alias agent): agent_id={agent_id}") + return { + "code": 0, + "data": { + "token": token, + "expires_in": TOKEN_TTL_SECONDS, + }, + } + + logger.warning(f"Token 刷新失败(alias): token 不存在或已过期") + raise AppException(401, "Token 已过期，请重新登录") diff --git a/backend/app/api/conversations.py b/backend/app/api/conversations.py index c2c5206..2305022 100644 --- a/backend/app/api/conversations.py +++ b/backend/app/api/conversations.py @@ -30,6 +30,7 @@ from app.schemas.conversation import ( ConversationStatusUpdate, InviteParticipantRequest, JoinConversationRequest, + UpdateTagsRequest, ) from app.services.session_service import SessionService from app.services.wecom_service import WecomService @@ -38,6 +39,9 @@ from app.utils.response import AppException, success_response # 坐席认证依赖（从 agents.py 导入） from app.api.agents import get_current_agent +# RBAC 权限装饰器 +from app.dependencies import require_role, require_permission + logger = logging.getLogger(__name__) # 创建路由器 @@ -48,6 +52,7 @@ router = APIRouter() # GET /api/conversations — 获取坐席会话列表（全局可见） # -------------------------------------------------------------------------- @router.get("/conversations") +@require_permission("conversation", "read", "all") async def list_conversations( status: Optional[str] = Query(None, description="按状态过滤: ai_handling/queued/serving/resolved"), agent_id: Optional[str] = Query(None, description="按坐席ID过滤"), @@ -142,6 +147,7 @@ async def list_conversations( # GET /api/conversations/{id} — 获取会话详情 # -------------------------------------------------------------------------- @router.get("/conversations/{conversation_id}") +@require_permission("conversation", "read", "all") async def get_conversation( conversation_id: str, db: AsyncSession = Depends(get_db), @@ -166,6 +172,7 @@ async def get_conversation( # POST /api/conversations/{id}/assign — 坐席接单 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/assign") +@require_permission("conversation", "update", "all") async def assign_conversation( conversation_id: str, body: ConversationAssign, @@ -216,6 +223,7 @@ async def assign_conversation( # POST /api/conversations/{id}/resolve — 结单 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/resolve") +@require_permission("conversation", "update", "own") async def resolve_conversation( conversation_id: str, db: AsyncSession = Depends(get_db), @@ -259,6 +267,7 @@ async def resolve_conversation( # POST /api/conversations/{id}/pin — 置顶/取消置顶 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/pin") +@require_permission("conversation", "update", "own") async def toggle_pin( conversation_id: str, db: AsyncSession = Depends(get_db), @@ -285,6 +294,7 @@ async def toggle_pin( # POST /api/conversations/{id}/todo — 代办/取消代办 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/todo") +@require_permission("conversation", "update", "own") async def toggle_todo( conversation_id: str, db: AsyncSession = Depends(get_db), @@ -311,6 +321,7 @@ async def toggle_todo( # POST /api/conversations/{id}/transfer — 转接 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/transfer") +@require_permission("conversation", "update", "all") async def transfer_conversation( conversation_id: str, body: ConversationAssign, @@ -342,6 +353,7 @@ async def transfer_conversation( # POST /api/conversations/{id}/grab — 接手会话（抢单） # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/grab") +@require_permission("conversation", "update", "all") async def grab_conversation( conversation_id: str, db: AsyncSession = Depends(get_db), @@ -439,6 +451,7 @@ async def grab_conversation( # POST /api/conversations/{id}/invite — 摇人（邀请坐席协作） # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/invite") +@require_permission("conversation", "update", "own") async def invite_collaborator( conversation_id: str, body: ConversationInvite, @@ -484,6 +497,7 @@ async def invite_collaborator( # POST /api/conversations/{id}/leave — 退出协作 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/leave") +@require_permission("conversation", "update", "own") async def leave_collaboration( conversation_id: str, db: AsyncSession = Depends(get_db), @@ -529,6 +543,7 @@ async def leave_collaboration( # POST /api/conversations/{id}/invite-participant — 邀请员工/部门加入会话 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/invite-participant") +@require_permission("conversation", "update", "own") async def invite_participant( conversation_id: str, body: InviteParticipantRequest, @@ -587,6 +602,7 @@ async def invite_participant( # POST /api/conversations/{id}/join — 被邀请人加入会话 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/join") +@require_permission("conversation", "update", "all") async def join_conversation( conversation_id: str, body: JoinConversationRequest, @@ -622,6 +638,7 @@ async def join_conversation( # DELETE /api/conversations/{id}/participants/{user_id} — 移除参与者 # -------------------------------------------------------------------------- @router.delete("/conversations/{conversation_id}/participants/{user_id}") +@require_permission("conversation", "update", "own") async def remove_participant( conversation_id: str, user_id: str, @@ -659,6 +676,7 @@ async def remove_participant( # POST /api/conversations/{id}/leave-participant — 参与者主动退出 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/leave-participant") +@require_permission("conversation", "update", "own") async def leave_as_participant( conversation_id: str, body: JoinConversationRequest, @@ -686,3 +704,60 @@ async def leave_as_participant( response_data = ConversationResponse.model_validate(conversation).model_dump() return success_response(data=response_data) + + +# -------------------------------------------------------------------------- +# POST /api/conversations/{conversation_id}/tags — 保存会话标签 +# -------------------------------------------------------------------------- +@router.post("/conversations/{conversation_id}/tags") +@require_permission("conversation", "update", "own") +async def update_conversation_tags( + conversation_id: str, + body: UpdateTagsRequest, + db: AsyncSession = Depends(get_db), + current_agent: Agent = Depends(get_current_agent), +): + """保存会话标签。 + + 坐席可以为会话添加/更新标签，如问题分类、优先级、情绪状态等。 + 标签以 JSON 形式存储在会话的 tags 字段中。 + + Args: + conversation_id: 会话ID + body: 标签更新请求，包含 tags 字典 + current_agent: 当前坐席（通过认证依赖注入） + db: 数据库会话 + + Returns: + 更新后的会话详情 + """ + # 1. 验证会话存在性 + stmt = select(Conversation).where(Conversation.id == conversation_id) + result = await db.execute(stmt) + conversation = result.scalars().first() + + if not conversation: + raise AppException("会话不存在", code=404) + + # 2. 合并现有标签（如果有） + existing_tags = {} + if conversation.tags: + existing_tags = ( + dict(conversation.tags) if isinstance(conversation.tags, dict) else {} + ) + + # 3. 合并新旧标签（body.tags 覆盖同名 key） + merged_tags = {**existing_tags, **body.tags} + + # 4. 保存到数据库 + conversation.tags = merged_tags + await db.commit() + await db.refresh(conversation) + + logger.info( + f"坐席 {current_agent.id} 更新会话 {conversation_id} 标签: {merged_tags}" + ) + + # 5. 返回更新后的会话 + response_data = ConversationResponse.model_validate(conversation).model_dump() + return success_response(data=response_data) diff --git a/backend/app/api/messages.py b/backend/app/api/messages.py index ec6dcad..cf18279 100644 --- a/backend/app/api/messages.py +++ b/backend/app/api/messages.py @@ -10,17 +10,19 @@ # 6. POST /api/conversations/{id}/mark-read — 标记已读 # 7. POST /api/messages/image — 上传图片 # 8. POST /api/messages/file — 上传文件 +# 9. GET /api/conversations/{id}/messages/search — 搜索消息（MSG-P1-04） # 消息发送需同时：存数据库 + 调用企微API发送给员工 # ============================================================================= import logging import os +import time from datetime import datetime, timedelta from typing import Optional from uuid import UUID from fastapi import APIRouter, Depends, File, Query, UploadFile -from sqlalchemy import select, update +from sqlalchemy import select, update, or_ from sqlalchemy.ext.asyncio import AsyncSession from app.database import get_db @@ -29,7 +31,12 @@ from app.models.conversation import Conversation from app.models.message import Message from app.schemas.message import MessageCreate, MessageResponse from app.api.agents import get_current_agent + +# RBAC 权限装饰器 +from app.dependencies import require_permission + from app.services.wecom_service import WecomService +from app.services.ws_manager import manager from app.utils.response import AppException, ERR_CONVERSATION_NOT_FOUND, ERR_CONVERSATION_RESOLVED, success_response logger = logging.getLogger(__name__) @@ -48,6 +55,7 @@ RECALLABLE_WINDOW_MINUTES = 2 # GET /api/conversations/{id}/messages — 获取会话消息列表 # -------------------------------------------------------------------------- @router.get("/conversations/{conversation_id}/messages") +@require_permission("conversation", "read", "all") async def list_messages( conversation_id: str, limit: int = Query(50, ge=1, le=100, description="每页消息数量"), @@ -129,6 +137,7 @@ async def list_messages( # POST /api/conversations/{id}/messages — 坐席发送消息 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/messages") +@require_permission("conversation", "create", "all") async def send_message( conversation_id: str, body: MessageCreate, @@ -184,6 +193,7 @@ async def send_message( status="sending", # 初始状态为发送中 recallable_until=recallable_until, is_read=True, # 坐席自己发的消息默认已读 + server_timestamp=int(time.time() * 1000), # [MSG-P0-03] 服务端时间戳(毫秒) ) db.add(message) @@ -235,6 +245,7 @@ async def send_message( # GET /api/conversations/{id}/messages/poll — 坐席轮询新消息 # -------------------------------------------------------------------------- @router.get("/conversations/{conversation_id}/messages/poll") +@require_permission("conversation", "read", "all") async def poll_messages( conversation_id: str, after_message_id: Optional[str] = Query(None, description="返回此消息ID之后的新消息"), @@ -297,6 +308,7 @@ async def poll_messages( # POST /api/messages/{id}/recall — 撤回消息（2分钟内） # -------------------------------------------------------------------------- @router.post("/messages/{message_id}/recall") +@require_permission("conversation", "update", "own") async def recall_message( message_id: str, agent: Agent = Depends(get_current_agent), @@ -342,8 +354,31 @@ async def recall_message( # 将消息内容置为空，表示已撤回 message.content = "[消息已撤回]" message.status = "recalled" + message.is_recalled = True # MSG-P1-01: 标记为已撤回 await db.flush() + # MSG-P1-01: 通过 WebSocket 广播撤回事件给所有参与者 + conv_stmt = select(Conversation).where(Conversation.id == message.conversation_id) + conv_result = await db.execute(conv_stmt) + conversation = conv_result.scalars().first() + if conversation: + participant_ids = [] + if conversation.assigned_agent_id: + participant_ids.append(conversation.assigned_agent_id) + if conversation.employee_id: + participant_ids.append(conversation.employee_id) + # 广播撤回事件 + await manager.broadcast_message_status( + conv_id=message.conversation_id, + msg_id=message.id, + status="recalled", + participant_ids=participant_ids, + extra={ + "recall_by": agent.user_id, + "recall_at": datetime.now().isoformat(), + }, + ) + return success_response(message="消息撤回成功") @@ -351,6 +386,7 @@ async def recall_message( # DELETE /api/messages/{id} — 删除消息 # -------------------------------------------------------------------------- @router.delete("/messages/{message_id}") +@require_permission("conversation", "update", "own") async def delete_message( message_id: str, agent: Agent = Depends(get_current_agent), @@ -394,6 +430,7 @@ async def delete_message( # POST /api/conversations/{id}/mark-read — 标记已读 # -------------------------------------------------------------------------- @router.post("/conversations/{conversation_id}/mark-read") +@require_permission("conversation", "update", "own") async def mark_read( conversation_id: str, agent: Agent = Depends(get_current_agent), @@ -557,4 +594,117 @@ async def upload_message_file( "file_size": file_size, "content_type": file.content_type, } - ) \ No newline at end of file + ) + + +# -------------------------------------------------------------------------- +# GET /api/conversations/{id}/messages/search — 搜索消息（MSG-P1-04） +# -------------------------------------------------------------------------- +@router.get("/conversations/{conversation_id}/messages/search") +async def search_messages( + conversation_id: str, + keyword: str = Query(..., description="搜索关键词"), + limit: int = Query(20, ge=1, le=100, description="返回结果数量限制"), + db: AsyncSession = Depends(get_db), +): + """搜索会话消息（按关键词）。 + + 使用 LIKE 查询匹配消息内容，支持模糊搜索。 + + Args: + conversation_id: 会话ID + keyword: 搜索关键词 + limit: 返回结果数量限制 + db: 数据库会话 + + Returns: + Dict: 统一响应格式，包含匹配的消息列表 + """ + # 校验会话存在 + conv_id_str = str(conversation_id) + conv_stmt = select(Conversation).where(Conversation.id == conv_id_str) + conv_result = await db.execute(conv_stmt) + conversation = conv_result.scalars().first() + if not conversation: + raise ERR_CONVERSATION_NOT_FOUND + + # 构建搜索查询（使用 LIKE 进行模糊匹配） + # 排除已撤回的消息 + search_pattern = f"%{keyword}%" + stmt = ( + select(Message) + .where(Message.conversation_id == conv_id_str) + .where(Message.is_recalled == False) # 排除已撤回的消息 + .where(Message.content.ilike(search_pattern)) # 不区分大小写匹配 + .order_by(Message.created_at.desc()) # 最新消息在前 + .limit(limit) + ) + + result = await db.execute(stmt) + messages = list(result.scalars().all()) + + # 转换为响应格式 + items = [MessageResponse.model_validate(m).model_dump() for m in messages] + + return success_response( + data={ + "items": items, + "total": len(items), + "keyword": keyword, + } + ) + + +# -------------------------------------------------------------------------- +# POST /api/conversations/{id}/typing — 发送 typing 事件（MSG-P1-03） +# -------------------------------------------------------------------------- +@router.post("/conversations/{conversation_id}/typing") +@require_permission("conversation", "read", "all") +async def send_typing_event( + conversation_id: str, + agent: Agent = Depends(get_current_agent), + db: AsyncSession = Depends(get_db), +): + """发送 typing 事件，通知对方正在输入。 + + 通过 WebSocket 广播 typing 事件给会话参与者。 + + Args: + conversation_id: 会话ID + agent: 当前坐席 + db: 数据库会话 + + Returns: + Dict: 统一响应格式 + """ + # 校验会话存在 + conv_id_str = str(conversation_id) + conv_stmt = select(Conversation).where(Conversation.id == conv_id_str) + conv_result = await db.execute(conv_stmt) + conversation = conv_result.scalars().first() + if not conversation: + raise ERR_CONVERSATION_NOT_FOUND + + # 构建参与者列表 + participant_ids = [] + if conversation.assigned_agent_id: + participant_ids.append(conversation.assigned_agent_id) + if conversation.employee_id: + participant_ids.append(conversation.employee_id) + + # 广播 typing 事件（排除发送者本人） + payload = { + "type": "typing", + "conv_id": conv_id_str, + "sender_id": agent.user_id, + "sender_name": agent.name or "坐席", + } + + for pid in participant_ids: + if pid != agent.user_id: # 不发给自己 + if pid in manager.active_connections: + await manager.send_to_agent(pid, payload) + elif pid in manager.employee_connections: + await manager.send_to_employee(pid, payload) + + return success_response(message="typing 事件已发送") \ No newline at end of file diff --git a/backend/app/dependencies.py b/backend/app/dependencies.py index ac75700..74ec06d 100644 --- a/backend/app/dependencies.py +++ b/backend/app/dependencies.py @@ -12,16 +12,22 @@ import json import logging from dataclasses import dataclass from functools import wraps -from typing import List, Optional +from typing import List, Optional, Union import redis.asyncio as aioredis from fastapi import Depends, HTTPException, status from fastapi.security import HTTPAuthorizationCredentials, HTTPBearer from app.config import settings +from app.models.agent import Agent from app.services.token_service import TokenService from app.utils.response import AppException +# 延迟导入 get_current_agent 以避免循环依赖 +def _get_current_agent(): + from app.api.agents import get_current_agent + return get_current_agent + logger = logging.getLogger(__name__) # HTTP Bearer 认证方案 @@ -324,30 +330,63 @@ def require_permission( def decorator(func): sig = inspect.signature(func) params = list(sig.parameters.values()) - params.append( - inspect.Parameter( - 'current_user', - inspect.Parameter.KEYWORD_ONLY, - annotation=UserInfo, - default=Depends(get_current_user), + param_names = {p.name for p in params} + + # 智能检测参数名：优先使用函数已定义的参数名 + # 支持 current_user (通用/管理端) 和 current_agent (坐席端) + if 'current_agent' in param_names: + param_name = 'current_agent' + param_annotation = Agent + param_default = Depends(_get_current_agent()) + else: + param_name = 'current_user' + param_annotation = UserInfo + param_default = Depends(get_current_user) + + # 检查是否需要添加参数 + needs_param = param_name not in param_names + if needs_param: + params.append( + inspect.Parameter( + param_name, + inspect.Parameter.KEYWORD_ONLY, + annotation=param_annotation, + default=param_default, + ) ) - ) new_sig = sig.replace(parameters=params) @wraps(func) async def wrapper(*args, **kwargs): - current_user = kwargs.pop('current_user') + # 提取注入的用户/坐席信息 + current_user = kwargs.pop(param_name) # 拉用户所有角色的 permissions # 注: UserInfo.roles 是角色名列表,permissions 是 {role: [perm]} 字典 # 首次实现简化: 角色判断 + admin 通配符 # 完整实现需要查 DB 拉 permissions,见 rbac_service.check_permission - user_roles = set(current_user.roles or []) + # 支持两种类型： + # 1. UserInfo (H5/管理端): 有 roles 属性 (List[str]) + # 2. Agent (坐席端): 有 role 属性 (str) + if hasattr(current_user, 'roles'): + user_roles = set(current_user.roles or []) + user_id = current_user.employee_id + elif hasattr(current_user, 'role'): + # Agent 类型：role 是字符串，直接作为角色 + user_roles = {current_user.role} if current_user.role else set() + user_id = current_user.user_id + else: + # 兼容：没有 roles 或 role 属性的情况 + user_roles = set() + user_id = getattr(current_user, 'user_id', 'unknown') + + # 保存 user_id 供后续使用 + current_user._rbac_user_id = user_id # 1. admin 角色直通(通配符 *:*:all) if "admin" in user_roles: - return await func(*args, current_user=current_user, **kwargs) + return await func(*args, **{param_name: current_user}, **kwargs) # 2. 其他角色: 走 rbac_service.check_permission # 简化: 这里只看角色名,不查 DB(性能考虑) @@ -372,7 +411,7 @@ def require_permission( if not has_perm: logger.warning( - f"用户 {current_user.employee_id} 权限不足: " + f"用户 {user_id} 权限不足: " f"角色 {list(user_roles)}, 缺 {perm_string}" ) raise HTTPException( @@ -380,7 +419,7 @@ def require_permission( detail=f"权限不足: 需要 {perm_string}", ) - return await func(*args, current_user=current_user, **kwargs) + return await func(*args, **{param_name: current_user}, **kwargs) wrapper.__signature__ = new_sig return wrapper diff --git a/backend/app/services/rbac_service.py b/backend/app/services/rbac_service.py index 22e8099..d9505f7 100644 --- a/backend/app/services/rbac_service.py +++ b/backend/app/services/rbac_service.py @@ -52,6 +52,7 @@ ROLE_PERMISSIONS: Dict[str, Set[Tuple[str, str, str]]] = { ("conversation", "read", "own"), ("conversation", "read", "all"), # 看所有未分配的会话(坐席工作台需要) ("conversation", "update", "own"), + ("conversation", "update", "all"), # 抢单需要能更新其他坐席的会话 ("conversation", "create", "all"), }, diff --git a/frontend-agent/src/api/index.ts b/frontend-agent/src/api/index.ts index 863e705..4efb468 100644 --- a/frontend-agent/src/api/index.ts +++ b/frontend-agent/src/api/index.ts @@ -4,7 +4,7 @@ // 说明：创建 Axios 实例，配置： // 1. 请求基础 URL // 2. 请求拦截器（添加认证头等） -// 3. 响应拦截器（统一错误处理） +// 3. 响应拦截器（统一错误处理 + Token 静默刷新） // ============================================================================= import axios from 'axios' @@ -29,6 +29,129 @@ const apiClient: AxiosInstance = axios.create({ }, }) +// -------------------------------------------------------------------------- +// AUTH-P0-02: Token 静默刷新机制 +// -------------------------------------------------------------------------- +// Token 存储键名 +const TOKEN_KEY = 'agent_token' + +// OAuth2 重定向计数器 key +const OAUTH_REDIRECT_COUNT_KEY = 'oauth_redirect_count' +// 最大允许重定向次数 +const OAUTH_MAX_REDIRECT_COUNT = 3 + +// 401 处理锁（防止并发请求同时触发多次认证过期处理） +let _authExpiredPromise: Promise | null = null + +// Token 刷新锁（防止多个并发请求同时触发刷新） +let _refreshPromise: Promise | null = null + +/** + * 静默刷新 Token + * 在 Token 过期前 5 分钟自动调用，成功后返回 true + */ +async function silentRefreshToken(): Promise { + // 如果已有刷新在进行中，等待同一个 Promise + if (_refreshPromise) { + return _refreshPromise + } + + const token = localStorage.getItem(TOKEN_KEY) + if (!token) { + return false + } + + console.log('[API] 尝试静默刷新 Token...') + + // 创建刷新 Promise 并缓存 + _refreshPromise = (async () => { + try { + const response = await axios.post('/api/auth/refresh', null, { + params: { token }, + timeout: 5000, // 刷新接口 5 秒超时 + }) + + if (response.data.code === 0) { + // 刷新成功，Token TTL 已延长 + console.log('[API] Token 静默刷新成功') + return true + } + + console.warn('[API] Token 刷新失败:', response.data.message) + return false + } catch (e) { + console.warn('[API] Token 刷新异常:', e) + return false + } finally { + _refreshPromise = null + } + })() + + return _refreshPromise +} + +/** + * 处理认证过期/未授权 + * AUTH-P0-02: 增强 401 处理，先尝试刷新 Token，成功则重放请求 + */ +async function handleAuthExpired(source: 'http401' | 'biz1002'): Promise { + const label = source === 'http401' ? 'HTTP 401' : '业务码 1002' + + // 如果已有 401 正在处理中，复用同一个 Promise，避免多次处理 + if (_authExpiredPromise) { + console.warn(`[API] ${label} 未授权 — 已有处理进行中，等待完成`) + return _authExpiredPromise + } + + console.warn(`[API] ${label} 未授权，尝试刷新 Token...`) + + // 创建处理 Promise 并缓存（去重用） + _authExpiredPromise = (async () => { + try { + // 第一步：尝试静默刷新 Token + const refreshSuccess = await silentRefreshToken() + + if (refreshSuccess) { + console.log('[API] Token 刷新成功，不需要跳转') + _authExpiredPromise = null + return + } + + // 第二步：刷新失败，清除凭证并跳转登录 + console.warn('[API] Token 刷新失败，清除凭证并跳转登录') + + // 清除本地 token + localStorage.removeItem(TOKEN_KEY) + localStorage.removeItem('portal_token') + + // 跳转登录页 + ElMessage.warning('登录已过期，请重新登录') + + // 防循环检测：超过最大重定向次数时停止跳转 + const currentCount = parseInt(localStorage.getItem(OAUTH_REDIRECT_COUNT_KEY) || '0', 10) + if (currentCount >= OAUTH_MAX_REDIRECT_COUNT) { + console.error('[API] 登录重定向次数超限，疑似无限循环，停止重定向') + ElMessage.error('登录状态异常，请刷新页面重试') + return + } + + console.warn(`[API] 登录重定向计数: ${currentCount}/${OAUTH_MAX_REDIRECT_COUNT}`) + + // 动态导入避免循环依赖 + import('@/router').then(router => { + router.default.push('/login') + }) + } catch (e) { + console.warn('[API] 401 处理失败:', e) + } finally { + // 处理完成后清除锁，允许未来的 401 重新触发 + _authExpiredPromise = null + } + })() + + return _authExpiredPromise +} + // -------------------------------------------------------------------------- // 请求拦截器 // -------------------------------------------------------------------------- @@ -36,7 +159,7 @@ const apiClient: AxiosInstance = axios.create({ apiClient.interceptors.request.use( (config: InternalAxiosRequestConfig) => { // 从 localStorage 获取坐席 token，添加到请求头 - const token = localStorage.getItem('agent_token') + const token = localStorage.getItem(TOKEN_KEY) if (token) { config.headers.Authorization = `Bearer ${token}` } @@ -63,13 +186,10 @@ apiClient.interceptors.response.use( // 业务错误：显示错误消息 ElMessage.error(res.message || '请求失败') - // 特殊错误码处理 + // AUTH-P0-02: 特殊错误码处理 - 尝试刷新 Token if (res.code === 1002) { - // 未授权：跳转到登录页 - // 动态导入避免循环依赖 - import('@/router').then(router => { - router.default.push('/login') - }) + // 未授权：先尝试刷新 Token + handleAuthExpired('biz1002') } // 返回 rejected Promise，让调用方的 catch 能捕获 @@ -79,7 +199,7 @@ apiClient.interceptors.response.use( // 业务成功：返回完整响应（调用方从 response.data.data 获取业务数据） return response }, - (error) => { + async (error) => { // 网络错误或服务器错误（HTTP 状态码非 2xx） let message = '网络异常，请稍后重试' @@ -87,7 +207,9 @@ apiClient.interceptors.response.use( // 服务器返回了错误状态码 switch (error.response.status) { case 401: - message = '未授权，请重新登录' + // AUTH-P0-02: 先尝试静默刷新 Token，成功则重放请求 + await handleAuthExpired('http401') + // 不显示通用提示，因为会自动处理 break case 403: message = '拒绝访问' @@ -106,8 +228,10 @@ apiClient.interceptors.response.use( message = '请求超时，请稍后重试' } - // 显示错误提示 - ElMessage.error(message) + // 显示错误提示（401 时不显示通用提示，因为会自动处理） + if (!error.response || error.response.status !== 401) { + ElMessage.error(message) + } return Promise.reject(error) } diff --git a/frontend-portal/src/router/index.ts b/frontend-portal/src/router/index.ts index aad171f..9bbcedd 100644 --- a/frontend-portal/src/router/index.ts +++ b/frontend-portal/src/router/index.ts @@ -41,6 +41,24 @@ const routes = [ title: '正在加载...', }, }, + { + // 错误提示页面（OAuth2 回调异常兜底） + path: '/error', + name: 'ErrorPage', + component: () => import('@/views/ErrorPage.vue'), + meta: { + title: '登录失败', + }, + }, + { + // 错误提示页面（/login/error 别名，兼容后端重定向） + path: '/login/error', + name: 'LoginError', + component: () => import('@/views/ErrorPage.vue'), + meta: { + title: '登录失败', + }, + }, { // 404 页面 path: '/:pathMatch(.*)*', diff --git a/frontend-portal/src/views/ErrorPage.vue b/frontend-portal/src/views/ErrorPage.vue new file mode 100644 index 0000000..1766da1 --- /dev/null +++ b/frontend-portal/src/views/ErrorPage.vue @@ -0,0 +1,156 @@ + + + + + diff --git a/jumpserver-webcli/requirements.txt b/jumpserver-webcli/requirements.txt new file mode 100644 index 0000000..b868bdd --- /dev/null +++ b/jumpserver-webcli/requirements.txt @@ -0,0 +1,3 @@ +playwright>=1.40.0 +pyotp>=2.9.0 +requests>=2.28.0 diff --git a/ops-tools/jms_ops.py b/ops-tools/jms_ops.py new file mode 100644 index 0000000..941a790 --- /dev/null +++ b/ops-tools/jms_ops.py @@ -0,0 +1,698 @@ +""" +JumpServer 运维工具集 — 统一入口 + +整合能力: + 1. exec — 远程命令执行 (单命令/批量/并行) + 2. upload — 文件上传 (base64 通道, 小文件) + 3. download — 文件下载 (base64 通道, 小文件) + 4. batch — 批量命令复用会话 (一次登录, 多命令) + +技术栈: + - Playwright (登录 + cookies) + - REST API (Connection Token) + - plink PTY (命令执行) + - base64 编码 (文件传输) + +用法: + python jms_ops.py exec -c "hostname" + python jms_ops.py exec -c "hostname" -c "uptime" -c "docker ps" + python jms_ops.py exec -c "hostname" -c "uptime" --parallel + python jms_ops.py upload local_file.txt /tmp/remote_file.txt + python jms_ops.py download /tmp/remote_file.txt local_file.txt + python jms_ops.py batch -f commands.txt +""" +import sys, os, json, base64, time, pyotp, re, subprocess, threading, argparse, hashlib +from pathlib import Path +from playwright.sync_api import sync_playwright + +try: + sys.stdout.reconfigure(encoding="utf-8") + sys.stderr.reconfigure(encoding="utf-8") +except: pass + +# ============================================================ +# 配置 +# ============================================================ +# 动态路径：基于环境变量或自动检测 +_WORKBUDDY_DIR = Path(os.environ.get("WORKBUDDY_DIR", Path.home() / ".workbuddy")) +_SKILL_DIR = Path(os.environ.get("JP_SKILL_DIR", _WORKBUDDY_DIR / "skills" / "jumpserver-automation-shareable")) +CONFIG_PATH = _SKILL_DIR / "config" / "jumpserver_config.json" +OTP_SECRET_PATH = _SKILL_DIR / "scripts" / "otp_secret.key" +OUTPUT_DIR = _SKILL_DIR / "scripts" / "webcli_output" +USER_DATA_DIR = str(Path(os.environ.get("TEMP", os.path.join(str(Path.home()), "AppData", "Local", "Temp"))) / "chrome-jumpserver-v10") +CDP_URL = os.environ.get("CDP_URL", "http://localhost:9224") + +SSH_GATEWAY = "jumpserver.dc.servyou-it.com" +SSH_PORT = "2222" +PLINK_EXE = r"C:\Program Files\PuTTY\plink.exe" + +# ANSI 转义码 (含 CSI + OSC) +ANSI_ESCAPE_RE = re.compile( + r'\x1B(?:[@-Z\\-_]|\[[0-?]*[ -/]*[@-~]|\][^\x07\x1B]*(?:\x07|\x1B\\))' +) +# shell prompt: [admin@host ~]$ +PROMPT_RE = re.compile(r'\[[^\]]+@[^\]]+\][^\$#]*[\$#]\s*$') + +# ============================================================ +# 配置加载 +# ============================================================ + +def load_config(): + with open(CONFIG_PATH) as f: cfg = json.load(f) + return cfg["url"].rstrip("/"), cfg.get("username", "sxn"), base64.b64decode(cfg["password"]).decode("utf-8") + +def load_otp(): + return OTP_SECRET_PATH.read_text().strip() if OTP_SECRET_PATH.exists() else None + +def strip_ansi(text): + return ANSI_ESCAPE_RE.sub('', text) + + +# ============================================================ +# 浏览器登录 + REST API (获取 Connection Token) +# ============================================================ + +def ensure_logged_in(context, url, username, password, otp_code): + """确保 workbench tab 已登录""" + workbench = None + for pg in context.pages: + if 'workbench' in pg.url.lower(): + workbench = pg + break + if not workbench: + workbench = context.new_page() + workbench.goto(f"{url}/users/login/", wait_until="networkidle", timeout=30000) + workbench.wait_for_timeout(2000) + if workbench.locator('input[name="username"]').count() > 0: + workbench.fill('input[name="username"]', username) + workbench.fill('input[type="password"]', password) + workbench.click('button[type="submit"]') + workbench.wait_for_timeout(3000) + if workbench.locator('input[name="code"]').count() > 0: + workbench.locator('input[name="code"]').fill(str(otp_code)) + workbench.locator('#submit_button, button[type="submit"]').first.click() + for i in range(40): + if workbench.locator('input[name="code"]').count() == 0: + break + try: + body_text = workbench.evaluate("() => document.body ? document.body.innerText : ''") + if any(kw in body_text for kw in ['验证码错误', 'OTP 错误', '已过期', 'expired', 'Invalid']): + break + except: pass + workbench.wait_for_timeout(500) + for _ in range(60): + if 'workbench' in workbench.url.lower(): + break + workbench.wait_for_timeout(500) + return workbench + + +def get_connection_tokens(count=1): + """ + 获取 N 个 Connection Token + + 一次 Playwright 登录 → REST API 创建 N 个 token + 返回: [(token_id, token_secret), ...] + """ + import requests + url, username, password = load_config() + otp = pyotp.TOTP(load_otp()).now() if load_otp() else None + + with sync_playwright() as p: + try: + browser = p.chromium.connect_over_cdp(CDP_URL) + contexts = browser.contexts + context = contexts[0] if contexts else browser.new_context() + except Exception as e: + context = p.chromium.launch_persistent_context( + USER_DATA_DIR, headless=False, channel="chrome", + args=["--no-sandbox", "--remote-debugging-port=9224"], + ignore_https_errors=True, + ) + + workbench = ensure_logged_in(context, url, username, password, otp) + if not workbench: + print("❌ 登录失败") + return [] + + session = requests.Session() + csrf_token = None + for c in context.cookies(): + session.cookies.set(c['name'], c['value'], domain=c.get('domain', ''), path=c.get('path', '/')) + if c['name'] == 'jms_csrftoken': + csrf_token = c['value'] + if csrf_token: + session.headers['X-CSRFToken'] = csrf_token + + # 获取资产 + resp = session.get(f"{url}/api/v1/perms/users/assets/", params={"offset": 0, "limit": 100}) + assets = resp.json() if resp.status_code == 200 else [] + if isinstance(assets, dict): assets = assets.get("results", []) + target_asset = next((a for a in assets if "hz-oa-ai-g-dataquery" in a.get("hostname", "").lower()), None) + if not target_asset: + print("❌ 未找到目标资产") + return [] + + # 获取系统用户 + resp = session.get(f"{url}/api/v1/perms/users/assets/{target_asset['id']}/system-users/") + system_users = resp.json() if resp.status_code == 200 else [] + if isinstance(system_users, dict): system_users = system_users.get("results", []) + target_su = next((su for su in system_users if "admin" in su.get("name", "").lower()), None) + if not target_su and system_users: target_su = system_users[0] + if not target_su: + print("❌ 未找到系统用户") + return [] + + # 创建 count 个 token + tokens = [] + for i in range(count): + resp = session.post( + f"{url}/api/v1/authentication/connection-token/", + json={"asset": target_asset['id'], "system_user": target_su['id'], "connect_method": "ssh_client"} + ) + if resp.status_code == 201: + td = resp.json() + tokens.append((td['id'], td['secret'])) + return tokens + + +# ============================================================ +# PlinkSession — plink PTY 会话管理器 +# ============================================================ + +class PlinkSession: + """ + plink PTY 会话管理器 + + 一次启动 plink, 在同一个会话里执行多个命令。 + """ + + def __init__(self, ssh_user, ssh_password, connect_timeout=30, verbose=True): + self.ssh_user = ssh_user + self.ssh_password = ssh_password + self.connect_timeout = connect_timeout + self.verbose = verbose + self.proc = None + self.stdout_chunks = [] + self.stderr_chunks = [] + self.stop_reading = threading.Event() + self._stdout_thread = None + self._stderr_thread = None + self._ready = False + + def connect(self): + """启动 plink, 等待 shell prompt 就绪""" + if self.verbose: + print(f" 🚀 启动 plink PTY 会话...") + + self.proc = subprocess.Popen( + [PLINK_EXE, '-ssh', '-P', SSH_PORT, '-t', '-pw', self.ssh_password, f'{self.ssh_user}@{SSH_GATEWAY}'], + stdin=subprocess.PIPE, + stdout=subprocess.PIPE, + stderr=subprocess.PIPE, + bufsize=0, + ) + + # 后台线程逐块读取 + def reader(stream, buf_list): + fd = stream.fileno() + while not self.stop_reading.is_set(): + try: + chunk = os.read(fd, 4096) + except (OSError, ValueError): + break + if not chunk: + break + buf_list.append(chunk) + + self._stdout_thread = threading.Thread(target=reader, args=(self.proc.stdout, self.stdout_chunks), daemon=True) + self._stderr_thread = threading.Thread(target=reader, args=(self.proc.stderr, self.stderr_chunks), daemon=True) + self._stdout_thread.start() + self._stderr_thread.start() + + # 等待 shell prompt 就绪 + start = time.time() + while time.time() - start < self.connect_timeout: + clean = strip_ansi(self._get_stdout()) + stderr_clean = strip_ansi(self._get_stderr()) + combined = clean + '\n' + stderr_clean + + if 'Press Return to begin session' in combined: + self._send('\r\n') + time.sleep(0.5) + elif PROMPT_RE.search(clean): + # 设置终端宽度, 防止长行折行 + self._send('stty cols 1000 2>/dev/null\r\n') + time.sleep(0.5) + self.stdout_chunks.clear() + self._ready = True + if self.verbose: + print(f" ✅ 会话就绪 ({time.time()-start:.1f}s)") + return True + + time.sleep(0.3) + + if self.verbose: + print(f" ❌ 会话超时 ({self.connect_timeout}s)") + return False + + def _get_stdout(self): + return b''.join(self.stdout_chunks).decode('utf-8', errors='replace') + + def _get_stderr(self): + return b''.join(self.stderr_chunks).decode('utf-8', errors='replace') + + def _send(self, text): + if self.proc and self.proc.stdin: + self.proc.stdin.write(text.encode('utf-8')) + self.proc.stdin.flush() + + def run_command(self, command, timeout=15): + """ + 在当前会话中执行一个命令 + + 返回: {"success": bool, "output": str, "elapsed": float, "timed_out": bool} + """ + if not self._ready: + return {"success": False, "output": "", "elapsed": 0, "timed_out": False, "error": "session not ready"} + + offset = len(self._get_stdout()) + cmd_start = time.time() + + if self.verbose: + print(f" 💻 [{time.strftime('%H:%M:%S')}] 发送: {command[:80]}") + self._send(f'{command}\r\n') + + # 轮询检测 prompt 重新出现 + while time.time() - cmd_start < timeout: + new_stdout = self._get_stdout()[offset:] + new_clean = strip_ansi(new_stdout) + + lines = new_clean.split('\n') + for line in lines[-4:]: + stripped = line.strip() + if command in stripped: + continue + if PROMPT_RE.search(stripped): + elapsed = time.time() - cmd_start + # 提取命令输出 + output_lines = [] + found_echo = False + for l in lines: + s = l.strip() + if not found_echo: + if command in s: + found_echo = True + continue + else: + if PROMPT_RE.search(s): + break + if s: + output_lines.append(s) + + if not output_lines and not found_echo: + for i, l in enumerate(lines): + s = l.strip() + if command in s: + for j in range(i + 1, len(lines)): + s2 = lines[j].strip() + if PROMPT_RE.search(s2): + break + if s2: + output_lines.append(s2) + break + + return { + "success": True, + "output": '\n'.join(output_lines), + "elapsed": elapsed, + "timed_out": False, + } + + time.sleep(0.3) + + # 超时 + elapsed = time.time() - cmd_start + new_stdout = self._get_stdout()[offset:] + if self.verbose: + print(f" ⏰ 超时 ({elapsed:.1f}s)") + return { + "success": False, + "output": strip_ansi(new_stdout).strip(), + "elapsed": elapsed, + "timed_out": True, + } + + def close(self): + """发送 exit, 关闭会话""" + if self._ready and self.proc: + try: + self._send('exit\r\n') + time.sleep(0.5) + except: pass + + self.stop_reading.set() + if self._stdout_thread: + self._stdout_thread.join(timeout=2) + if self._stderr_thread: + self._stderr_thread.join(timeout=2) + + if self.proc: + try: self.proc.stdin.close() + except: pass + try: self.proc.wait(timeout=3) + except: + try: self.proc.kill(); self.proc.wait(timeout=2) + except: pass + + +# ============================================================ +# 命令执行: exec / batch +# ============================================================ + +def cmd_exec(commands, cmd_timeout=15, parallel=False): + """执行命令 (单条或多条, 串行或并行)""" + os.makedirs(str(OUTPUT_DIR), exist_ok=True) + total_start = time.time() + results = [] + + if not parallel: + # 串行: 一个 token + 一个会话 + print(f"模式: 串行 ({len(commands)} 命令, 超时 {cmd_timeout}s/命令)") + tokens = get_connection_tokens(1) + if not tokens: + print("❌ 获取 token 失败") + return [] + + token_id, token_secret = tokens[0] + session = PlinkSession(f"JMS-{token_id}", token_secret) + if not session.connect(): + print("❌ 会话启动失败") + return [{"command": c, "success": False, "output": "", "elapsed": 0, "timed_out": False} for c in commands] + + for i, cmd in enumerate(commands, 1): + print(f"\n--- [{i}/{len(commands)}] ---") + result = session.run_command(cmd, timeout=cmd_timeout) + result["command"] = cmd + result["index"] = i + status = "✅" if result["success"] else ("⏰" if result.get("timed_out") else "❌") + print(f" {status} 完成 ({result['elapsed']:.1f}s)") + if result["output"]: + for line in result["output"].split("\n"): + if line.strip(): + print(f" | {line.strip()[:120]}") + results.append(result) + + session.close() + + else: + # 并行: N 个 token + N 个会话 + print(f"模式: 并行 ({len(commands)} 命令, 超时 {cmd_timeout}s/命令)") + tokens = get_connection_tokens(len(commands)) + if len(tokens) < len(commands): + print(f"⚠️ 只获取到 {len(tokens)}/{len(commands)} 个 token, 降级为串行") + return cmd_exec(commands, cmd_timeout=cmd_timeout, parallel=False) + + def execute_single(idx, cmd, tid, tsec): + session = PlinkSession(f"JMS-{tid}", tsec, verbose=False) + if not session.connect(): + return {"command": cmd, "success": False, "output": "", "elapsed": 0, "timed_out": False, "index": idx} + result = session.run_command(cmd, timeout=cmd_timeout) + result["command"] = cmd + result["index"] = idx + session.close() + return result + + threads = [] + thread_results = [None] * len(commands) + + def worker(idx, cmd, tid, tsec): + thread_results[idx] = execute_single(idx, cmd, tid, tsec) + + for i, cmd in enumerate(commands): + tid, tsec = tokens[i] + t = threading.Thread(target=worker, args=(i, cmd, tid, tsec)) + threads.append(t) + t.start() + + for t in threads: + t.join(timeout=cmd_timeout + 60) + + results = [r for r in thread_results if r is not None] + results.sort(key=lambda x: x.get("index", 0)) + + for r in results: + status = "✅" if r.get("success") else "❌" + print(f" [{r.get('index')}] {status} {r.get('elapsed', 0):.1f}s — {r.get('command', '')[:50]}") + + total_elapsed = time.time() - total_start + success_count = sum(1 for r in results if r.get("success")) + + print(f"\n{'=' * 50}") + print(f"汇总: {success_count}/{len(commands)} 成功, 总耗时 {total_elapsed:.1f}s") + print(f"{'=' * 50}") + + # 保存报告 + timestamp = time.strftime("%Y%m%d_%H%M%S") + report_path = OUTPUT_DIR / f"exec_{timestamp}.md" + lines = [ + f"# JumpServer 执行报告", + f"", + f"**时间**: {time.strftime('%Y-%m-%d %H:%M:%S')}", + f"**模式**: {'并行' if parallel else '串行'}", + f"**命令数**: {len(commands)}", + f"**成功**: {success_count}/{len(commands)}", + f"**总耗时**: {total_elapsed:.1f}s", + f"", + ] + for r in results: + status = "✅" if r.get("success") else "❌" + lines.append(f"## [{r.get('index', '?')}] {status} {r.get('elapsed', 0):.1f}s — {r.get('command', '')[:60]}") + lines.append(f"") + lines.append(f"```") + lines.append(r.get("output", "(空)")) + lines.append(f"```") + lines.append(f"") + report_path.write_text("\n".join(lines), encoding="utf-8") + print(f"💾 报告: {report_path}") + + return results + + +# ============================================================ +# 文件传输: upload / download (base64 通道) +# ============================================================ + +def cmd_upload(local_path, remote_path, cmd_timeout=60): + """ + 上传本地文件到远程 (base64 编码, 分块发送) + + 适用: 小文件 (配置/脚本/文本, <100KB) + 大文件建议用 elFinder Web UI + """ + local_file = Path(local_path) + if not local_file.exists(): + print(f"❌ 本地文件不存在: {local_path}") + return False + + local_data = local_file.read_bytes() + local_md5 = hashlib.md5(local_data).hexdigest() + b64_data = base64.b64encode(local_data).decode('ascii') + + print(f"📤 上传: {local_path} → {remote_path}") + print(f" 原始: {len(local_data)} bytes, base64: {len(b64_data)} chars, MD5: {local_md5[:12]}...") + + # 分块 (每块 500 chars, PTY 宽度 1000 留余量) + CHUNK_SIZE = 500 + chunks = [b64_data[i:i+CHUNK_SIZE] for i in range(0, len(b64_data), CHUNK_SIZE)] + print(f" 分 {len(chunks)} 块发送") + + # 获取 token + 启动会话 + tokens = get_connection_tokens(1) + if not tokens: + print("❌ 获取 token 失败") + return False + + token_id, token_secret = tokens[0] + session = PlinkSession(f"JMS-{token_id}", token_secret) + if not session.connect(): + print("❌ 会话启动失败") + return False + + try: + # 1. 清空目标文件 + session.run_command(f'> {remote_path}', timeout=5) + + # 2. 逐块追加 + for i, chunk in enumerate(chunks): + cmd = f"echo '{chunk}' | base64 -d >> {remote_path}" + r = session.run_command(cmd, timeout=10) + if not r["success"]: + print(f" ❌ 块 {i+1}/{len(chunks)} 发送失败") + return False + if (i+1) % 20 == 0 or (i+1) == len(chunks): + print(f" 📦 已发送 {i+1}/{len(chunks)} 块") + + # 3. 验证大小 + r = session.run_command(f'wc -c < {remote_path}', timeout=5) + if r["success"]: + remote_size = int(r["output"].strip()) if r["output"].strip().isdigit() else -1 + if remote_size == len(local_data): + print(f" ✅ 上传成功! 大小匹配 ({remote_size} bytes)") + + # 4. MD5 验证 (可选, 文本文件可能因换行符不匹配) + r2 = session.run_command(f'md5sum {remote_path}', timeout=5) + if r2["success"]: + remote_md5 = r2["output"].split()[0] + if remote_md5 == local_md5: + print(f" ✅ MD5 匹配! 文件完整") + else: + print(f" ⚠️ MD5 不匹配 (本地 {local_md5[:12]}, 远程 {remote_md5[:12]})") + print(f" 可能原因: PTY 换行符转换, 文本内容应可读") + return True + else: + print(f" ❌ 大小不匹配 (本地 {len(local_data)}, 远程 {remote_size})") + return False + else: + print(f" ⚠️ 无法验证远程文件大小") + return True + finally: + session.close() + + +def cmd_download(remote_path, local_path, cmd_timeout=30): + """ + 下载远程文件到本地 (base64 编码, marker 提取) + + 适用: 小文件 + """ + print(f"📥 下载: {remote_path} → {local_path}") + + # 获取 token + 启动会话 + tokens = get_connection_tokens(1) + if not tokens: + print("❌ 获取 token 失败") + return False + + token_id, token_secret = tokens[0] + session = PlinkSession(f"JMS-{token_id}", token_secret) + if not session.connect(): + print("❌ 会话启动失败") + return False + + try: + # 用 marker 包围 base64 输出, 精确提取 + cmd = f'echo "<<>>" && base64 {remote_path} && echo "<<>>"' + result = session.run_command(cmd, timeout=cmd_timeout) + + if not result["success"]: + print(f"❌ 命令失败: {result.get('output', '')[:200]}") + return False + + output = result["output"] + if '<<>>' in output and '<<>>' in output: + b64_part = output.split('<<>>', 1)[1] + b64_part = b64_part.split('<<>>', 1)[0] + b64_clean = ''.join(b64_part.split()) + + try: + file_data = base64.b64decode(b64_clean) + local_file = Path(local_path) + local_file.write_bytes(file_data) + local_md5 = hashlib.md5(file_data).hexdigest() + print(f" ✅ 下载成功! {len(file_data)} bytes, MD5: {local_md5[:12]}...") + print(f" 💾 已保存: {local_path}") + return True + except Exception as e: + print(f" ❌ base64 解码失败: {e}") + return False + else: + print(f" ❌ 未找到 marker (文件可能不存在)") + print(f" 输出: {output[:200]}") + return False + finally: + session.close() + + +# ============================================================ +# CLI +# ============================================================ + +def main(): + parser = argparse.ArgumentParser( + description="JumpServer 运维工具集 — 远程命令执行 + 文件传输", + formatter_class=argparse.RawDescriptionHelpFormatter, + epilog=""" +示例: + # 远程命令执行 + %(prog)s exec -c "hostname" + %(prog)s exec -c "hostname" -c "uptime" -c "docker ps" + %(prog)s exec -c "hostname" -c "uptime" --parallel + + # 批量命令 (从文件) + %(prog)s batch -f commands.txt + + # 文件传输 + %(prog)s upload local_config.conf /tmp/remote_config.conf + %(prog)s download /etc/nginx/nginx.conf ./nginx.conf.bak + """, + ) + subparsers = parser.add_subparsers(dest="subcommand", help="子命令") + + # exec + p_exec = subparsers.add_parser("exec", help="远程命令执行 (单条/多条/并行)") + p_exec.add_argument("-c", "--command", action="append", required=True, help="远程命令 (可多次指定)") + p_exec.add_argument("--cmd-timeout", type=int, default=15, help="每命令超时秒数 (默认 15)") + p_exec.add_argument("--parallel", action="store_true", help="并行模式 (每命令独立 token+会话)") + + # batch + p_batch = subparsers.add_parser("batch", help="批量命令 (从文件读取)") + p_batch.add_argument("-f", "--file", required=True, help="命令文件 (每行一个, # 开头为注释)") + p_batch.add_argument("--cmd-timeout", type=int, default=15, help="每命令超时秒数") + p_batch.add_argument("--parallel", action="store_true", help="并行模式") + + # upload + p_upload = subparsers.add_parser("upload", help="文件上传 (base64 通道, 小文件)") + p_upload.add_argument("local", help="本地文件路径") + p_upload.add_argument("remote", help="远程文件路径") + + # download + p_download = subparsers.add_parser("download", help="文件下载 (base64 通道, 小文件)") + p_download.add_argument("remote", help="远程文件路径") + p_download.add_argument("local", help="本地保存路径") + + args = parser.parse_args() + + if args.subcommand == "exec": + results = cmd_exec(args.command, cmd_timeout=args.cmd_timeout, parallel=args.parallel) + success = all(r.get("success") for r in results) if results else False + sys.exit(0 if success else 1) + + elif args.subcommand == "batch": + commands = [] + for line in Path(args.file).read_text(encoding="utf-8").splitlines(): + line = line.strip() + if line and not line.startswith("#"): + commands.append(line) + if not commands: + print("❌ 命令文件为空") + sys.exit(1) + results = cmd_exec(commands, cmd_timeout=args.cmd_timeout, parallel=args.parallel) + success = all(r.get("success") for r in results) if results else False + sys.exit(0 if success else 1) + + elif args.subcommand == "upload": + success = cmd_upload(args.local, args.remote) + sys.exit(0 if success else 1) + + elif args.subcommand == "download": + success = cmd_download(args.remote, args.local) + sys.exit(0 if success else 1) + + else: + parser.print_help() + sys.exit(1) + + +if __name__ == "__main__": + main()