from __future__ import annotations import getpass import hashlib import json import os import platform import re import sys import threading import tempfile import time import tkinter as tk import uuid from datetime import datetime, timezone from pathlib import Path from tkinter import messagebox, scrolledtext, ttk from typing import Any, Callable, Iterable, Optional, Tuple from factory_common.firmware_envelope import ( FirmwareEnvelopeError, decrypt_firmware_blob, load_firmware_decrypt_key, ) from factory_common.signing import SignatureError, verify_file_signature from factory_private_config import ( EncryptedConfigError, load_private_config_from_path, resolve_private_config_path, ) from .config import ( _load_flash_private_config, _ui_text, ) from .constants import ( DEFAULT_BAUD, DEFAULT_FLASH_MAX_ATTEMPTS, DEFAULT_FLASH_SECURE_MAX_ATTEMPTS, DEFAULT_READY_TIMEOUT_S, ) from .flash_utils import ( _extract_json_blob, _parse_offset, _sha256_hex, _value_to_int, _wipe_bytearray, build_flash_entries, collect_error_hints, detect_boot_chain_header_state, detect_secure_boot_block, extract_mac_from_output, evaluate_post_flash_status, get_efuse_flag, recommend_mode_from_json, ) from .models import ( DeviceCheckConfig, FlashEntryReport, FlashPrivateConfig, FlashReport, FlashSource, ) from .path_utils import _display_path, get_base_dir, is_frozen, resolve_python, select_bin_dir from .redact import _redact_sensitive, _sanitize_filename from .serial_utils import ensure_pyserial, list_serial_ports, pick_preferred_port from .time_utils import _now_iso from .tooling import run_command, run_esptool_api, run_tool_module class FlashGUI: def __init__(self) -> None: self.base_dir = get_base_dir() self.frozen = is_frozen() self.python_cmd: Optional[Path] = None if self.frozen else resolve_python(self.base_dir) self.run_record: Optional[dict] = None self.current_device_id: Optional[str] = None self.device_id_hint: Optional[str] = None self._post_flash_reset_done = False self.ready_timeout_s = DEFAULT_READY_TIMEOUT_S self.flash_max_attempts = DEFAULT_FLASH_MAX_ATTEMPTS self.flash_secure_max_attempts = DEFAULT_FLASH_SECURE_MAX_ATTEMPTS self.private_cfg_path = resolve_private_config_path(self.base_dir) if self.private_cfg_path is None or not self.private_cfg_path.exists(): raise FileNotFoundError( "缺少工厂私有配置文件（factory_private.json）。请设置环境变量 " "FACTORY_PRIVATE_CONFIG/FACTORY_PRIVATE_CONFIG_PATH 指定配置路径。" ) self.private_cfg_loaded = False self.ui_title = "产线烧录工具" self.ui_subtitle = "插入设备后点击开始，系统将自动完成流程" self.label_port = "串口" self.label_secure = "启动信息" self.label_flash_enc = "写入信息" self.label_key = "配置信息" self.label_recommend = "下一步" self.label_log = "日志" self.plain_text = "模式 A" self.app_text = "模式 C" self.safe_log = True raw_cfg: Optional[dict] = None try: raw_cfg = load_private_config_from_path(self.private_cfg_path) except EncryptedConfigError: raw_cfg = None if raw_cfg is not None: self._apply_private_config(raw_cfg) else: self.private_cfg = self._build_placeholder_config() self.root = tk.Tk() self._apply_window_icon() ui_cfg = self.private_cfg.ui self.ui_title = _ui_text(ui_cfg, "flash_title", "产线烧录工具") self.ui_subtitle = _ui_text(ui_cfg, "flash_subtitle", "插入设备后点击开始，系统将自动完成流程") self.label_port = _ui_text(ui_cfg, "label_port", "串口") self.label_secure = _ui_text(ui_cfg, "label_secure", "启动信息") self.label_flash_enc = _ui_text(ui_cfg, "label_flash_enc", "写入信息") self.label_key = _ui_text(ui_cfg, "label_key", "配置信息") self.label_recommend = _ui_text(ui_cfg, "label_recommend", "下一步") self.label_log = _ui_text(ui_cfg, "label_log", "日志") safe_log_value = ui_cfg.get("safe_log") self.safe_log = bool(safe_log_value) if isinstance(safe_log_value, bool) else True self.root.title(self.ui_title) self.root.geometry("820x620") self.root.minsize(760, 560) self.root.resizable(True, True) self.plain_text = _ui_text(ui_cfg, "flash_mode_plain", "模式 A") self.app_text = _ui_text(ui_cfg, "flash_mode_app", "模式 C") self.port_var = tk.StringVar(value="COM3") self.status_var = tk.StringVar(value="就绪。") self.recommend_var = tk.StringVar(value="插入设备后点击开始。") self.hint_var = tk.StringVar(value="") self.mode_status_var = tk.StringVar(value="") self.key_status_var = tk.StringVar(value="") self.progress_value = tk.DoubleVar(value=0.0) self.active_mode_label = "" self.suppress_ready_status = False self.checked_port: Optional[str] = None self._last_port_value = "" self.worker: Optional[threading.Thread] = None self.is_busy = False self.summary_attempted = False self.summary_ok = False self.last_rec_mode: Optional[str] = None self.last_flash_app_only: Optional[bool] = None self.secure_boot_blocked = False self.secure_boot_enabled: Optional[bool] = None self.boot_chain_present: Optional[bool] = None self.boot_chain_probe_reason: Optional[str] = None self.download_mode_disabled: Optional[bool] = None self.secure_download_enabled: Optional[bool] = None self.manual_encrypt_disabled: Optional[bool] = None self.last_flash_block_reason: Optional[str] = None self.card_port_var = tk.StringVar(value="未选择") self.card_secure_boot_var = tk.StringVar(value="未检测") self.card_flash_enc_var = tk.StringVar(value="未检测") self.card_recommend_var = tk.StringVar(value="待检测") self._build_ui() self.preflight_check() self._auto_set_port(initial=True) self._last_port_value = self._normalize_port(self.port_var.get()) self.port_var.trace_add("write", self._on_port_change) def _apply_window_icon(self) -> None: icon_path = self._resolve_icon_path() if not icon_path.exists(): return try: self.root.iconbitmap(default=str(icon_path)) except Exception: # noqa: BLE001 pass def _resolve_icon_path(self) -> Path: if self.frozen and hasattr(sys, "_MEIPASS"): return Path(getattr(sys, "_MEIPASS")) / "talkingq_logo_256x256.ico" return self.base_dir / "talkingq_logo_256x256.ico" def _build_placeholder_config(self) -> FlashPrivateConfig: device_check = DeviceCheckConfig( crypt_cnt_fields=("SPI_BOOT_CRYPT_CNT",), secure_boot_field="SECURE_BOOT_EN", secure_boot_expected=None, crypt_cnt_expect=None, flash_key_purpose_match=None, secure_boot_key_purpose_match=None, key_read_disable_field="RD_DIS", ) flash_cfg = FlashPrivateConfig( chip=os.getenv("FACTORY_FLASH_CHIP", "esp32s3"), baud=str(DEFAULT_BAUD), bin_dir=Path("bin"), layout=[], app_bin_name="", app_partition_offset="0x0", flash_args={ "flash_mode": "dio", "flash_freq": "80m", "flash_size": "keep", "before": "default-reset", "after": "no-reset", }, device_check=device_check, ui={}, ) return flash_cfg def _apply_private_config(self, raw_cfg: dict) -> None: self.private_cfg = _load_flash_private_config(self.base_dir, raw_cfg) self.private_cfg_loaded = True ui_cfg = self.private_cfg.ui if isinstance(self.private_cfg.ui, dict) else {} self.ui_title = _ui_text(ui_cfg, "flash_title", self.ui_title) self.ui_subtitle = _ui_text(ui_cfg, "flash_subtitle", self.ui_subtitle) self.label_port = _ui_text(ui_cfg, "label_port", self.label_port) self.label_secure = _ui_text(ui_cfg, "label_secure", self.label_secure) self.label_flash_enc = _ui_text(ui_cfg, "label_flash_enc", self.label_flash_enc) self.label_key = _ui_text(ui_cfg, "label_key", self.label_key) self.label_recommend = _ui_text(ui_cfg, "label_recommend", self.label_recommend) self.label_log = _ui_text(ui_cfg, "label_log", self.label_log) safe_log_value = ui_cfg.get("safe_log") if isinstance(safe_log_value, bool): self.safe_log = safe_log_value self.plain_text = _ui_text(ui_cfg, "flash_mode_plain", self.plain_text) self.app_text = _ui_text(ui_cfg, "flash_mode_app", self.app_text) if self.run_record and not self.run_record.get("firmware_version"): self.run_record["firmware_version"] = self.private_cfg.app_bin_name def _ensure_private_config(self, port: str, show_dialog: bool) -> bool: if self.private_cfg_loaded: return True if self.private_cfg_path is None or not self.private_cfg_path.exists(): if show_dialog: self._notify("error", "缺少配置", "未找到私有配置文件。") return False try: raw_cfg = load_private_config_from_path(self.private_cfg_path) except EncryptedConfigError: self.log("[错误] 检测到加密私有配置。当前仅支持明文配置文件。") if show_dialog: self._notify( "error", "不支持加密配置", "当前仅支持明文 conf/factory_private.json。", ) return False except (PermissionError, ValueError, FileNotFoundError) as exc: self.log(f"[错误] 私有配置加载失败：{exc}") if show_dialog: self._notify("error", "配置加载失败", str(exc)) return False self._apply_private_config(raw_cfg) self._refresh_key_status() if hasattr(self, "root"): self.root.after(0, lambda: self.root.title(self.ui_title)) return True def _refresh_mode_status(self) -> None: status = "运行模式：本地离线" self.root.after(0, lambda: self.mode_status_var.set(status)) def _read_chip_mac(self, port: str) -> Optional[str]: if not port: return None candidates: list[Optional[str]] = [] if self.private_cfg_loaded: candidates.append(self.private_cfg.chip) candidates.extend(["auto", None]) tried: set[Optional[str]] = set() for chip in candidates: if chip in tried: continue tried.add(chip) if chip: args = ["--chip", chip, "--port", port, "read_mac"] else: args = ["--port", port, "read_mac"] code, output = self.run_tool("esptool", args, stream=False) if code != 0: continue mac = extract_mac_from_output(output) if mac: return mac return None def _refresh_key_status(self) -> None: status = "配置签名：已启用" self.root.after(0, lambda: self.key_status_var.set(status)) def _build_ui(self) -> None: self.style = ttk.Style(self.root) for theme in ("clam", "vista"): if theme in self.style.theme_names(): self.style.theme_use(theme) break self.root.option_add("*Font", ("Segoe UI", 10)) palette = { "bg": "#eef4fb", "panel": "#f8fbff", "text": "#15233b", "muted": "#5a6b85", "accent": "#1a73e8", "accent_soft": "#d6e7ff", "border": "#c9d7ee", } self.root.configure(bg=palette["bg"]) self.style.configure(".", background=palette["bg"], foreground=palette["text"]) self.style.configure("TFrame", background=palette["bg"]) self.style.configure("Root.TFrame", background=palette["bg"]) self.style.configure("Panel.TFrame", background=palette["panel"]) self.style.configure("TLabelframe", background=palette["panel"], bordercolor=palette["border"]) self.style.configure("TLabelframe.Label", background=palette["panel"], foreground=palette["text"]) self.style.configure("TLabel", background=palette["bg"], foreground=palette["text"]) self.style.configure("Root.TLabel", background=palette["bg"], foreground=palette["text"]) self.style.configure("Panel.TLabel", background=palette["panel"], foreground=palette["text"]) self.style.configure( "Header.TLabel", font=("Segoe UI", 14, "bold"), foreground=palette["accent"], background=palette["bg"], ) self.style.configure( "SubHeader.TLabel", font=("Segoe UI", 9), foreground=palette["muted"], background=palette["bg"], ) self.style.configure( "Step.TLabel", font=("Segoe UI", 9), foreground=palette["muted"], background=palette["bg"], ) self.style.configure( "StepActive.TLabel", font=("Segoe UI", 9, "bold"), foreground=palette["accent"], background=palette["bg"], ) self.style.configure( "Action.TButton", padding=(14, 6), font=("Segoe UI", 10, "bold"), foreground=palette["text"], background=palette["accent_soft"], ) self.style.map( "Action.TButton", background=[("active", "#bfd7ff"), ("pressed", "#a8c8ff")], ) self.style.configure("TButton", padding=(10, 4), background=palette["panel"]) self.style.map("TButton", background=[("active", "#e6f0ff")]) self.style.configure("Disabled.TButton", foreground="#777777") self.style.configure("Section.TLabelframe.Label", font=("Segoe UI", 10, "bold")) self.style.configure("TEntry", fieldbackground=palette["panel"]) frame = ttk.Frame(self.root, padding=14, style="Root.TFrame") frame.pack(fill="both", expand=True) header = ttk.Frame(frame, style="Root.TFrame") header.pack(fill="x", pady=(0, 8)) ttk.Label( header, text=self.ui_title, style="Header.TLabel", ).pack(anchor="w") ttk.Label( header, text=self.ui_subtitle, style="SubHeader.TLabel", ).pack(anchor="w", pady=(2, 0)) port_section = ttk.LabelFrame(frame, text="1. 串口连接", padding=10, style="Section.TLabelframe") port_section.pack(fill="x", pady=(0, 8)) port_row = ttk.Frame(port_section, style="Panel.TFrame") port_row.pack(fill="x") ttk.Label(port_row, text="串口：", style="Panel.TLabel").pack(side="left") self.port_entry = ttk.Entry(port_row, textvariable=self.port_var, width=18) self.port_entry.pack(side="left", padx=(6, 12)) self.detect_btn = ttk.Button(port_row, text="刷新串口", command=self.on_auto_detect) self.detect_btn.pack(side="left", padx=(0, 12)) ttk.Label( port_row, text="提示：开始烧录时按住 BOOT，点一下 RESET。", style="Panel.TLabel", ).pack(side="left", padx=(10, 0)) status_row = ttk.Frame(port_section, style="Panel.TFrame") status_row.pack(fill="x", pady=(6, 0)) ttk.Label(status_row, textvariable=self.mode_status_var, style="Panel.TLabel").pack( side="left" ) ttk.Label(status_row, textvariable=self.key_status_var, style="Panel.TLabel").pack( side="left", padx=(12, 0) ) detect_section = ttk.LabelFrame(frame, text="2. 一键流程", padding=10, style="Section.TLabelframe") detect_section.pack(fill="x", pady=(0, 8)) detect_row = ttk.Frame(detect_section, style="Panel.TFrame") detect_row.pack(fill="x") self.start_btn = ttk.Button( detect_row, text="开始", command=self.on_start, style="Action.TButton" ) self.start_btn.pack(side="left") self.mode_label = ttk.Label( detect_row, textvariable=self.recommend_var, foreground="#1a73e8", font=("Segoe UI", 10, "bold"), wraplength=520, justify="left", style="Panel.TLabel", ) self.mode_label.pack(side="left", padx=(10, 0), fill="x", expand=True) hint_frame = ttk.Frame(detect_section, style="Panel.TFrame") hint_frame.pack(fill="x", pady=(6, 0)) ttk.Label(hint_frame, text="提示：", style="Panel.TLabel").pack(side="left", anchor="n") self.hint_label = ttk.Label( hint_frame, textvariable=self.hint_var, wraplength=680, justify="left", foreground="#5a6b85", style="Panel.TLabel", ) self.hint_label.pack(side="left", fill="x", expand=True) log_section = ttk.LabelFrame( frame, text=self.label_log, padding=10, style="Section.TLabelframe" ) log_section.pack(fill="both", expand=True) log_actions = ttk.Frame(log_section, style="Panel.TFrame") log_actions.pack(fill="x", pady=(0, 6)) self.clear_btn = ttk.Button(log_actions, text="清空日志", command=self.clear_log) self.clear_btn.pack(side="right") self.copy_btn = ttk.Button(log_actions, text="复制日志", command=self.copy_log) self.copy_btn.pack(side="right", padx=(0, 8)) self.log_box = scrolledtext.ScrolledText( log_section, height=16, wrap="word", background="#ffffff", foreground="#1f2a33", insertbackground="#1f2a33", ) self.log_box.pack(fill="both", expand=True) self.log_box.configure(state="disabled", font=("Consolas", 10)) status_frame = ttk.Frame(frame, style="Root.TFrame") status_frame.pack(fill="x", pady=(8, 0)) ttk.Label(status_frame, text="提示：", style="Root.TLabel").pack(side="left") self.status_label = ttk.Label(status_frame, textvariable=self.status_var, style="Root.TLabel") self.status_label.pack(side="left", padx=(6, 0)) self.progress = ttk.Progressbar( status_frame, mode="indeterminate", length=140, maximum=100, variable=self.progress_value, ) self.progress.pack(side="right", padx=(8, 0)) self.progress.pack_forget() self.update_flash_buttons() def _notify(self, kind: str, title: str, message: str) -> None: def _show() -> None: if kind == "info": messagebox.showinfo(title, message) return if kind == "warn": messagebox.showwarning(title, message) return messagebox.showerror(title, message) self.root.after(0, _show) def log(self, message: str, timestamp: bool = True) -> None: safe_message = _redact_sensitive(message) def _append() -> None: self.log_box.configure(state="normal") prefix = "" if timestamp and safe_message: prefix = f"[{datetime.now().strftime('%H:%M:%S')}] " line = prefix + safe_message if safe_message else "" self.log_box.insert("end", line + "\n") self.log_box.see("end") self.log_box.configure(state="disabled") self.root.after(0, _append) def log_section(self, title: str) -> None: self.log("", timestamp=False) self.log(title, timestamp=False) self.log("-" * len(title), timestamp=False) def _resolve_operator(self) -> str: for key in ("FACTORY_OPERATOR", "OPERATOR_ID", "OPERATOR", "USERNAME", "USER"): value = os.getenv(key) if isinstance(value, str) and value.strip(): return value.strip() try: return getpass.getuser() except Exception: return "unknown" def _log_dir(self) -> Path: env_path = os.getenv("FACTORY_LOG_DIR") if isinstance(env_path, str) and env_path.strip(): return Path(env_path).expanduser() if os.name == "nt": base = os.getenv("LOCALAPPDATA") or os.getenv("APPDATA") if base: return Path(base) / "TalkingQFactory" / "logs" return Path.home() / "AppData" / "Local" / "TalkingQFactory" / "logs" if sys.platform == "darwin": return Path.home() / "Library" / "Logs" / "TalkingQFactory" return Path.home() / ".cache" / "talkingq-factory" def _begin_run_record(self, port: str) -> None: firmware_version = os.getenv("FACTORY_FIRMWARE_VERSION") or os.getenv("FIRMWARE_VERSION") if isinstance(firmware_version, str) and firmware_version.strip(): firmware_version = firmware_version.strip() else: firmware_version = self.private_cfg.app_bin_name run_id = uuid.uuid4().hex[:8] self.run_record = { "tool": "factory_flash_gui", "run_id": run_id, "started_at": _now_iso(), "operator": self._resolve_operator(), "host": platform.node(), "port": port, "device_id": None, "firmware_version": firmware_version, "steps": {}, } def _record_error(self, stage: str, message: str, code: Optional[str] = None) -> None: if not self.run_record: return self.run_record["error"] = { "stage": stage, "message": message, "code": code, } def _record_step_result( self, step: str, start_ts: float, status: str, details: Optional[dict] = None, ) -> None: if not self.run_record: return started_at = datetime.fromtimestamp(start_ts, tz=timezone.utc).isoformat() payload = { "started_at": started_at, "duration_s": int(max(0.0, time.time() - start_ts)), "status": status, } if details: payload.update(details) self.run_record.setdefault("steps", {})[step] = payload def _sanitize_run_record(self, record: dict) -> dict: safe: dict = { "tool": record.get("tool"), "run_id": record.get("run_id"), "device_id": record.get("device_id"), "started_at": record.get("started_at"), "ended_at": record.get("ended_at"), "duration_s": record.get("duration_s"), "success": record.get("success"), } error = record.get("error") if isinstance(error, dict): safe_error: dict = {} for key in ("stage", "message", "code"): value = error.get(key) if isinstance(value, str): safe_error[key] = _redact_sensitive(value) elif value is not None: safe_error[key] = value if safe_error: safe["error"] = safe_error steps = {} raw_steps = record.get("steps", {}) if isinstance(raw_steps, dict): for step, payload in raw_steps.items(): if not isinstance(payload, dict): continue step_safe: dict = {} for key in ("status", "duration_s", "reason", "attempts", "errors"): if key not in payload: continue value = payload.get(key) if isinstance(value, str): step_safe[key] = _redact_sensitive(value) elif isinstance(value, list): step_safe[key] = [ _redact_sensitive(item) if isinstance(item, str) else item for item in value ] else: step_safe[key] = value if step_safe: steps[step] = step_safe if steps: safe["steps"] = steps return safe def _write_run_record(self, record: dict) -> Optional[Path]: log_dir = self._log_dir() try: log_dir.mkdir(parents=True, exist_ok=True) except OSError as exc: self.log(f"[警告] 无法创建记录目录: {exc}") return None record_id = record.get("device_id") or "unknown" success = record.get("success") if success is True: status_tag = "ok" elif success is False: status_tag = "failed" else: status_tag = "unknown" ts = datetime.now().strftime("%Y%m%d_%H%M%S") filename = f"{ts}_{_sanitize_filename(str(record_id))}_{status_tag}.json" path = log_dir / filename try: path.write_text(json.dumps(record, ensure_ascii=False, indent=2) + "\n", encoding="utf-8") except OSError as exc: self.log(f"[警告] 记录落盘失败: {exc}") return None return path def _finalize_run_record(self, success: bool) -> None: if not self.run_record: return self.run_record["ended_at"] = _now_iso() started_at = self.run_record.get("started_at") if isinstance(started_at, str): try: start_dt = datetime.fromisoformat(started_at) self.run_record["duration_s"] = int((datetime.now(timezone.utc) - start_dt).total_seconds()) except ValueError: pass if self.current_device_id and not self.run_record.get("device_id"): self.run_record["device_id"] = self.current_device_id if self.device_id_hint and not self.run_record.get("device_id"): self.run_record["device_id"] = self.device_id_hint self.run_record["success"] = bool(success) safe_record = self._sanitize_run_record(self.run_record) path = self._write_run_record(safe_record) if path: if self.safe_log: device_id = safe_record.get("device_id") if isinstance(device_id, str) and device_id: self.log(f"[OK] 产测记录已保存 (device_id={device_id})。") else: self.log("[OK] 产测记录已保存。") else: self.log(f"[OK] 产测记录已保存: {path}") self.run_record = None def _on_flash_output(self, line: str) -> None: self._parse_flash_progress(line) if self.safe_log: return if line: self.log(line) def _parse_flash_progress(self, line: str) -> None: if not line: return if "Wrote " in line and "bytes" in line: self.set_progress(100.0) if self.active_mode_label: self.set_status(f"{self.active_mode_label}进行中... 100%") return match = re.search(r"(\d+(?:\.\d+)?)%", line) if not match: return try: percent = float(match.group(1)) except ValueError: return self.set_progress(percent) if self.active_mode_label: self.set_status(f"{self.active_mode_label}进行中... {percent:.0f}%") def set_status(self, message: str) -> None: self.root.after(0, lambda: self.status_var.set(message)) def set_status_tone(self, tone: str) -> None: palette = { "info": "#1a73e8", "success": "#0b8a3c", "warn": "#c77c02", "error": "#c0362c", } color = palette.get(tone, "#000") self.root.after(0, lambda: self.status_label.configure(foreground=color)) def set_busy_indicator(self, busy: bool, mode: str = "indeterminate") -> None: def _apply() -> None: if busy: self.progress.pack(side="right", padx=(8, 0)) if mode == "determinate": self.progress.configure(mode="determinate") self.progress_value.set(0.0) else: self.progress.configure(mode="indeterminate") self.progress.start(8) return self.progress.stop() self.progress.pack_forget() self.root.after(0, _apply) def set_progress(self, value: float) -> None: def _apply() -> None: self.progress_value.set(max(0.0, min(100.0, value))) self.root.after(0, _apply) def clear_log(self, update_status: bool = True) -> None: self.log_box.configure(state="normal") self.log_box.delete("1.0", "end") self.log_box.configure(state="disabled") if update_status: self.set_status("日志已清空。") self.set_status_tone("info") def copy_log(self) -> None: content = self.log_box.get("1.0", "end").strip() self.root.clipboard_clear() if content: self.root.clipboard_append(content) self.set_status("日志已复制到剪贴板。") self.set_status_tone("success") else: self.set_status("日志为空。") self.set_status_tone("warn") def set_hint_text(self, text: str, tone: str = "info") -> None: palette = { "info": "#555555", "success": "#0b8a3c", "warn": "#c77c02", "error": "#c0362c", } color = palette.get(tone, "#555555") def _apply() -> None: self.hint_var.set(text) self.hint_label.configure(foreground=color) self.root.after(0, _apply) def _normalize_port(self, port: Optional[str]) -> str: if not port: return "" return port.strip().upper() def _on_port_change(self, *_args) -> None: port = self._normalize_port(self.port_var.get()) if port == self._last_port_value: return self._last_port_value = port if self.is_busy: self.update_flash_buttons() return if not port: self.checked_port = None self.summary_attempted = False self.summary_ok = False self.last_rec_mode = None self.secure_boot_blocked = False self.secure_boot_enabled = None self.boot_chain_present = None self.boot_chain_probe_reason = None self.download_mode_disabled = None self.secure_download_enabled = None self.manual_encrypt_disabled = None self.device_id_hint = None self.recommend_var.set("请选择串口后点击开始。") self.set_hint_text("", tone="info") self.card_port_var.set("未选择") self.card_secure_boot_var.set("未检测") self.card_flash_enc_var.set("未检测") self.card_recommend_var.set("待检测") self.update_flash_buttons() return if self.summary_attempted: self.summary_attempted = False self.summary_ok = False self.last_rec_mode = None self.secure_boot_blocked = False self.secure_boot_enabled = None self.boot_chain_present = None self.boot_chain_probe_reason = None self.download_mode_disabled = None self.secure_download_enabled = None self.manual_encrypt_disabled = None self.device_id_hint = None self.recommend_var.set("串口已变更，请点击开始重新检测。") self.set_hint_text("检测结果仅对当前串口有效。", tone="warn") elif not self.summary_attempted: self.recommend_var.set("准备开始：点击开始自动检测并执行流程。") self.set_hint_text("", tone="info") self.card_port_var.set(port) self.card_secure_boot_var.set("待检测") self.card_flash_enc_var.set("待检测") self.card_recommend_var.set("请先设备检测") self.update_flash_buttons() def update_flash_buttons(self) -> None: self.root.after(0, self._apply_flash_buttons) def _apply_flash_buttons(self) -> None: busy = self.is_busy self.detect_btn.state(["disabled"] if busy else ["!disabled"]) self.port_entry.configure(state="disabled" if busy else "normal") start_state = ["!disabled"] if busy: start_state = ["disabled"] self.start_btn.state(start_state) def disable_buttons(self, progress_mode: str = "indeterminate") -> None: self.is_busy = True self.set_busy_indicator(True, mode=progress_mode) self.update_flash_buttons() def enable_buttons(self) -> None: self.is_busy = False self.set_busy_indicator(False) self.update_flash_buttons() self.active_mode_label = "" def on_auto_detect(self) -> None: port = self._auto_detect_port(show_dialog=True) if port: self.port_var.set(port) def on_start(self) -> None: self.start_task(self.flow_task, progress_mode="determinate") def start_task(self, func, progress_mode: str = "indeterminate") -> None: if self.worker and self.worker.is_alive(): return self.disable_buttons(progress_mode=progress_mode) self.set_status("处理中...") self.set_status_tone("info") self.worker = threading.Thread(target=self._run_wrapper, args=(func,), daemon=True) self.worker.start() def _run_wrapper(self, func) -> None: try: func() finally: self.root.after(0, self.enable_buttons) if self.suppress_ready_status: self.suppress_ready_status = False return self.set_status("就绪。") self.set_status_tone("info") def prepare_secure_entries( self, entries: Iterable[Tuple[object, Path]] ) -> Optional[list[FlashSource]]: missing: list[Path] = [] sources: list[Tuple[int, Path, Path]] = [] for offset, path in entries: source = Path(str(path) + ".enc") try: offset_int = _parse_offset(offset) except ValueError as exc: self.log("[错误] 固件偏移配置无效。") if not self.safe_log: self.log(f" - {exc}") self._notify("error", "偏移无效", "固件偏移配置无效，请查看日志。") return None sources.append((offset_int, path, source)) if not source.exists(): missing.append(source) if missing: self.log("[错误] 缺少必要固件文件。") if not self.safe_log: for item in missing: self.log(f" - {_display_path(self.base_dir, item)}") self._notify("error", "文件缺失", "缺少必要固件文件，请查看日志。") return None try: load_firmware_decrypt_key() except FirmwareEnvelopeError as exc: self.log("[错误] .enc 固件解密密钥不可用。") if not self.safe_log: self.log(f" - {exc}") self._notify("error", "缺少解密密钥", str(exc)) return None resolved: list[FlashSource] = [] for offset_int, orig, source in sources: try: verify_file_signature(source) except SignatureError as exc: self.log("[错误] 固件校验失败。") if not self.safe_log: self.log(f" - {_display_path(self.base_dir, source)}") self.log(f" - {exc}") self._notify("error", "文件校验失败", "文件校验失败，请查看日志。") return None resolved.append( FlashSource( offset=offset_int, path=source, name=orig.name, ) ) return resolved def _is_force_bootloader_entry(self, entry: FlashSource) -> bool: name = entry.name.strip().lower() return entry.offset == 0x0 and "bootloader" in name and name.endswith(".bin") @staticmethod def _normalize_chip_name(value: object) -> str: text = str(value or "").strip().lower() return re.sub(r"[^0-9a-z]", "", text) @staticmethod def _format_rev_full(rev: int) -> str: if rev < 0: return "unknown" return f"v{rev // 100}.{rev % 100}" def _guard_force_bootloader_write( self, esp: Any, entry: FlashSource, payload: bytes, *, encrypt: bool, ) -> None: from esptool.bin_image import LoadFirmwareImage if not self._is_force_bootloader_entry(entry): raise RuntimeError( f"force 风险拦截：仅允许 bootloader@0x0，当前条目为 {entry.name} @ {hex(entry.offset)}" ) expected_chip = self._normalize_chip_name(self.private_cfg.chip) chip_name = str(getattr(esp, "CHIP_NAME", "") or "") actual_chip = self._normalize_chip_name(chip_name) if expected_chip and expected_chip != "auto" and actual_chip and expected_chip != actual_chip: raise RuntimeError( "force 风险拦截：设备芯片与配置不一致，" f"cfg={self.private_cfg.chip}, detected={chip_name or 'unknown'}" ) try: image = LoadFirmwareImage(chip_name or self.private_cfg.chip, payload) except Exception as exc: # noqa: BLE001 raise RuntimeError( f"force 风险拦截：无法解析 bootloader 镜像头（{entry.name}），已拒绝写入。" ) from exc image_chip_id = getattr(image, "chip_id", None) target_chip_id = getattr(esp, "IMAGE_CHIP_ID", None) if image_chip_id != target_chip_id: raise RuntimeError( "force 风险拦截：bootloader 芯片 ID 与目标芯片不匹配，" f"image={image_chip_id}, target={target_chip_id}" ) def _to_int(value: object, default: int = 0) -> int: try: return int(value) except (TypeError, ValueError): return default image_min_rev = _to_int(getattr(image, "min_rev", 0), 0) image_min_rev_full = _to_int(getattr(image, "min_rev_full", 0), 0) image_max_rev_full = _to_int(getattr(image, "max_rev_full", 0), 0) if image_max_rev_full == 0: use_rev_full_fields = False elif image_max_rev_full == 65535: use_rev_full_fields = not (image_min_rev_full == 0 and image_min_rev != 0) else: use_rev_full_fields = True if use_rev_full_fields: try: chip_rev_full = _to_int(getattr(esp, "get_chip_revision")(), -1) except Exception as exc: # noqa: BLE001 raise RuntimeError("force 风险拦截：无法读取芯片 revision 信息。") from exc if chip_rev_full < image_min_rev_full or chip_rev_full > image_max_rev_full: upper = ( "max rev not set" if image_max_rev_full == 65535 else self._format_rev_full(image_max_rev_full) ) raise RuntimeError( "force 风险拦截：bootloader revision 约束不满足，" f"requires=[{self._format_rev_full(image_min_rev_full)} - {upper}], " f"chip={self._format_rev_full(chip_rev_full)}" ) else: try: if chip_name == "ESP32-C3": chip_rev = _to_int(getattr(esp, "get_minor_chip_version")(), -1) else: chip_rev = _to_int(getattr(esp, "get_major_chip_version")(), -1) except Exception as exc: # noqa: BLE001 raise RuntimeError("force 风险拦截：无法读取芯片 revision 信息。") from exc if chip_rev < image_min_rev: raise RuntimeError( "force 风险拦截：bootloader 最小 revision 要求不满足，" f"requires={image_min_rev}, chip={chip_rev}" ) if not encrypt and chip_name != "ESP8266": try: flash_encryption_enabled = bool(getattr(esp, "get_flash_encryption_enabled")()) except Exception as exc: # noqa: BLE001 raise RuntimeError( "force 风险拦截：无法确认 Flash Encryption 状态，已拒绝明文 force 写入。" ) from exc if flash_encryption_enabled: raise RuntimeError( "force 风险拦截：检测到 Flash Encryption 已启用，拒绝明文 force 写入。" ) secure_download_mode = bool(getattr(esp, "secure_download_mode", False)) if chip_name != "ESP32" and secure_download_mode: try: security_info = getattr(esp, "get_security_info")() flash_crypt_cnt = _to_int( security_info.get("flash_crypt_cnt") if isinstance(security_info, dict) else 0 ) except Exception as exc: # noqa: BLE001 raise RuntimeError( "force 风险拦截：无法读取 secure download 状态下的 flash_crypt_cnt。" ) from exc if bin(flash_crypt_cnt).count("1") & 1 != 0: raise RuntimeError( "force 风险拦截：检测到加密状态，拒绝在 secure download 模式下明文 force 写入。" ) def ensure_files_exist(self, paths: Iterable[Path]) -> bool: missing = [str(p) for p in paths if not p.exists()] if missing: self.log("[错误] 缺少必要文件。") if not self.safe_log: for item in missing: self.log(f" - {_display_path(self.base_dir, Path(item))}") self._notify("error", "文件缺失", "缺少必要文件，请查看日志。") return False return True def _auto_detect_port(self, show_dialog: bool) -> Optional[str]: ports = list_serial_ports(self.python_cmd, show_dialog) if not ports: if show_dialog: self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。") return None selected = pick_preferred_port(ports) if len(ports) > 1: if self.safe_log: self.log("[提示] 检测到多个串口，已自动选择。") else: self.log(f"[提示] 检测到多个串口: {', '.join(ports)}") self.log(f"[提示] 自动选择: {selected}") if show_dialog: message = "检测到多个串口，已自动选择。" if not self.safe_log: message = f"检测到多个串口:\n{', '.join(ports)}\n\n已自动选择: {selected}" self._notify("info", "检测到多个串口", message) return selected def _auto_set_port(self, initial: bool) -> None: if not initial and self.port_var.get().strip(): return port = self._auto_detect_port(show_dialog=False) if port: self.port_var.set(port) def preflight_check(self) -> None: if not self.ensure_tools_available(): self.set_status("缺少工具。") self.set_status_tone("error") return self.log("[OK] 检测到工具，准备就绪。") self._refresh_mode_status() self._refresh_key_status() self.set_status("就绪。") self.set_status_tone("success") self.update_flash_buttons() def flow_task(self) -> None: self.suppress_ready_status = True self.current_device_id = None self.last_flash_app_only = None self._post_flash_reset_done = False self._begin_run_record("") success = False try: port = self.port_var.get().strip() if not port: port = self._auto_detect_port(show_dialog=False) or "" if not port: self.set_status("未检测到串口。") self.set_status_tone("warn") self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。") self._record_error("preflight", "no_port") return self.port_var.set(port) if self.run_record is not None: self.run_record["port"] = port if not self._ensure_private_config(port, show_dialog=True): self.set_status("配置加载失败。") self.set_status_tone("error") self._record_error("preflight", "private_config_missing") return self.set_progress(0.0) self.recommend_var.set("步骤 1/3：设备检测中...") if not self.device_check_task(): self._record_error("device_check", "failed") return flash_start_ts = time.time() mode = self._decide_flash_mode() if mode is None: reason = self.last_flash_block_reason or "mode_unknown" self._record_error("flash", reason) self._record_step_result( "flash", flash_start_ts, "failed", {"reason": reason}, ) return encrypt, app_only, mode_label = mode self.last_flash_app_only = app_only self.recommend_var.set(f"步骤 2/3：{mode_label}烧录中...") if not self.flash_task(encrypt=encrypt, app_only=app_only): self._record_error("flash", "failed") return self.recommend_var.set("步骤 3/3：安全复检中...") if not self.post_flash_check_task(port): self._record_error("post_flash_check", "failed") return self._attempt_reset(port) time.sleep(1.0) self.recommend_var.set("完成") self.set_status("流程完成。") self.set_status_tone("success") self._notify("info", "完成", "烧录流程完成。") success = True finally: self._finalize_run_record(success) def _decide_flash_mode(self) -> Optional[Tuple[bool, bool, str]]: self.last_flash_block_reason = None if not self.summary_ok: self._notify("error", "检测失败", "设备检测未通过，无法继续。") self.last_flash_block_reason = "summary_failed" return None if self.download_mode_disabled is True: self._notify("error", "下载模式已禁用", "ROM 下载模式已禁用，无法串口烧录。") self.last_flash_block_reason = "download_mode_disabled" return None if self.secure_download_enabled is True: self._notify( "error", "安全下载模式", "检测到安全下载模式已启用，当前工具不支持安全下载协议。", ) self.last_flash_block_reason = "secure_download_enabled" return None encrypt: Optional[bool] = None if self.last_rec_mode == "1": encrypt = False elif self.last_rec_mode == "2": encrypt = True if encrypt is None: self._notify("error", "模式未知", "无法判断烧录模式，请检查设备状态。") self.last_flash_block_reason = "rec_mode_unknown" return None if encrypt and self.manual_encrypt_disabled is True: self._notify( "error", "串口加密受限", "检测到禁用串口手动加密，无法使用当前工具进行加密写入。", ) self.last_flash_block_reason = "manual_encrypt_disabled" return None if self.secure_boot_enabled is True: if self.boot_chain_present is False: if encrypt: return True, False, "加密全量（初次写入）" return False, False, f"{self.plain_text}（初次写入）" mode_label = self.app_text if encrypt else f"{self.app_text}（不加密）" return encrypt, True, mode_label if encrypt is False: return False, False, f"{self.plain_text}" if encrypt is True: return True, False, "加密全量" self._notify("error", "模式未知", "无法判断烧录模式，请检查设备状态。") return None def _read_efuse_summary(self, port: str) -> Tuple[int, str, Optional[dict]]: args = [ "--chip", self.private_cfg.chip, "--port", port, "summary", "--format", "json", ] code, output = self.run_tool("espefuse", args, stream=False) summary_json = _extract_json_blob(output) if code == 0 else None return code, output, summary_json def _log_efuse_summary(self, output: str) -> None: if not output: return self.log_section("eFuse Summary") if self.safe_log: self.log("已获取（内容已脱敏）。", timestamp=False) return for line in output.splitlines(): self.log(line, timestamp=False) def device_check_task(self) -> bool: start_ts = time.time() port = self.port_var.get().strip() if not port: port = self._auto_detect_port(show_dialog=False) or "" if not port: self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。") self._record_step_result( "device_check", start_ts, "failed", {"reason": "no_port"}, ) return False self.port_var.set(port) if not self._ensure_private_config(port, show_dialog=True): self._record_step_result( "device_check", start_ts, "failed", {"reason": "private_config_missing"}, ) return False if not self.ensure_tools_available(): self._record_step_result( "device_check", start_ts, "failed", {"reason": "tools_missing"}, ) return False self.set_status("正在检测设备...") self.set_status_tone("info") self.log_section("设备检测") if not self.safe_log: self.log(f"串口: {port}") self.recommend_var.set("正在检测设备...") self.set_hint_text("", tone="info") code, summary_output, summary_json = self._read_efuse_summary(port) if summary_output: self._log_efuse_summary(summary_output) self.summary_attempted = True if code != 0 or summary_json is None: if code != 0: self.log("[警告] 设备检测失败。") else: self.log("[警告] 设备检测失败：JSON 解析失败。") self.set_status("检测失败。") self.set_status_tone("warn") self.summary_ok = False self.last_rec_mode = None self.secure_boot_blocked = False self.secure_boot_enabled = None self.boot_chain_present = None self.boot_chain_probe_reason = None self.download_mode_disabled = None self.secure_download_enabled = None self.manual_encrypt_disabled = None self.checked_port = None self.card_secure_boot_var.set("检测失败") self.card_flash_enc_var.set("检测失败") self.card_recommend_var.set("请检查连接后重试") hints = collect_error_hints(summary_output) # 不展示检测结果，只提示信息 self.recommend_var.set("检测失败，请查看提示后重试。") self.set_hint_text("\n".join(f"- {hint}" for hint in hints), tone="warn") self.log("操作指引:", timestamp=False) for hint in hints: self.log(f" - {hint}", timestamp=False) self.update_flash_buttons() self.log("流程已停止，请重试设备检测。") self._record_step_result( "device_check", start_ts, "failed", { "reason": "summary_json_failed" if summary_json is None else "summary_failed", "hints": hints, "port": port, }, ) return False rec_mode, rec_reason = recommend_mode_from_json( summary_json, self.private_cfg.device_check.crypt_cnt_fields ) self.summary_ok = True self.last_rec_mode = rec_mode self.secure_boot_blocked = False self.secure_boot_enabled = get_efuse_flag( summary_json, self.private_cfg.device_check.secure_boot_field ) self.download_mode_disabled = get_efuse_flag( summary_json, "DIS_DOWNLOAD_MODE", "UART_DOWNLOAD_DIS" ) self.secure_download_enabled = get_efuse_flag( summary_json, "ENABLE_SECURITY_DOWNLOAD", "SECURE_DOWNLOAD_MODE" ) self.manual_encrypt_disabled = get_efuse_flag( summary_json, "DIS_DOWNLOAD_MANUAL_ENCRYPT" ) if self.secure_boot_enabled is True and rec_mode == "1": self.boot_chain_present, self.boot_chain_probe_reason = self._probe_boot_chain_header(port) elif self.secure_boot_enabled is True and rec_mode == "2": # When flash encryption is enabled, raw flash reads are ciphertext and # cannot be used to infer plaintext image headers at 0x0. self.boot_chain_present = True self.boot_chain_probe_reason = "skip_probe_flash_encryption_enabled" else: self.boot_chain_present = None self.boot_chain_probe_reason = "skip_probe_not_applicable" self.checked_port = port mac = self._read_chip_mac(port) if mac: self.device_id_hint = mac if self.run_record and not self.run_record.get("device_id"): self.run_record["device_id"] = mac flash_enc_status = "未知" if rec_mode == "1": flash_enc_status = "未启用" elif rec_mode == "2": flash_enc_status = "已启用" self.card_flash_enc_var.set(flash_enc_status) if self.secure_boot_enabled is True: self.card_secure_boot_var.set("已启用") elif self.secure_boot_enabled is False: self.card_secure_boot_var.set("未启用") else: self.card_secure_boot_var.set("未知") boot_chain_missing = self.boot_chain_present is False if self.secure_boot_enabled: if boot_chain_missing: if rec_mode == "1": recommend_text = f"下一步：{self.plain_text}（初次全量）" elif rec_mode == "2": recommend_text = "检测到安全启动已启用且启动链缺失，将执行加密全量初次写入。" else: recommend_text = "检测到安全启动已启用且启动链缺失，请先核对设备信息。" elif rec_mode == "1": recommend_text = f"下一步：{self.app_text}（不加密）" elif rec_mode == "2": recommend_text = f"下一步：{self.app_text}（加密）" else: recommend_text = "请先核对设备信息" elif rec_mode == "1": recommend_text = f"下一步：{self.plain_text}" elif rec_mode == "2": recommend_text = "检测到闪存加密已启用但安全启动未启用，将执行加密全量烧录并在首启后启用安全启动。" else: recommend_text = "请先核对设备信息" if self.download_mode_disabled is True: recommend_text = "检测到下载模式已禁用，请按工艺处理。" elif self.secure_download_enabled is True: recommend_text = "检测到安全下载模式已启用，请按工艺处理。" elif self.manual_encrypt_disabled is True and rec_mode == "2": recommend_text = "检测到禁用串口手动加密，请使用预加密固件或 OTA。" self.recommend_var.set(recommend_text) self.card_recommend_var.set(recommend_text) hint_lines = [] hint_tone = "info" if rec_mode == "1": if self.secure_boot_enabled: if boot_chain_missing: hint_lines.append("检测到启动链缺失，将执行模式 A（全量写入）完成初次写入。") else: hint_lines.append("提示：安全启动已启用但闪存加密未启用，将以非加密写入方式执行模式 C。") hint_tone = "warn" else: hint_lines.append("已就绪：执行模式 A（全量写入）。") elif rec_mode == "2": if self.secure_boot_enabled: if boot_chain_missing: hint_lines.append("检测到启动链缺失，将执行加密全量初次写入。") hint_tone = "warn" else: hint_lines.append("已就绪：执行模式 C（加密）。") else: hint_lines.append("提示：将执行加密全量烧录，首启后自动启用安全启动。") hint_tone = "warn" else: hint_lines.append("信息不明确：请按工艺要求核对后再选择模式。") hint_tone = "warn" if self.secure_boot_enabled: if boot_chain_missing: hint_lines.append("检测到 flash@0x0 启动头无效/空白，将改为全量写入并在需要时使用 force。") else: hint_lines.append("已限制，仅允许模式 C。") if self.download_mode_disabled is True: hint_lines.append("检测到下载模式已禁用，串口烧录不可用。") hint_tone = "error" if self.secure_download_enabled is True: hint_lines.append("检测到安全下载模式已启用，当前工具不支持安全下载。") if hint_tone != "error": hint_tone = "warn" if self.manual_encrypt_disabled is True: if rec_mode == "2": hint_lines.append("检测到禁用串口手动加密，无法进行加密写入。") hint_tone = "error" else: hint_lines.append("检测到禁用串口手动加密（仅影响加密写入）。") if hint_tone != "error": hint_tone = "warn" self.set_hint_text("\n".join(hint_lines), tone=hint_tone) self.log("-" * 46, timestamp=False) self.log("设备检测完成，已自动确定烧录模式。") self.set_status("检测完成。") self.set_status_tone("success") self.update_flash_buttons() self._record_step_result( "device_check", start_ts, "ok", { "port": port, "recommend_mode": rec_mode, "recommend_reason": rec_reason, "secure_boot_enabled": self.secure_boot_enabled, "boot_chain_present": self.boot_chain_present, "boot_chain_probe_reason": self.boot_chain_probe_reason, "flash_enc_status": flash_enc_status, "download_mode_disabled": self.download_mode_disabled, "secure_download_enabled": self.secure_download_enabled, "manual_encrypt_disabled": self.manual_encrypt_disabled, }, ) return True def _format_recommendation( self, mode: Optional[str], reason: Optional[str], warn_reason: Optional[str] ) -> str: if warn_reason and not mode: return f"摘要不可用（{warn_reason}）。" if mode: if mode == "1": label = self.plain_text elif mode == "2": label = self.app_text else: label = f"模式 {mode}" reason_suffix = f"（{reason}）" if reason else "" if mode in ("1", "2"): return f"下一步：{label}（模式 {mode}{reason_suffix}）" return f"模式提示：{mode}{reason_suffix}" return "暂无指引。" def flash_task(self, encrypt: bool, app_only: bool = False) -> bool: start_ts = time.time() if not self.summary_attempted: self.log("[警告] 请先完成设备检测。") self._notify("warn", "请先设备检测", "请先完成设备检测，再开始烧录。") self._record_step_result( "flash", start_ts, "failed", {"reason": "summary_missing"}, ) return False port = self.port_var.get().strip() if not port: port = self._auto_detect_port(show_dialog=False) or "" if not port: self._notify("warn", "未检测到串口", "未检测到可用串口，请手动输入。") self._record_step_result( "flash", start_ts, "failed", {"reason": "no_port"}, ) return False self.port_var.set(port) if not self._ensure_private_config(port, show_dialog=True): self._record_step_result( "flash", start_ts, "failed", {"reason": "private_config_missing"}, ) return False bin_dir = select_bin_dir(self.base_dir, self.private_cfg.bin_dir) entries = build_flash_entries( bin_dir, self.private_cfg.layout, app_only, self.private_cfg.app_partition_offset, self.private_cfg.app_bin_name, ) secure_entries = self.prepare_secure_entries(entries) if not secure_entries: return False self.log("固件目录已加载。") if app_only: mode_label = self.app_text if encrypt else f"{self.app_text}（不加密）" else: mode_label = self.plain_text self.active_mode_label = mode_label self.suppress_ready_status = True self.set_progress(0.0) self.set_status(f"{mode_label}进行中... 0%") self.set_status_tone("info") self.log_section(f"{mode_label}烧录开始") if not self.safe_log: self.log(f"串口: {port} | 模式: {mode_label}") # Force is only allowed for secure-boot recovery of a missing boot chain. force_full_flash = bool( self.secure_boot_enabled is True and not app_only and self.boot_chain_present is False ) if force_full_flash: low_entries = [entry for entry in secure_entries if entry.offset < 0x8000] force_entries = [entry for entry in low_entries if self._is_force_bootloader_entry(entry)] if len(low_entries) != 1 or len(force_entries) != 1: self.log("[错误] force 安全策略校验失败：仅允许 bootloader@0x0 使用 force。") self._record_step_result( "flash", start_ts, "failed", { "reason": "force_policy_mismatch", "entries": [ {"name": entry.name, "offset": entry.offset, "path": str(entry.path)} for entry in secure_entries ], }, ) self._notify("error", "烧录策略异常", "force 策略校验失败，已停止以避免误写低地址。") return False self.log("[提示] 检测到安全启动已启用且启动链缺失：仅对 bootloader@0x0 启用 force。") strict_flash = bool(encrypt or self.secure_boot_enabled) max_attempts = self.flash_secure_max_attempts if strict_flash else self.flash_max_attempts attempts_used = 0 flash_ok = False last_output = "" last_report: Optional[FlashReport] = None last_failure_stage = "固件烧录" baud_override: Optional[str] = None for attempt in range(1, max_attempts + 1): attempts_used = attempt if attempt > 1: self.log(f"[提示] 第 {attempt} 次烧录尝试。") code, output, report = self.do_flash( port, encrypt=encrypt, entries=secure_entries, on_output=self._on_flash_output, baud_override=baud_override, force=force_full_flash, ) last_output = output last_report = report if not output.strip(): self.log("(无输出)") if code == 0: self.set_status("写后校验中...") self.set_status_tone("info") verify_ok, verify_err = self._verify_flash_readback( port, report, encrypt, baud_override=baud_override ) self._log_flash_report(report) if verify_ok: self.log("[OK] 固件烧录完成。") self.set_progress(100.0) flash_ok = True break last_failure_stage = "写后校验" last_output = verify_err or "readback_verify_failed" self.log("[错误] 写后校验失败。") else: if detect_secure_boot_block(output): self.secure_boot_blocked = True self.update_flash_buttons() if strict_flash: break if attempt < max_attempts: self.log("[警告] 烧录失败，准备复位后重试。") if baud_override is None: baud_override = "115200" self.log("[提示] 已自动降速到 115200 后重试。") self._attempt_reset(port) time.sleep(1.0) if not flash_ok: failure_entries = None if last_report: failure_entries = [ { "name": item.name, "offset": item.offset, "size": item.size, "sha256": item.sha256, "readback_sha256": item.readback_sha256, "verified": item.verified, } for item in last_report.entries ] self._record_step_result( "flash", start_ts, "failed", { "reason": last_failure_stage, "attempts": attempts_used, "mode": mode_label, "encrypt": encrypt, "app_only": app_only, "force": force_full_flash, "app_bin_name": self.private_cfg.app_bin_name, "readback_errors": last_report.readback_errors if last_report else None, "entries": failure_entries, }, ) self.handle_flash_failure(last_failure_stage, last_output) return False if last_report: app_entry = next( (item for item in last_report.entries if item.name == self.private_cfg.app_bin_name), None, ) self._record_step_result( "flash", start_ts, "ok", { "attempts": attempts_used, "mode": mode_label, "encrypt": encrypt, "app_only": app_only, "force": force_full_flash, "app_bin_name": self.private_cfg.app_bin_name, "entries": [ { "name": item.name, "offset": item.offset, "size": item.size, "sha256": item.sha256, "readback_sha256": item.readback_sha256, "verified": item.verified, } for item in last_report.entries ], "readback_ok": last_report.readback_ok, "readback_errors": last_report.readback_errors, "app_sha256": app_entry.sha256 if app_entry else None, }, ) else: self._record_step_result( "flash", start_ts, "ok", { "attempts": attempts_used, "mode": mode_label, "encrypt": encrypt, "app_only": app_only, "force": force_full_flash, }, ) self.log("[OK] 烧录完成。") needs_reset = self._needs_post_flash_reset() if not needs_reset: self._post_flash_reset_done = True if needs_reset: self.log("[提示] 正在复位设备进入应用模式...") self._attempt_reset(port) self._post_flash_reset_done = True time.sleep(1.0) self.set_status("烧录完成。") self.set_status_tone("success") self.set_hint_text("", tone="success") return True def post_flash_check_task(self, port: str) -> bool: start_ts = time.time() if not self.ensure_tools_available(): self._record_step_result( "post_flash_check", start_ts, "failed", {"reason": "tools_missing"}, ) return False self.set_status("安全复检中...") self.set_status_tone("info") self.log_section("安全复检") if not self.safe_log: self.log(f"串口: {port}") timeout_s = max(5.0, self.ready_timeout_s) deadline = time.time() + timeout_s attempts = 0 last_errors: list[str] = [] last_details: dict = {} last_output = "" if not self._post_flash_reset_done: self._attempt_reset(port) self._post_flash_reset_done = True time.sleep(1.0) while time.time() < deadline: attempts += 1 wait_cap = 60.0 if self.last_flash_app_only: wait_cap = 5.0 wait_s = min(wait_cap, max(1.0, deadline - time.time())) self.log(f"[提示] 等待设备完成安全流程（约 {int(wait_s)} 秒）...") time.sleep(wait_s) code, output, summary_json = self._read_efuse_summary(port) last_output = output if code == 0 and summary_json is not None: ok, details, errors = evaluate_post_flash_status( summary_json, self.private_cfg.device_check ) last_details = details if ok: self.log("[OK] 安全复检通过。") self._record_step_result( "post_flash_check", start_ts, "ok", { "attempts": attempts, "details": details, }, ) self.set_status("安全复检完成。") self.set_status_tone("success") return True last_errors = errors else: last_errors = [ "summary_failed" if code != 0 else "summary_json_failed" ] self.log("[错误] 安全复检失败。") if not self.safe_log and last_output: self.log(last_output) if last_errors: self.log(f"[错误] 安全复检未通过: {', '.join(last_errors)}") self.set_status("安全复检失败。") self.set_status_tone("error") if last_errors: self.set_hint_text( "\n".join(f"- {item}" for item in last_errors), tone="error" ) self._notify("error", "安全复检失败", "安全复检失败，请查看日志。") self._record_step_result( "post_flash_check", start_ts, "failed", { "attempts": attempts, "errors": last_errors, "details": last_details, }, ) return False def handle_flash_failure(self, stage: str, output: str) -> None: self.log(f"[错误] {stage}失败。请检查连接、串口及 BOOT/RESET 操作。") self.set_status(f"{stage}失败。") self.set_status_tone("error") self.set_progress(0.0) if detect_secure_boot_block(output): self.secure_boot_blocked = True if self.secure_boot_enabled is not True: self.secure_boot_enabled = True self.card_secure_boot_var.set("已启用") self.update_flash_buttons() hints = collect_error_hints(output) if detect_secure_boot_block(output): if self.boot_chain_present is False: hints.insert(0, "检测到安全启动限制：启动链缺失时请使用全量写入（必要时 force）。") else: hints.insert(0, "检测到安全启动已启用，常规更新仅允许模式 C（App-only）。") self.set_hint_text("\n".join(f"- {hint}" for hint in hints), tone="error") self.log("操作指引:", timestamp=False) for hint in hints: self.log(f" - {hint}", timestamp=False) self._notify("error", f"{stage}失败", f"{stage}失败，请查看日志与提示。") def _attempt_reset(self, port: str) -> None: args = ["--chip", self.private_cfg.chip, "--port", port, "run"] code, output = self.run_tool("esptool", args, stream=False) if code != 0: self.log(f"[警告] 复位失败：{output.strip()}") def _needs_post_flash_reset(self) -> bool: after = self.private_cfg.flash_args.get("after") if after is None: return False after_norm = str(after).strip() if not after_norm: return False after_norm = after_norm.replace("_", "-").lower() return after_norm in ("no-reset", "noreset", "none", "no") def _read_flash_bytes( self, port: str, offset: int, size: int, baud_override: Optional[str] = None ) -> Optional[bytes]: fd, temp_path = tempfile.mkstemp(suffix=".bin") os.close(fd) args = ["--chip", self.private_cfg.chip, "--port", port] baud_value = baud_override or self.private_cfg.baud if baud_value: baud_int = _value_to_int(baud_value) if baud_int is not None and baud_int > 921600 and baud_override is None: baud_value = "115200" args.extend(["--baud", str(baud_value)]) try: read_args = args + ["read-flash", hex(offset), str(size), temp_path] code, output = self.run_tool("esptool", read_args, stream=False) if code != 0 and ("Unknown command" in output or "invalid choice" in output): read_args = args + ["read_flash", hex(offset), str(size), temp_path] code, output = self.run_tool("esptool", read_args, stream=False) if code != 0: self.log(f"[警告] 读取闪存失败：{output.strip()}") return None try: return Path(temp_path).read_bytes() except OSError as exc: self.log(f"[警告] 读取闪存临时文件失败：{exc}") return None finally: try: os.remove(temp_path) except OSError: pass def _partition_table_offset_from_layout(self) -> Optional[int]: for offset, name in self.private_cfg.layout: if not isinstance(name, str): continue lowered = name.lower() if "partition-table" not in lowered and "partition_table" not in lowered: continue try: return _parse_offset(offset) except ValueError: self.log(f"[警告] 分区表偏移无效：{offset}") return None return None def _probe_boot_chain_header(self, port: str) -> tuple[Optional[bool], Optional[str]]: header = self._read_flash_bytes(port, 0x0, 32) if header is None: self.log("[警告] 启动链检测跳过：读取 flash@0x0 失败。") return None, "probe_failed" has_boot_chain, reason = detect_boot_chain_header_state(header) if not has_boot_chain: self.log(f"[警告] 启动链缺失（flash@0x0，{reason}）。") return False, reason partition_offset = self._partition_table_offset_from_layout() if partition_offset is None: self.log("[提示] 未找到分区表偏移，已按 bootloader 头有效处理。") return True, "boot_ok_partition_unknown" partition_header = self._read_flash_bytes(port, partition_offset, 4) if partition_header is None: self.log(f"[警告] 启动链检测跳过：读取分区表头失败（{hex(partition_offset)}）。") return None, "partition_probe_failed" if all(value == 0xFF for value in partition_header): self.log(f"[警告] 启动链缺失（分区表空白，{hex(partition_offset)}）。") return False, "partition_blank_ff" if partition_header[:2] != b"\xAA\x50": self.log( f"[警告] 启动链缺失（分区表头无效，{hex(partition_offset)}，magic=0x{partition_header[0]:02x}{partition_header[1]:02x}）。" ) return False, "partition_invalid_magic" self.log("[OK] 启动链检测通过（bootloader/partition table 头有效）。") return True, "ok" def _read_flash_segment( self, port: str, offset: int, size: int, baud_override: Optional[str] = None ) -> Optional[str]: fd, temp_path = tempfile.mkstemp(suffix=".bin") os.close(fd) args = ["--chip", self.private_cfg.chip, "--port", port] baud_value = baud_override or self.private_cfg.baud if baud_value: baud_int = _value_to_int(baud_value) if baud_int is not None and baud_int > 921600 and baud_override is None: baud_value = "115200" args.extend(["--baud", str(baud_value)]) try: read_args = args + ["read-flash", hex(offset), str(size), temp_path] code, output = self.run_tool("esptool", read_args, stream=False) if code != 0 and ("Unknown command" in output or "invalid choice" in output): read_args = args + ["read_flash", hex(offset), str(size), temp_path] code, output = self.run_tool("esptool", read_args, stream=False) if code != 0: self.log(f"[错误] 读回校验失败：{output.strip()}") return None try: hasher = hashlib.sha256() total = 0 with open(temp_path, "rb") as handle: while True: chunk = handle.read(65536) if not chunk: break total += len(chunk) hasher.update(chunk) except OSError as exc: self.log(f"[错误] 读回文件读取失败：{exc}") return None if total != size: self.log(f"[警告] 读回长度异常：{total} != {size}") return hasher.hexdigest() finally: try: os.remove(temp_path) except OSError: pass def _verify_flash_readback( self, port: str, report: FlashReport, encrypt: bool, baud_override: Optional[str] = None, ) -> Tuple[bool, str]: if encrypt: report.readback_ok = True report.readback_errors = [] return True, "" # esptool already performs in-band hash verification for plaintext writes. # Extra readback can fail due transient port locking on some hosts; keep it opt-in. if self.private_cfg.ui.get("strict_readback") is not True: report.readback_ok = True report.readback_errors = [] return True, "" if not report.entries: return True, "" ok = True errors: list[str] = [] for entry in report.entries: readback_hash = self._read_flash_segment( port, entry.offset, entry.size, baud_override=baud_override ) if readback_hash is None: ok = False errors.append(f"{entry.name}:readback_failed") continue entry.readback_sha256 = readback_hash if encrypt: entry.verified = None continue entry.verified = (entry.readback_sha256 == entry.sha256) if entry.verified is False: ok = False errors.append(f"{entry.name}:hash_mismatch") report.readback_ok = ok report.readback_errors = errors return ok, "; ".join(errors) def _log_flash_report(self, report: FlashReport) -> None: if self.safe_log: return for entry in report.entries: self.log(f"[HASH] {entry.name} sha256={entry.sha256}") if entry.readback_sha256: self.log(f"[READBACK] {entry.name} sha256={entry.readback_sha256}") if entry.verified is True: self.log(f"[OK] 校验一致: {entry.name}") elif entry.verified is False: self.log(f"[错误] 校验不一致: {entry.name}") def do_write_flash( self, port: str, entries: Iterable[FlashSource], encrypt: bool, on_output: Optional[Callable[[str], None]], baud_override: Optional[str] = None, force: bool = False, ) -> Tuple[int, str, FlashReport]: if not self.ensure_esptool_api_available(): return 1, "esptool API unavailable", FlashReport(entries=[], readback_ok=False, readback_errors=[]) entries_list = list(entries) try: decrypt_key = load_firmware_decrypt_key() except FirmwareEnvelopeError as exc: return 1, str(exc), FlashReport(entries=[], readback_ok=False, readback_errors=[]) report = FlashReport(entries=[], readback_ok=False, readback_errors=[]) def _run() -> None: import esptool from esptool import cmds from esptool.logger import EsptoolLogger, log as esptool_log try: EsptoolLogger._set_smart_features(False) except Exception: pass try: esptool_log.set_verbosity("normal") except Exception: pass baud_value = _value_to_int(baud_override or self.private_cfg.baud) if baud_value is None: raise ValueError(f"无效波特率: {baud_override or self.private_cfg.baud}") bootstrap_baud = 115200 use_stub = bool(self.private_cfg.flash_args.get("use_stub", True)) esp = esptool.detect_chip( port=port, baud=bootstrap_baud, connect_mode=self.private_cfg.flash_args["before"], ) try: if use_stub: try: esp = cmds.run_stub(esp) except Exception as exc: # noqa: BLE001 if on_output: on_output(f"[警告] 启动 stub 失败，已降级为 no-stub: {exc}\n") if baud_value != bootstrap_baud: try: esp.change_baud(baud_value) except Exception as exc: # noqa: BLE001 if on_output: on_output( f"[警告] 切换波特率到 {baud_value} 失败，继续使用 {bootstrap_baud}: {exc}\n" ) # Keep API flow consistent with esptool CLI: attach SPI flash before write. cmds.attach_flash(esp, None) for entry in entries_list: try: payload = entry.path.read_bytes() except OSError as exc: raise RuntimeError(f"固件读取失败: {entry.path.name}") from exc try: payload = decrypt_firmware_blob(payload, decrypt_key) except FirmwareEnvelopeError as exc: raise RuntimeError(f"固件解密失败: {entry.path.name}: {exc}") from exc plain: Optional[bytearray] = None try: plain = bytearray(payload) payload = b"" sha256 = _sha256_hex(plain) report.entries.append( FlashEntryReport( name=entry.name, offset=entry.offset, size=len(plain), sha256=sha256, readback_sha256=None, verified=None, ) ) write_kwargs = { "flash_freq": self.private_cfg.flash_args["flash_freq"], "flash_mode": self.private_cfg.flash_args["flash_mode"], "flash_size": self.private_cfg.flash_args["flash_size"], "encrypt": encrypt, } plain_bytes = bytes(plain) if force and entry.offset < 0x8000 and not self._is_force_bootloader_entry(entry): raise RuntimeError( f"force 策略拒绝低地址条目: {entry.name} @ {hex(entry.offset)}" ) if force and self._is_force_bootloader_entry(entry): self._guard_force_bootloader_write( esp, entry, plain_bytes, encrypt=encrypt, ) write_kwargs["force"] = True try: cmds.write_flash( esp, [(entry.offset, plain_bytes)], **write_kwargs, ) except TypeError as exc: if write_kwargs.get("force") and "force" in str(exc): # Compatibility fallback for older esptool APIs. write_kwargs.pop("force", None) if on_output: on_output("[警告] 当前 esptool API 不支持 force 参数，已回退默认写入。\n") cmds.write_flash( esp, [(entry.offset, plain_bytes)], **write_kwargs, ) else: raise finally: _wipe_bytearray(plain) after = self.private_cfg.flash_args["after"] if after: after_norm = str(after).strip() if after_norm: after_norm = after_norm.replace("_", "-").lower() if after_norm not in ("no-reset", "noreset", "none", "no"): esptool.reset_chip(esp, reset_mode=after_norm) finally: try: esp._port.close() except Exception: pass code, output = run_esptool_api(_run, stream=True, on_output=on_output) return code, output, report def ensure_esptool_api_available(self) -> bool: try: import esptool # noqa: F401 return True except Exception as exc: # noqa: BLE001 if self.frozen: self._notify( "error", "缺少 esptool", "内置环境缺少 esptool，请重新打包。\n\n" + str(exc), ) return False script_path = self.base_dir / "flash_encrypted_gui.py" self._notify( "error", "缺少 esptool", "当前 Python 环境缺少 esptool（API 模式需要同一解释器可用）。\n" f"请使用虚拟环境运行：\n {self.python_cmd} {script_path}\n" f"或安装：\n {sys.executable} -m pip install esptool\n\n{exc}", ) return False def ensure_tools_available(self) -> bool: if not self.ensure_esptool_api_available(): return False code, output = self.run_tool("esptool", ["version"], stream=False) if code != 0: if self.frozen: self._notify("error", "缺少 esptool", "内置环境缺少 esptool，请重新打包。\n\n" + output) return False self._notify( "error", "缺少 esptool", "未检测到 esptool。请先安装：\n" f"{self.python_cmd} -m pip install esptool\n\n{output}", ) return False if self.frozen: try: import importlib importlib.import_module("espefuse") except Exception as exc: # noqa: BLE001 self._notify( "error", "缺少 espefuse", "内置环境缺少 espefuse，请重新打包。\n\n" + str(exc), ) return False return True code, output = self.run_tool("espefuse", ["--help"], stream=False) if code != 0: self._notify( "error", "缺少 espefuse", "未检测到 espefuse。请先安装：\n" f"{self.python_cmd} -m pip install esptool\n\n{output}", ) return False return True def run_tool( self, tool: str, args: Iterable[str], stream: bool, on_output: Optional[Callable[[str], None]] = None, ) -> Tuple[int, str]: if self.frozen: return run_tool_module(tool, args, cwd=self.base_dir, stream=stream, on_output=on_output) if self.python_cmd is None: return 1, "Python 环境未就绪。" cmd = [str(self.python_cmd), "-m", tool, *args] return run_command(cmd, cwd=self.base_dir, stream=stream, on_output=on_output) def do_flash( self, port: str, encrypt: bool, entries: Iterable[FlashSource], on_output: Optional[Callable[[str], None]], baud_override: Optional[str] = None, force: bool = False, ) -> Tuple[int, str, FlashReport]: return self.do_write_flash( port, entries, encrypt=encrypt, on_output=on_output, baud_override=baud_override, force=force, ) def run(self) -> None: self.root.mainloop() def main() -> None: try: app = FlashGUI() except (FileNotFoundError, PermissionError, ValueError, KeyError, TypeError) as exc: try: messagebox.showerror("配置错误", str(exc)) except tk.TclError: print(exc, file=sys.stderr) raise SystemExit(1) from exc app.run() if __name__ == "__main__": try: main() except KeyboardInterrupt: sys.exit(1)