"""JSONL-only bridge for the Windows EchoEar flashing app. ``ports`` and ``preview`` never open a serial port. ``flash`` accepts only a short-lived, one-use preview session and an explicit execute=true acknowledgement. No credentials, setup passwords or raw subprocess output are sent to the GUI. """ from __future__ import annotations import argparse import contextlib from datetime import datetime, timezone import hashlib import hmac import io import json import os from pathlib import Path import re import secrets import subprocess import sys import time import device_flash import device_prepare PROJECT = Path(__file__).resolve().parents[1] DEFAULT_OUTPUT = PROJECT / "backups" / "gui" SESSION_TTL = 30 * 60 SESSION_FIELDS = {"schema_version", "created_at", "created_epoch", "expires_epoch", "credential_path", "build_dir", "mode", "expected_mac", "port", "factory_dir", "backup_dir", "input_hashes", "plan_hash", "signature"} PREVIEW_FIELDS = {"credential_path", "build_dir", "mode", "expected_mac", "port", "output_root"} FLASH_FIELDS = {"session_file", "execute", "port", "expected_mac"} MESSAGES = { "invalid_request": "输入无效。请检查烧录信息 JSON、固件目录、COM 端口和设备 MAC。", "invalid_credential": "烧录信息 JSON 无效，请从线上设备管理后台重新下载。", "invalid_build": "固件文件校验失败。请选择完整的 EchoEar ESP32-S3 16MB 构建目录。", "preparation_failed": "初始化信息生成失败。请检查 ESP-IDF Python 环境和输出目录权限，再重新预览。", "stale_preview": "预览已过期、已使用或输入文件发生变化。请重新预览后再烧录。", "confirmation_required": "尚未确认执行。请在界面中确认清除本地 Wi-Fi、语音、会话和 OTA 状态后再开始。", "port_mismatch": "设备端口或 MAC 与预览不一致。请重新预览。", "ports_failed": "串口列表读取失败。请检查 Python 环境是否安装 pyserial。", "flash_failed": "烧录未完成。请检查 USB 连接、设备 MAC 和串口占用；已生成的备份及诊断记录已保留。重新尝试前请重新预览。", "internal_error": "操作失败，未向界面输出敏感诊断内容。请检查运行环境和文件权限。", } STAGES = { "version": ("checking", "检查 esptool 版本"), "read_mac": ("checking", "连接设备并核对芯片与 MAC"), "flash_id": ("checking", "核对 16MB Flash 容量"), "get_security_info": ("checking", "核对设备安全配置"), "read_flash": ("backup", "正在完整备份 16MB Flash，请保持 USB 连接"), "write_flash": ("writing", "备份完成，正在写入，请保持 USB 连接"), "verify_flash": ("verifying", "正在校验设备中写入的数据"), } class BridgeError(ValueError): def __init__(self, code): self.code = code super().__init__(MESSAGES[code]) def require(condition, code="invalid_request"): if not condition: raise BridgeError(code) def canonical(value): return json.dumps(value, sort_keys=True, separators=(",", ":"), ensure_ascii=True).encode("utf-8") def digest(value): return hashlib.sha256(canonical(value)).hexdigest() def safe_path(value, code="invalid_request"): require(isinstance(value, str) and value and len(value) < 4096 and "\0" not in value, code) return Path(value).expanduser().resolve() def read_object(path, code="invalid_request"): try: require(path.is_file() and path.stat().st_size <= 65536, code) value = json.loads(path.read_text(encoding="utf-8-sig")) require(isinstance(value, dict), code) return value except (OSError, ValueError, UnicodeError): raise BridgeError(code) from None def normalize_port(value): require(isinstance(value, str) and re.fullmatch(r"COM[1-9][0-9]{0,4}", value, re.I)) return value.upper() def hidden_run(args, **kwargs): if os.name == "nt": kwargs["creationflags"] = kwargs.get("creationflags", 0) | subprocess.CREATE_NO_WINDOW return subprocess.run(args, **kwargs) class _HiddenSubprocess: """Keep existing preparation/ACL helpers quiet without changing their source.""" def run(self, args, **kwargs): return hidden_run(args, **kwargs) def __getattr__(self, name): return getattr(subprocess, name) @contextlib.contextmanager def hidden_preparation(): original = device_prepare.subprocess device_prepare.subprocess = _HiddenSubprocess() try: yield finally: device_prepare.subprocess = original def input_snapshot(credential, build, factory=None): """Include every reference build image even when retaining existing firmware.""" manifest = device_flash.load_json(build / "flasher_args.json") files = [credential, build / "flasher_args.json"] for name in device_flash.REGIONS: entry = device_flash.object_field(manifest, name) files.append(device_flash.contained_file(build, entry.get("file"))) if factory is not None: files.extend(factory / name for name in ("manifest.json", "ble_factory.bin", "factory.csv", "setup-label.json")) return {str(path): hashlib.sha256(path.read_bytes()).hexdigest() for path in sorted(set(files))} def phase(emit, stage, message): emit({"event": "phase", "stage": stage, "message": message}) def enumerate_ports(request, emit): require(not request) try: from serial.tools import list_ports ports = [] for item in list_ports.comports(): if not re.fullmatch(r"COM[1-9][0-9]{0,4}", item.device, re.I): continue ports.append({"port": item.device.upper(), **{ name: getattr(item, name, None) for name in ("description", "hwid", "serial_number", "manufacturer", "product", "vid", "pid")}}) ports.sort(key=lambda item: int(item["port"][3:])) except Exception: raise BridgeError("ports_failed") from None emit({"event": "ports", "ports": ports}) def preview(request, emit): require(set(request) <= PREVIEW_FIELDS) phase(emit, "validating", "离线检查初始化信息和固件文件，不连接设备") credential = safe_path(request.get("credential_path")) build = safe_path(request.get("build_dir", str(PROJECT / "build"))) mode = request.get("mode", "identity") require(mode in ("identity", "full")) port = normalize_port(request.get("port")) try: mac = device_flash.normalize_mac(request.get("expected_mac")) except ValueError: raise BridgeError("invalid_request") from None output = safe_path(request.get("output_root", str(DEFAULT_OUTPUT))) try: require(credential.is_file() and credential.stat().st_size <= 16384, "invalid_credential") credential_hash = hashlib.sha256(credential.read_bytes()).hexdigest() info = device_prepare.read_credential(credential) except (ValueError, OSError): raise BridgeError("invalid_credential") from None try: original_inputs = input_snapshot(credential, build) require(original_inputs[str(credential)] == credential_hash, "stale_preview") device_flash.build_plan(build, reset_runtime=True) except BridgeError: raise except (ValueError, OSError): raise BridgeError("invalid_build") from None phase(emit, "preparing", "生成本机私有初始化文件和烧录预览") now = time.time() created_at = datetime.fromtimestamp(now, timezone.utc).isoformat() directory = output / (datetime.fromtimestamp(now, timezone.utc).strftime("%Y%m%dT%H%M%SZ-") + secrets.token_hex(6)) factory, backup = directory / "factory", directory / "backup" try: with hidden_preparation(): device_prepare.private_directory(directory) device_prepare.prepare(credential, factory, mac) plan = device_flash.build_plan(build, factory, mode == "identity", True, mac) require(original_inputs == input_snapshot(credential, build), "stale_preview") frozen_inputs = input_snapshot(credential, build, factory) session = {"schema_version": 1, "created_at": created_at, "created_epoch": now, "expires_epoch": now + SESSION_TTL, "credential_path": str(credential), "build_dir": str(build), "mode": mode, "expected_mac": mac, "port": port, "factory_dir": str(factory), "backup_dir": str(backup), "input_hashes": frozen_inputs, "plan_hash": digest(plan.summary())} integrity_key = secrets.token_bytes(32) key_path = directory / "integrity.key" with key_path.open("xb") as stream: stream.write(integrity_key) os.chmod(key_path, 0o600) session["signature"] = hmac.new(integrity_key, canonical(session), hashlib.sha256).hexdigest() session_file = directory / "session.json" device_prepare.private_json(session_file, session) except BridgeError: raise except Exception: raise BridgeError("preparation_failed") from None summary = {"mode": mode, "port": port, "expected_mac": mac, "device_id": info["device_id"], "model": info["model"], "origin": info["origin"], "key_version": info["key_version"], "firmware_version": None if mode == "identity" else plan.application["firmware_version"], "reference_firmware_version": plan.application["firmware_version"], "reset_runtime": True, "runtime_effect": "清除本地 Wi-Fi、语音、会话和 OTA 状态；完成后须重新蓝牙配网。", "regions": [{key: value for key, value in image.summary().items() if key != "source"} for image in plan.images], "factory_dir": str(factory), "backup_dir": str(backup), "created_at": created_at, "expires_in_seconds": SESSION_TTL} phase(emit, "ready", "预览完成。确认清除配网信息后方可开始烧录") emit({"event": "preview", "session_file": str(session_file), "summary": summary}) def load_session(path): require(path.name == "session.json", "stale_preview") session = read_object(path, "stale_preview") require(set(session) == SESSION_FIELDS and session.get("schema_version") == 1, "stale_preview") signature = session.pop("signature") try: key = (path.parent / "integrity.key").read_bytes() require(len(key) == 32 and isinstance(signature, str), "stale_preview") require(hmac.compare_digest(signature, hmac.new(key, canonical(session), hashlib.sha256).hexdigest()), "stale_preview") require(type(session["created_epoch"]) in (int, float) and type(session["expires_epoch"]) in (int, float), "stale_preview") now = time.time() require(session["created_epoch"] - 60 <= now < session["expires_epoch"], "stale_preview") require(session["expires_epoch"] - session["created_epoch"] == SESSION_TTL, "stale_preview") require(session["mode"] in ("identity", "full"), "stale_preview") require(normalize_port(session["port"]) == session["port"], "stale_preview") require(device_flash.normalize_mac(session["expected_mac"]) == session["expected_mac"], "stale_preview") require(safe_path(session["factory_dir"]) == path.parent / "factory", "stale_preview") require(safe_path(session["backup_dir"]) == path.parent / "backup", "stale_preview") require(not (path.parent / "execute.json").exists(), "stale_preview") except (OSError, ValueError, TypeError, KeyError): raise BridgeError("stale_preview") from None return session def flash(request, emit): require(set(request) <= FLASH_FIELDS) require(request.get("execute") is True, "confirmation_required") path = safe_path(request.get("session_file"), "stale_preview") session = load_session(path) if "port" in request: require(normalize_port(request["port"]) == session["port"], "port_mismatch") if "expected_mac" in request: try: require(device_flash.normalize_mac(request["expected_mac"]) == session["expected_mac"], "port_mismatch") except device_flash.FlashError: raise BridgeError("port_mismatch") from None phase(emit, "validating", "重新核对已确认的文件、端口、MAC 和烧录范围") credential = safe_path(session["credential_path"], "stale_preview") build = safe_path(session["build_dir"], "stale_preview") factory = safe_path(session["factory_dir"], "stale_preview") try: require(input_snapshot(credential, build, factory) == session["input_hashes"], "stale_preview") device_prepare.read_credential(credential) plan = device_flash.build_plan(build, factory, session["mode"] == "identity", True, session["expected_mac"]) require(digest(plan.summary()) == session["plan_hash"], "stale_preview") require(input_snapshot(credential, build, factory) == session["input_hashes"], "stale_preview") # Exclusive creation consumes the session before even the first probe. device_prepare.private_json(path.parent / "execute.json", {"started_at": datetime.now(timezone.utc).isoformat()}) except (ValueError, OSError): raise BridgeError("stale_preview") from None def runner(args, **kwargs): for argument in args: stage = STAGES.get(argument.replace("-", "_")) if stage: phase(emit, *stage) break return hidden_run(args, **kwargs) try: # Existing CLI progress and raw tool output are intentionally not GUI logs. with hidden_preparation(), contextlib.redirect_stdout(io.StringIO()): result = device_flash.execute_plan(plan, python=sys.executable, port=session["port"], expect_mac=session["expected_mac"], backup_dir=Path(session["backup_dir"]), runner=runner) except Exception: raise BridgeError("flash_failed") from None result["next_step"] = "烧录和校验完成。请手动重启设备，然后在家庭端重新蓝牙配网和绑定。" emit({"event": "complete", "result": result}) def main(argv=None, output=None): output = output or sys.stdout def emit(value): output.write(json.dumps(value, ensure_ascii=True, separators=(",", ":")) + "\n") output.flush() parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("command", choices=("ports", "preview", "flash")) parser.add_argument("--request", required=True, type=Path) try: args = parser.parse_args(argv) request = read_object(args.request) {"ports": enumerate_ports, "preview": preview, "flash": flash}[args.command](request, emit) return 0 except BridgeError as exc: emit({"event": "error", "code": exc.code, "message": MESSAGES[exc.code]}) return 2 except Exception: emit({"event": "error", "code": "internal_error", "message": MESSAGES["internal_error"]}) return 2 if __name__ == "__main__": raise SystemExit(main())