317 lines
15 KiB
Python
317 lines
15 KiB
Python
|
"""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())
|