小米云设备令牌提取器解锁智能家居生态集成的核心技术架构【免费下载链接】Xiaomi-cloud-tokens-extractorThis tool retrieves tokens for all devices connected to Xiaomi cloud and encryption keys for BLE devices.项目地址: https://gitcode.com/gh_mirrors/xia/Xiaomi-cloud-tokens-extractorXiaomi-cloud-tokens-extractor是一款专业的开源工具专门用于从小米云平台安全提取连接设备的访问令牌和BLE设备的加密密钥。该项目通过模块化架构设计和多重认证机制为开发者提供了与小米智能家居生态系统深度集成的技术基础解决了设备接入和本地控制的关键技术难题。技术架构深度解析抽象基类设计模式项目的核心采用了抽象基类设计模式通过XiaomiCloudConnector类定义了统一的设备连接接口实现了认证逻辑与设备操作的完全分离。这种设计模式使得系统具有良好的扩展性可以轻松支持新的认证方式而不影响核心业务逻辑。class XiaomiCloudConnector(ABC): def __init__(self): self._agent self.generate_agent() self._device_id self.generate_device_id() self._session requests.session() self._ssecurity None self.userId None self._serviceToken None abstractmethod def login(self) - bool: pass def get_homes(self, country): # 获取家庭列表实现 pass def get_devices(self, country, home_id, owner_id): # 获取设备列表实现 pass双重认证机制实现项目支持两种主流的认证方式分别通过不同的子类实现密码认证模式PasswordXiaomiCloudConnector支持邮箱、手机号或用户ID登录自动处理验证码和二次验证流程完整的错误处理和重试机制二维码认证模式QrCodeXiaomiCloudConnector通过小米官方扫码方式登录内置HTTP服务器展示二维码图片支持本地文件系统备用方案加密通信安全层小米云API采用了RC4加密算法进行数据传输保护项目通过encrypt_rc4和decrypt_rc4方法实现了完整的加密通信协议staticmethod def encrypt_rc4(password, payload): r ARC4.new(base64.b64decode(password)) r.encrypt(bytes(1024)) return base64.b64encode(r.encrypt(payload.encode())).decode() staticmethod def decrypt_rc4(password, payload): r ARC4.new(base64.b64decode(password)) r.encrypt(bytes(1024)) return r.encrypt(base64.b64decode(payload))加密流程采用小米官方协议标准包括nonce生成、签名计算和参数加密等多个安全步骤确保通信过程的安全性。多区域服务器支持项目支持全球多个小米云服务器区域通过统一的API接口适配不同地区的服务SERVERS [cn, de, us, ru, tw, sg, in, i2]每个区域对应不同的API端点cn: 中国大陆服务器de: 欧洲服务器德国us: 北美服务器ru: 俄罗斯服务器tw: 台湾服务器sg: 新加坡服务器in: 印度服务器i2: 国际服务器设备信息获取与解析设备数据结构工具能够获取并解析完整的设备信息包括字段名称数据类型描述did字符串设备唯一标识符name字符串设备显示名称model字符串设备型号token字符串设备访问令牌localip字符串设备本地IP地址mac字符串设备MAC地址beaconkey字符串BLE设备加密密钥BLE设备密钥提取对于支持蓝牙的设备工具能够获取加密密钥这是实现本地蓝牙控制的关键def get_beaconkey(self, country, did): url self.get_api_url(country) /v2/device/blt_get_beaconkey params { data: {did: did ,pdid:1} } return self.execute_api_call_encrypted(url, params)部署与集成方案命令行使用模式项目提供了灵活的命令行接口支持交互式和非交互式两种运行模式# 交互式模式推荐 python3 token_extractor.py # 非交互式模式自动化脚本 python3 token_extractor.py --non_interactive --username userexample.com --password your_password --server cn --output devices.jsonDocker容器化部署项目提供完整的Docker支持便于在各种环境中快速部署# 使用Docker运行 bash (curl -L https://github.com/PiotrMachowski/Xiaomi-cloud-tokens-extractor/raw/master/run_docker.sh)Dockerfile采用Alpine Linux作为基础镜像确保镜像体积最小化FROM python:3-alpine WORKDIR /usr/src/app COPY requirements.txt ./ RUN apk add build-base RUN pip3 install --no-cache-dir -r requirements.txt COPY token_extractor.py ./ ENTRYPOINT [ python, ./token_extractor.py ]Home Assistant集成作为智能家居自动化平台的核心组件项目与Home Assistant无缝集成# 在Home Assistant SSH Web Terminal中运行 bash (curl -L https://github.com/PiotrMachowski/Xiaomi-cloud-tokens-extractor/raw/master/run.sh)技术实现细节会话管理与状态保持项目通过requests.session()维护会话状态确保认证后的所有API调用都携带正确的cookies和headersdef __init__(self): self._session requests.session() self._ssecurity None self.userId None self._serviceToken None用户代理与设备ID生成为模拟真实设备行为项目动态生成用户代理和设备IDstaticmethod def generate_agent(): agent_id .join( map(lambda i: chr(i), [random.randint(65, 69) for _ in range(13)]) ) random_text .join(map(lambda i: chr(i), [random.randint(97, 122) for _ in range(18)])) return f{random_text}-{agent_id} APP/com.xiaomi.mihome APTV/10.5.201 staticmethod def generate_device_id(): return .join(map(lambda i: chr(i), [random.randint(97, 122) for _ in range(6)]))错误处理与容错机制项目实现了完善的错误处理机制包括网络异常重试自动处理网络超时和连接错误验证码处理支持图形验证码识别流程二次验证完整的邮箱二次验证流程API限流处理优雅处理API调用频率限制企业级应用场景批量设备管理基于项目的核心架构可以构建企业级设备管理系统class EnterpriseDeviceManager: def __init__(self, config_file): self.connectors {} self.load_config(config_file) def batch_token_extraction(self, accounts): 批量提取多个账号的设备令牌 results {} for account in accounts: connector self.create_connector(account) if connector.login(): devices self.get_all_devices(connector) results[account[username]] devices return results def device_inventory_report(self): 生成设备清单报告 inventory [] for server in SERVERS: devices self.get_server_devices(server) inventory.extend(devices) return self.generate_report(inventory)监控与告警系统集成将设备状态监控与现有监控系统集成class DeviceMonitoringIntegration: def __init__(self, prometheus_client): self.prometheus prometheus_client self.device_status prometheus_client.Gauge( xiaomi_device_online_status, 小米设备在线状态, [server, device_id, device_type] ) def update_metrics(self, connector): 更新监控指标 servers get_servers_to_check() for server in servers: homes connector.get_homes(server) for home in homes.get(result, {}).get(homelist, []): devices connector.get_devices(server, home[id], home[owner]) for device in devices.get(result, {}).get(device_info, []): status 1 if device.get(online, False) else 0 self.device_status.labels( serverserver, device_iddevice.get(did, unknown), device_typedevice.get(model, unknown) ).set(status)安全最佳实践令牌安全管理最小权限原则仅获取必要的设备访问权限定期轮换建议定期更新设备令牌安全存储使用加密存储方案保存敏感信息访问审计记录所有令牌获取操作网络通信安全HTTPS加密所有通信都通过HTTPS加密传输证书验证启用完整的SSL证书验证请求签名所有API请求都包含数字签名防重放攻击使用nonce机制防止请求重放性能优化策略并发处理优化import concurrent.futures from typing import List, Dict class ConcurrentDeviceFetcher: def __init__(self, max_workers: int 5): self.max_workers max_workers def fetch_multiple_servers(self, connector, servers: List[str]) - Dict[str, List]: 并发获取多服务器设备信息 results {} with concurrent.futures.ThreadPoolExecutor(max_workersself.max_workers) as executor: future_to_server { executor.submit(self.fetch_server_devices, connector, server): server for server in servers } for future in concurrent.futures.as_completed(future_to_server): server future_to_server[future] try: results[server] future.result(timeout30) except Exception as e: print(fFailed to fetch devices from {server}: {e}) results[server] [] return results缓存机制实现from datetime import datetime, timedelta import json class DeviceCache: def __init__(self, cache_ttl: int 300): self.cache_ttl cache_ttl self.cache {} self.timestamps {} def get_cached_devices(self, server: str, account_id: str): 获取缓存的设备信息 cache_key f{server}_{account_id} current_time datetime.now() if cache_key in self.cache: last_update self.timestamps[cache_key] if current_time - last_update timedelta(secondsself.cache_ttl): return self.cache[cache_key] return None def update_cache(self, server: str, account_id: str, devices: List): 更新设备缓存 cache_key f{server}_{account_id} self.cache[cache_key] devices self.timestamps[cache_key] datetime.now()故障排除与调试常见问题解决方案问题类型可能原因解决方案登录失败账号密码错误确认使用小米云账号而非Roborock应用账号验证码错误网络环境限制禁用Cloudflare DNS和广告拦截器二次验证失败邮箱验证码未收到检查垃圾邮件文件夹API调用限制请求频率过高等待24小时后重试每日限制3-5次网络连接问题防火墙或代理设置检查网络配置使用QR码认证替代调试日志配置项目内置了详细的日志系统支持多级别日志输出# 启用调试日志 python3 token_extractor.py --log_level DEBUG # 日志级别选项 CRITICAL, FATAL, ERROR, WARN, WARNING, INFO, DEBUG, NOTSET扩展开发指南自定义认证方式通过继承XiaomiCloudConnector基类可以实现新的认证方式class CustomAuthConnector(XiaomiCloudConnector): def __init__(self, api_key: str): super().__init__() self.api_key api_key def login(self) - bool: 实现自定义认证逻辑 # 自定义认证流程 # 设置必要的认证参数 self._ssecurity self.get_ssecurity_from_api_key() self.userId self.get_user_id_from_api_key() return True def get_ssecurity_from_api_key(self) - str: 从API密钥生成ssecurity # 实现自定义安全令牌生成逻辑 pass数据输出格式扩展支持多种数据输出格式便于与其他系统集成class OutputFormatter: staticmethod def format_json(devices: List[Dict]) - str: JSON格式输出 return json.dumps(devices, indent2, ensure_asciiFalse) staticmethod def format_csv(devices: List[Dict]) - str: CSV格式输出 import csv import io output io.StringIO() writer csv.DictWriter(output, fieldnames[server, home_id, device_id, name, model, token, ip]) writer.writeheader() for device in devices: writer.writerow({ server: device.get(server), home_id: device.get(home_id), device_id: device.get(did), name: device.get(name), model: device.get(model), token: device.get(token), ip: device.get(localip) }) return output.getvalue() staticmethod def format_yaml(devices: List[Dict]) - str: YAML格式输出 import yaml return yaml.dump(devices, default_flow_styleFalse, allow_unicodeTrue)技术栈与依赖管理核心依赖库依赖包版本要求功能描述requests最新HTTP客户端库pycryptodome最新加密算法实现Pillow最新图像处理验证码显示colorama最新终端颜色输出环境兼容性项目支持多种运行环境操作系统: Windows, Linux, macOSPython版本: Python 3.6容器平台: Docker, Kubernetes云环境: 所有主流云平台总结与展望Xiaomi-cloud-tokens-extractor作为小米智能家居生态的技术桥梁为开发者提供了稳定可靠的设备接入解决方案。通过其模块化架构设计、完善的安全机制和灵活的部署选项项目不仅解决了设备令牌提取的技术难题更为智能家居自动化、企业设备管理和物联网平台开发提供了坚实的技术基础。随着小米生态系统的不断扩展该项目将继续演进支持更多的设备类型、认证方式和集成场景为智能家居开发者社区提供持续的技术支持。【免费下载链接】Xiaomi-cloud-tokens-extractorThis tool retrieves tokens for all devices connected to Xiaomi cloud and encryption keys for BLE devices.项目地址: https://gitcode.com/gh_mirrors/xia/Xiaomi-cloud-tokens-extractor创作声明：本文部分内容由AI辅助生成（AIGC），仅供参考