# ============================================================================= # 企微IT智能服务台 — 扫码登录 API # ============================================================================= # 说明：扫码登录是 Phase 1.1 的核心功能,用于替代坐席端"用户名密码+企微 # OAuth"双因素登录,提供"用企微 App 扫一扫登录浏览器坐席端"的体验。 # # 完整流程: # ┌─────────┐ create ┌─────────────┐ scan ┌──────────┐ # │ 浏览器 │ ───────→ │ ticket(120s)│ ←───── │ 企微 App │ # │ 前端 │ ←─────── │ +OAuth URL │ OAuth │ 扫码授权 │ # └─────────┘ qrcode_url └─────────────┘ code └──────────┘ # │ │ │ # │ poll │ scan │ # │ waiting/scanned │ 写 scan:{ticket} │ # │ ↓ │ # │ ┌────────────────┐ │ # │ │ 已登录坐席(企微)│ confirm │ # │ │ 点"确认登录"按钮 │ ────────→ │ # │ └────────────────┘ │ # │ │ │ # │ poll │ confirm │ # │ confirmed+token │ 写 confirm:{ticket} │ # ↓ ↓ │ # 拿到 token,跳坐席端主页 │ # # 端点列表(4 个): # POST /api/auth_qrcode/create — 浏览器前端生成 ticket # GET /api/auth_qrcode/poll/{ticket} — 前端轮询扫码状态 # POST /api/auth_qrcode/scan — 企微 OAuth2 回调(接收 code) # POST /api/auth_qrcode/confirm — 当前登录坐席点确认 # # 鉴权说明: # - create / scan / poll: 无需登录(浏览器刚加载登录页,用户未登录) # - confirm: 需要已登录坐席点确认(角色: agent / admin) # - 票据状态全部存 Redis,TTL 到期自动失效,无 DB 表 # ============================================================================= import logging import time import secrets as secrets_mod from typing import Optional import redis.asyncio as aioredis from fastapi import APIRouter, Depends, Path, Query, Request 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, QrcodeConfirmResponse, QrcodeCreateResponse, QrcodePollResponse, QrcodeScanRequest, QrcodeScanResponse, ) from app.services.qrcode_service import QrcodeService from app.utils.response import AppException, success_response logger = logging.getLogger(__name__) # 创建路由器 # prefix="/auth_qrcode" + tags=["扫码登录"] 用于 Swagger 分组 router = APIRouter(prefix="/auth_qrcode", tags=["扫码登录"]) def _get_qrcode_service(redis_client: aioredis.Redis) -> QrcodeService: """工厂函数: 构造扫码登录业务服务。 拆出来便于测试时 monkey-patch,以及后续接入 DI。 """ return QrcodeService(redis_client) # -------------------------------------------------------------------------- # POST /api/auth_qrcode/create — 创建扫码登录票据 # -------------------------------------------------------------------------- @router.post("/create", response_model=None) async def create_qrcode( redis_client: aioredis.Redis = Depends(dep_redis), ): """创建扫码登录票据。 无需鉴权(用户尚未登录,正在登录页)。 返回 ticket + 企微 OAuth2 授权 URL,前端渲染二维码。 Returns: Dict: 统一响应格式,data 字段是 QrcodeCreateResponse """ try: service = _get_qrcode_service(redis_client) result = await service.create_ticket() return success_response(data={ "ticket": result["ticket"], "qrcode_url": result["qrcode_url"], "qrcode_png_base64": result["qrcode_png_base64"], "expires_in": result["expires_in"], "expires_at": result["expires_at"].isoformat(), }) except Exception as e: logger.error(f"创建扫码票据异常: {e}", exc_info=True) raise AppException(1005, f"创建扫码票据失败: {str(e)}") # -------------------------------------------------------------------------- # GET /api/auth_qrcode/poll/{ticket} — 前端轮询扫码状态 # -------------------------------------------------------------------------- @router.get("/poll/{ticket}", response_model=None) async def poll_qrcode( ticket: str = Path(..., description="扫码登录票据"), redis_client: aioredis.Redis = Depends(dep_redis), ): """轮询扫码状态。 无需鉴权(浏览器未登录态访问)。 状态机: - waiting: ticket 有效,等待扫码 - scanned: 已扫码,等待 confirm - confirmed: 已确认,返回 token - expired: ticket 过期/不存在 Returns: Dict: 统一响应格式,data 字段是 QrcodePollResponse """ try: service = _get_qrcode_service(redis_client) result = await service.get_poll_state(ticket) return success_response(data={ "status": result["status"], "employee_id": result.get("employee_id"), "name": result.get("name"), "token": result.get("token"), }) except Exception as e: logger.error(f"轮询扫码状态异常: ticket={ticket[:8]}..., error={e}", exc_info=True) raise AppException(1005, f"轮询扫码状态失败: {str(e)}") # -------------------------------------------------------------------------- # GET|POST /api/auth_qrcode/scan — 企微 OAuth code 回调 # -------------------------------------------------------------------------- @router.api_route("/scan", methods=["GET", "POST"], response_model=None) async def scan_qrcode( request: Request, 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), db: AsyncSession = Depends(get_db), ): """处理企微 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 调用。 """ try: # 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) # ========================================================================== # 扫码后自动确认（auto-confirm） # ========================================================================== # 原设计：requester 需已登录坐席调 /confirm 来授权新登录 # 问题：首次登录时电脑端无人登录，没有合法 current_user 可调 confirm # 修正：扫码即确认，直接为扫码的企微用户签发 token # ========================================================================== from app.services.token_service import TokenService from app.services.role_mapping_service import RoleMappingService import json from datetime import datetime token_service = TokenService(redis_client) # 获取用户的真实角色（而非写死 agent） role_service = RoleMappingService(db) user_roles = await role_service.get_user_roles(result["employee_id"]) logger.info( f"扫码登录角色: employee_id={result['employee_id']}, " f"roles={user_roles}" ) auto_token = await token_service.create_token( employee_id=result["employee_id"], name=result["name"], roles=user_roles, avatar=result.get("avatar", ""), login_source="qrcode_scan", ) confirm_payload = { "token": auto_token, "confirmed_at": datetime.now().isoformat(), "roles": user_roles, "employee_id": result["employee_id"], "name": result["name"], } CONFIRM_TTL = 60 await redis_client.setex( f"qrcode:confirm:{final_ticket}", CONFIRM_TTL, json.dumps(confirm_payload, ensure_ascii=False), ) logger.info( f"扫码自动确认: ticket={final_ticket[:8]}..., " f"employee_id={result['employee_id']}, name={result['name']}" ) # GET 请求(企微 OAuth 回调)→ 已自动确认，显示成功页 + 自动关闭 from fastapi.responses import HTMLResponse if final_code is not None and final_ticket is not None and body is None: user_name = result.get('name', '') # ============================================================ # 生成企微 JS-SDK 签名（用于 wx.closeWindow() 关闭当前页面） # ============================================================ # 签名算法: sha1(jsapi_ticket=X&noncestr=X×tamp=X&url=X) # url 必须是企微 webview 中当前页面的完整 URL（含 query，不含 #hash） jsapi_signature = "" jsapi_timestamp = 0 jsapi_nonce = "" jsapi_appid = getattr(settings, "wecom_corp_id", "") current_url = "" # 用于调试显示和 JSAPI 签名 URL try: from app.services.wecom_service import WecomService wecom_svc = WecomService(redis_client) jsapi_ticket = await wecom_svc.get_jsapi_ticket() # 当前页面 URL（不含 # 后面部分） # ⚠️ 关键：request.url 默认用内部 HTTP 协议（Nginx→后端是 HTTP） # 但企微 WebView 看到的是 HTTPS，协议不一致会导致签名校验失败 # 必须从 X-Forwarded-Proto 头重建正确的 HTTPS URL scheme = request.headers.get("x-forwarded-proto", "https") host = request.headers.get("host", "") path = request.url.path query = str(request.url.query) if request.url.query else "" current_url = f"{scheme}://{host}{path}" if query: current_url += f"?{query}" jsapi_timestamp = int(time.time()) jsapi_nonce = secrets_mod.token_hex(8) jsapi_signature = wecom_svc.generate_jsapi_signature( ticket=jsapi_ticket, nonce_str=jsapi_nonce, timestamp=jsapi_timestamp, url=current_url, ) logger.info( f"JS-SDK 签名生成成功: url={current_url}, " f"timestamp={jsapi_timestamp}, appid={jsapi_appid}" ) try: await wecom_svc.close() except Exception: pass except Exception as e: logger.warning(f"JS-SDK 签名生成失败（降级为 WeixinJSBridge）: {e}") html = f""" 登录成功 - IT智能服务台

