# ============================================================================= # 企微IT智能服务台 — Mock适配器（开发期使用） # ============================================================================= # 说明： # 1. 在 Mock 模式下（EXT_MOCK_MODE=True），所有外部系统查询 # 返回预置的 Mock 数据，不调用任何真实 API # 2. Mock 数据覆盖 P0 场景（终端查询、安全状态、VPN在线） # 3. 凭证未配置时自动降级到 MockAdapter，保证开发期无外部依赖 # # 使用方式： # EXT_MOCK_MODE=True → 所有系统走 Mock # 某系统凭证未配置 → 单个系统自动降级到 Mock（在 service.py 中处理） # ============================================================================= import logging from datetime import datetime, timedelta from typing import Any, Dict, List, Optional from app.services.external.base import ( ExternalSystemAdapter, TerminalInfo, SecurityStatus, VpnSession, ) logger = logging.getLogger(__name__) # ============================================================================= # Mock 数据工厂 # ============================================================================= def _make_mock_terminal(username: str) -> TerminalInfo: """生成 Mock 终端信息 做什么：为指定用户生成一个逼真的模拟终端数据 为什么：开发期没有真实凭证时需要终端数据支撑会话排查流程 """ return TerminalInfo( source_system="mock", terminal_id=f"mock-terminal-{username}", computer_name=f"{username.upper()}-PC01", ip_addresses=[f"192.168.{hash(username) % 255}.{100 + hash(username) % 155}"], mac_addresses=[f"00:16:3E:{hash(username) % 256:02X}:{hash(username + 'a') % 256:02X}:{hash(username + 'b') % 256:02X}"], os_version="Windows 11 专业版 23H2", is_online=True, logged_in_user=username, logged_in_user_name=_username_to_display_name(username), department=_guess_department(username), hardware_summary={ "cpu": "Intel Core i7-12700", "memory_total_gb": 16, "memory_used_gb": 8, "disk_total_gb": 512, "disk_free_gb": 128, "disk_usage_pct": 75, # 模拟磁盘使用率较高 }, last_seen=datetime.now() - timedelta(minutes=5), raw_data=None, # Mock 数据不保留原始响应 ) def _make_mock_security_status(terminal_id: str) -> SecurityStatus: """生成 Mock 安全状态 做什么：生成一个模拟的安全状态数据 为什么：开发期需要验证安全状态卡片、漏洞警告等UI渲染 """ return SecurityStatus( source_system="mock", terminal_id=terminal_id, computer_name=f"MOCK-PC01", virus_total=2, virus_uncleaned=1, vulnerabilities=[ { "name": "Microsoft Windows 安全更新 (CVE-2025-12345)", "level": "high", "description": "远程代码执行漏洞，需立即修补", "publish_time": (datetime.now() - timedelta(days=7)).isoformat(), }, { "name": "火绒安全漏洞扫描：弱密码检测", "level": "medium", "description": "账户密码强度不足，建议修改", "publish_time": (datetime.now() - timedelta(days=3)).isoformat(), }, ], high_vuln_count=1, is_isolated=False, isolation_source=None, checked_at=datetime.now(), ) def _make_mock_vpn_session(username: str) -> VpnSession: """生成 Mock VPN 会话""" return VpnSession( source_system="mock", session_id=f"mock-session-{username}", username=username, display_name=_username_to_display_name(username), remote_ip=f"1{hash(username) % 100}.{hash(username + 'r') % 256}.{hash(username + 's') % 256}.{hash(username + 't') % 256}", vpn_ip=f"10.200.{hash(username) % 255}.{100 + hash(username) % 155}", is_trusted=True, os="Windows 11", last_login=datetime.now() - timedelta(minutes=30), domain="servyou.local", ) def _username_to_display_name(username: str) -> str: """Mock 用户名转换（简单映射）""" name_map = { "songxian": "宋献", "zhangsan": "张三", "lisi": "李四", "wangwu": "王五", } return name_map.get(username, username) def _guess_department(username: str) -> str: """Mock 部门推断""" dept_map = { "songxian": "IT支持组", "zhangsan": "财务部", "lisi": "人力资源部", "wangwu": "研发部", } return dept_map.get(username, "未知部门") # ============================================================================= # MockAdapter 实现 # ============================================================================= class MockAdapter(ExternalSystemAdapter): """Mock 适配器 — 开发期替代所有外部系统 做什么：提供逼真的模拟数据，让开发期可以不依赖任何外部系统 为什么：阶段一MVP验证、前端开发、单元测试都需要稳定的数据来源 降级规则： - 所有方法均返回 Mock 数据 - 支持常用测试用户：songxian / zhangsan / lisi / wangwu - is_available 固定返回 True（Mock 永远可用） """ @property def system_name(self) -> str: return "mock" @property def is_available(self) -> bool: """Mock 永远可用""" return True async def health_check(self) -> bool: """Mock 健康检查永远通过""" logger.debug("[MockAdapter] 健康检查 → OK（Mock模式）") return True # ── 终端查询能力 ── async def get_terminal_by_user(self, username: str) -> Optional[TerminalInfo]: """Mock：通过账号查询终端 做什么：返回预置的 Mock 终端信息 为什么：开发期坐席打开会话时需要看到终端画像 """ logger.info(f"[MockAdapter] get_terminal_by_user({username}) → Mock数据") return _make_mock_terminal(username) async def get_terminal_by_computer(self, computer_name: str) -> Optional[TerminalInfo]: """Mock：通过计算机名查询终端""" logger.info(f"[MockAdapter] get_terminal_by_computer({computer_name}) → Mock数据") # 从计算机名反推用户名（简单逻辑） username = computer_name.split("-")[0].lower() if "-" in computer_name else "songxian" return _make_mock_terminal(username) async def get_terminal_detail(self, terminal_id: str) -> Optional[TerminalInfo]: """Mock：查询终端详细信息""" logger.info(f"[MockAdapter] get_terminal_detail({terminal_id}) → Mock数据") return _make_mock_terminal("songxian") # ── 安全能力 ── async def get_security_status(self, terminal_id: str) -> Optional[SecurityStatus]: """Mock：获取安全状态""" logger.info(f"[MockAdapter] get_security_status({terminal_id}) → Mock数据") return _make_mock_security_status(terminal_id) async def isolate_terminal(self, terminal_id: str, reason: str) -> bool: """Mock：隔离终端（Mock 模式仅记录日志）""" logger.warning( f"[MockAdapter] 隔离终端（Mock，不真实执行）: " f"terminal={terminal_id}, reason={reason}" ) return True # Mock 永远返回成功 async def unisolate_terminal(self, terminal_id: str) -> bool: """Mock：解除隔离""" logger.warning( f"[MockAdapter] 解除隔离（Mock，不真实执行）: terminal={terminal_id}" ) return True # ── VPN/在线状态 ── async def get_vpn_sessions(self, username: Optional[str] = None) -> List[VpnSession]: """Mock：查询VPN在线会话""" if username: return [_make_mock_vpn_session(username)] # 返回多个 Mock 会话 return [ _make_mock_vpn_session("songxian"), _make_mock_vpn_session("zhangsan"), ] async def get_online_status(self, username: str) -> bool: """Mock：查询在线状态（Mock 永远返回 True）""" return True