登录成功

已自动确认登录

你好，{user_name}
请返回电脑端查看

页面即将自动关闭 · 税友集团
""" return HTMLResponse(content=html, status_code=200) # POST 模式:返回 JSON return success_response(data={ "success": result["success"], "message": result["message"], }) except ValueError as ve: # 票据过期/不存在 → 业务错误 logger.warning(f"扫码业务错误: {ve}") raise AppException(1003, str(ve)) except Exception as e: logger.error(f"扫码处理异常: error={e}", exc_info=True) raise AppException(1005, f"扫码处理失败: {str(e)}") # -------------------------------------------------------------------------- # POST /api/auth_qrcode/confirm — 当前已登录坐席确认授权 # -------------------------------------------------------------------------- @router.post("/confirm", response_model=None) 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), ): """处理当前已登录坐席的扫码确认授权。 需要鉴权: 只有已登录的坐席/管理员能确认授权。 把扫码用户身份变成可登录 Token(roles=['agent']), 写 Redis confirm:{ticket},前端 poll 拿到后跳坐席主页。 otp_code: admin 场景下可选,Phase 1.1 仅记录日志, 真实 OTP 校验留给 Phase 2.1(参考 agents.py:272-274 的 totp.verify)。 Args: body: 包含 ticket 和 otp_code(可选) current_user: 当前已登录用户(由 get_current_user 注入) redis_client: Redis 客户端 Returns: Dict: 统一响应格式,data 字段是 QrcodeConfirmResponse """ try: service = _get_qrcode_service(redis_client) result = await service.process_confirm( ticket=body.ticket, current_user_id=current_user.employee_id, current_user_name=current_user.name, current_roles=current_user.roles, otp_code=body.otp_code, ) # 同步头像：扫码时已从企微API拿到最新头像URL，这里落库 + 清缓存 # （要求 A：确保所有登录路径刷新头像；头像更新失败不阻塞登录） confirm_avatar = result.get("avatar", "") if confirm_avatar: try: from app.services.avatar_service import sync_employee_avatar await sync_employee_avatar( db, redis_client, result["employee_id"], confirm_avatar ) except Exception as e: logger.warning( f"扫码确认同步头像失败（不阻塞）: " f"employee_id={result.get('employee_id')}, error={e}" ) # 记录扫码登录日志（成功） 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"], "name": result["name"], "roles": result["roles"], "require_otp": result.get("require_otp"), }) except ValueError as ve: # 票据过期/未扫码 → 业务错误 logger.warning( f"扫码确认业务错误: ticket={body.ticket[:8]}..., " f"current_user={current_user.employee_id}, error={ve}" ) raise AppException(1003, str(ve)) except Exception as e: logger.error( f"扫码确认异常: ticket={body.ticket[:8]}..., " f"current_user={current_user.employee_id}, error={e}", exc_info=True, ) raise AppException(1005, f"扫码确认失败: {str(e)}")