453 lines
15 KiB
Python
453 lines
15 KiB
Python
from __future__ import annotations
|
||
|
||
import hashlib
|
||
import json
|
||
import re
|
||
from pathlib import Path
|
||
from typing import Iterable, Optional, Tuple
|
||
|
||
from .models import DeviceCheckConfig
|
||
|
||
_MAC_RE = re.compile(r"(?i)([0-9a-f]{2}:){5}[0-9a-f]{2}")
|
||
_MAC_OUTPUT_RE = re.compile(r"(?i)\bMAC:\s*([0-9a-f]{2}(?::[0-9a-f]{2}){5})")
|
||
_ESP_IMAGE_MAGIC = 0xE9
|
||
|
||
|
||
def _sha256_hex(data: bytes) -> str:
|
||
return hashlib.sha256(data).hexdigest()
|
||
|
||
|
||
def _wipe_bytearray(buf: Optional[bytearray]) -> None:
|
||
if not buf:
|
||
return
|
||
for idx in range(len(buf)):
|
||
buf[idx] = 0
|
||
|
||
|
||
def collect_error_hints(output: str) -> list[str]:
|
||
lowered = output.lower()
|
||
hints: list[str] = []
|
||
|
||
if "secure boot detected" in lowered or "writing to flash regions < 0x8000 is disabled" in lowered:
|
||
hints.append("检测到写入限制：请按工艺选择模式。")
|
||
hints.append("请按工艺选择模式后重试。")
|
||
if "secure download mode is enabled" in lowered:
|
||
hints.append("检测到下载限制：请按工艺选择模式。")
|
||
if "could not open port" in lowered or "permissionerror" in lowered or "access is denied" in lowered:
|
||
hints.append("串口被占用或权限不足，关闭串口工具后重试。")
|
||
if "no serial data received" in lowered or "failed to connect to esp32" in lowered:
|
||
hints.append("确认串口号正确，未被其他软件占用。")
|
||
hints.append("按住 BOOT，点一下 RESET 进入下载模式。")
|
||
if "download mode successfully detected" in lowered and "no sync reply" in lowered:
|
||
hints.append("下载模式已进入但无法同步，通常是 TX/RX 线序或供电问题。")
|
||
if "serial tx path seems to be down" in lowered:
|
||
hints.append("检查 USB 转串口的 TX/RX 线序、焊接和接触。")
|
||
if "invalid head of packet" in lowered or "possible serial noise" in lowered:
|
||
hints.append("串口噪声/干扰，尝试更换 USB 线/接口或降低波特率（如 115200）。")
|
||
if "stopiteration" in lowered:
|
||
hints.append("串口通信异常中断，重新插拔设备后重试。")
|
||
|
||
if not hints:
|
||
hints.append("请检查串口号、线缆/供电，并按住 BOOT 点一下 RESET 进入下载模式。")
|
||
|
||
deduped: list[str] = []
|
||
seen = set()
|
||
for hint in hints:
|
||
if hint in seen:
|
||
continue
|
||
seen.add(hint)
|
||
deduped.append(hint)
|
||
return deduped
|
||
|
||
|
||
def detect_secure_boot_block(output: str) -> bool:
|
||
lowered = output.lower()
|
||
return (
|
||
"secure boot detected" in lowered
|
||
or "writing to flash regions < 0x8000 is disabled" in lowered
|
||
)
|
||
|
||
|
||
def detect_boot_chain_header_state(header_bytes: bytes) -> tuple[bool, str]:
|
||
if not header_bytes:
|
||
return False, "empty"
|
||
if all(value == 0xFF for value in header_bytes):
|
||
return False, "blank_ff"
|
||
if all(value == 0x00 for value in header_bytes):
|
||
return False, "blank_00"
|
||
first = header_bytes[0]
|
||
if first != _ESP_IMAGE_MAGIC:
|
||
return False, f"invalid_magic_0x{first:02x}"
|
||
return True, "ok"
|
||
|
||
|
||
def _extract_json_blob(output: str) -> Optional[dict]:
|
||
start = output.find("{")
|
||
end = output.rfind("}")
|
||
if start < 0 or end <= start:
|
||
return None
|
||
try:
|
||
return json.loads(output[start : end + 1])
|
||
except json.JSONDecodeError:
|
||
return None
|
||
|
||
|
||
def _json_field_value(data: Optional[dict], name: str) -> Optional[object]:
|
||
if not isinstance(data, dict):
|
||
return None
|
||
field = data.get(name)
|
||
if not isinstance(field, dict):
|
||
return None
|
||
return field.get("value")
|
||
|
||
|
||
def _value_to_int(value: object) -> Optional[int]:
|
||
if isinstance(value, bool):
|
||
return int(value)
|
||
if isinstance(value, (int, float)):
|
||
return int(value)
|
||
if isinstance(value, str):
|
||
stripped = value.strip()
|
||
lowered = stripped.lower()
|
||
if lowered in ("true", "yes", "enable", "enabled", "on"):
|
||
return 1
|
||
if lowered in ("false", "no", "disable", "disabled", "off"):
|
||
return 0
|
||
if lowered.startswith("0x"):
|
||
try:
|
||
return int(stripped, 16)
|
||
except ValueError:
|
||
return None
|
||
if lowered.startswith("0b"):
|
||
try:
|
||
return int(stripped, 2)
|
||
except ValueError:
|
||
return None
|
||
if stripped.isdigit():
|
||
return int(stripped)
|
||
return None
|
||
|
||
|
||
def get_efuse_flag(data: Optional[dict], *field_names: str) -> Optional[bool]:
|
||
if not isinstance(data, dict):
|
||
return None
|
||
for name in field_names:
|
||
if not isinstance(name, str) or not name:
|
||
continue
|
||
value = _json_field_value(data, name)
|
||
value_int = _value_to_int(value)
|
||
if value_int is None:
|
||
continue
|
||
return bool(value_int)
|
||
return None
|
||
|
||
|
||
def _parse_offset(value: object) -> int:
|
||
parsed = _value_to_int(value)
|
||
if parsed is None:
|
||
raise ValueError(f"无效偏移: {value}")
|
||
return parsed
|
||
|
||
|
||
def recommend_mode_from_json(
|
||
data: dict, crypt_cnt_fields: Iterable[str]
|
||
) -> Tuple[Optional[str], Optional[str]]:
|
||
targets = tuple(crypt_cnt_fields)
|
||
if not targets:
|
||
return None, "CRYPT_CNT_NOT_CONFIGURED"
|
||
for name in targets:
|
||
value = _json_field_value(data, name)
|
||
if value is None:
|
||
continue
|
||
value_int = _value_to_int(value)
|
||
if value_int is not None:
|
||
ones = bin(value_int).count("1")
|
||
if ones == 0:
|
||
return "1", f"{name}_ZERO"
|
||
if ones % 2 == 1:
|
||
return "2", f"{name}_ODD"
|
||
return None, f"{name}_UNEXPECTED"
|
||
if isinstance(value, str):
|
||
lowered = value.strip().lower()
|
||
if lowered in ("enable", "enabled"):
|
||
return "2", f"{name}_ENABLE"
|
||
if lowered in ("disable", "disabled"):
|
||
return "1", f"{name}_DISABLE"
|
||
return None, "CRYPT_CNT_NOT_FOUND"
|
||
|
||
|
||
def _crypt_cnt_parity(data: dict, crypt_cnt_fields: Iterable[str]) -> Optional[int]:
|
||
for name in crypt_cnt_fields:
|
||
value = _json_field_value(data, name)
|
||
value_int = _value_to_int(value)
|
||
if value_int is None:
|
||
continue
|
||
ones = bin(value_int).count("1")
|
||
return ones % 2
|
||
return None
|
||
|
||
|
||
def _purpose_matches(value: object, match_text: Optional[str]) -> Optional[bool]:
|
||
if match_text is None:
|
||
return None
|
||
if isinstance(value, str):
|
||
return match_text.upper() in value.upper()
|
||
if isinstance(value, (int, float)):
|
||
if int(value) == 0:
|
||
return False
|
||
return None
|
||
|
||
|
||
def _match_secure_boot_purpose(value: object) -> bool:
|
||
if not isinstance(value, str):
|
||
return False
|
||
return value.upper().startswith("SECURE_BOOT_DIGEST")
|
||
|
||
|
||
def _key_index_from_purpose_name(name: Optional[str]) -> Optional[int]:
|
||
if not name:
|
||
return None
|
||
if not name.startswith("KEY_PURPOSE_"):
|
||
return None
|
||
suffix = name[len("KEY_PURPOSE_") :]
|
||
if not suffix.isdigit():
|
||
return None
|
||
return int(suffix)
|
||
|
||
|
||
def _rd_dis_for_key(rd_dis_value: Optional[int], key_name: Optional[str]) -> Optional[bool]:
|
||
if rd_dis_value is None:
|
||
return None
|
||
key_index = _key_index_from_purpose_name(key_name)
|
||
if key_index is None:
|
||
return None
|
||
return bool(rd_dis_value & (1 << key_index))
|
||
|
||
|
||
def _iter_key_purpose_fields(data: dict) -> list[tuple[str, object]]:
|
||
rows: list[tuple[str, object]] = []
|
||
if not isinstance(data, dict):
|
||
return rows
|
||
for key_name, raw_field in data.items():
|
||
if not isinstance(key_name, str):
|
||
continue
|
||
if not key_name.startswith("KEY_PURPOSE_"):
|
||
continue
|
||
value = None
|
||
if isinstance(raw_field, dict):
|
||
value = raw_field.get("value")
|
||
rows.append((key_name, value))
|
||
|
||
def _sort_key(row: tuple[str, object]) -> tuple[int, str]:
|
||
key_name = row[0]
|
||
idx = _key_index_from_purpose_name(key_name)
|
||
if idx is None:
|
||
return (1_000_000, key_name)
|
||
return (idx, key_name)
|
||
|
||
rows.sort(key=_sort_key)
|
||
return rows
|
||
|
||
|
||
def _purpose_value_matches(
|
||
value: object,
|
||
expected_match: Optional[str],
|
||
allow_secure_boot_prefix: bool = False,
|
||
) -> bool:
|
||
matched = _purpose_matches(value, expected_match)
|
||
if matched is True:
|
||
return True
|
||
if not allow_secure_boot_prefix or not expected_match:
|
||
return False
|
||
if "SECURE_BOOT_DIGEST" not in expected_match.upper():
|
||
return False
|
||
return _match_secure_boot_purpose(value)
|
||
|
||
|
||
def _collect_expected_purpose_matches(cfg: DeviceCheckConfig) -> tuple[list[str], list[str]]:
|
||
flash_expected: list[str] = []
|
||
secure_expected: list[str] = []
|
||
|
||
flash_match = cfg.flash_key_purpose_match
|
||
if flash_match:
|
||
flash_match = flash_match.strip()
|
||
if flash_match:
|
||
flash_expected.append(flash_match)
|
||
|
||
secure_boot_match = cfg.secure_boot_key_purpose_match
|
||
if secure_boot_match:
|
||
secure_boot_match = secure_boot_match.strip()
|
||
if secure_boot_match:
|
||
secure_expected.append(secure_boot_match)
|
||
|
||
return flash_expected, secure_expected
|
||
|
||
|
||
def _match_expected_keys(
|
||
key_values: list[tuple[str, object]],
|
||
expected_matches: list[str],
|
||
allow_secure_boot_prefix: bool = False,
|
||
) -> tuple[list[dict], list[str]]:
|
||
matches: list[dict] = []
|
||
missing: list[str] = []
|
||
used_keys: set[str] = set()
|
||
for expected in expected_matches:
|
||
found: Optional[dict] = None
|
||
for key_name, value in key_values:
|
||
if key_name in used_keys:
|
||
continue
|
||
if _purpose_value_matches(
|
||
value,
|
||
expected,
|
||
allow_secure_boot_prefix=allow_secure_boot_prefix,
|
||
):
|
||
found = {"key": key_name, "value": value}
|
||
break
|
||
if found is None:
|
||
missing.append(expected)
|
||
continue
|
||
used_keys.add(found["key"])
|
||
matches.append(found)
|
||
return matches, missing
|
||
|
||
|
||
def evaluate_post_flash_status(
|
||
data: dict, cfg: DeviceCheckConfig
|
||
) -> Tuple[bool, dict, list[str]]:
|
||
details: dict = {}
|
||
errors: list[str] = []
|
||
|
||
crypt_cnt_mode, crypt_cnt_reason = recommend_mode_from_json(data, cfg.crypt_cnt_fields)
|
||
details["crypt_cnt_mode"] = crypt_cnt_mode
|
||
details["crypt_cnt_reason"] = crypt_cnt_reason
|
||
parity = _crypt_cnt_parity(data, cfg.crypt_cnt_fields)
|
||
if parity is not None:
|
||
details["crypt_cnt_parity"] = "odd" if parity == 1 else "even"
|
||
if cfg.crypt_cnt_expect and cfg.crypt_cnt_expect != "any":
|
||
expect = cfg.crypt_cnt_expect
|
||
if expect == "odd" and crypt_cnt_mode != "2":
|
||
errors.append("crypt_cnt_not_odd")
|
||
elif expect == "zero" and crypt_cnt_mode != "1":
|
||
errors.append("crypt_cnt_not_zero")
|
||
elif expect == "even" and crypt_cnt_mode == "2":
|
||
errors.append("crypt_cnt_unexpected_odd")
|
||
|
||
secure_boot_value = _value_to_int(_json_field_value(data, cfg.secure_boot_field))
|
||
secure_boot_enabled: Optional[bool] = None
|
||
if secure_boot_value is not None:
|
||
secure_boot_enabled = bool(secure_boot_value)
|
||
details["secure_boot_enabled"] = secure_boot_enabled
|
||
if cfg.secure_boot_expected is not None and secure_boot_enabled is not None:
|
||
if secure_boot_enabled != cfg.secure_boot_expected:
|
||
errors.append("secure_boot_unexpected")
|
||
elif cfg.secure_boot_expected is not None and secure_boot_enabled is None:
|
||
errors.append("secure_boot_unknown")
|
||
|
||
rd_dis_value = _value_to_int(_json_field_value(data, cfg.key_read_disable_field))
|
||
details["rd_dis_value"] = rd_dis_value
|
||
|
||
key_details: dict = {}
|
||
|
||
key_values = _iter_key_purpose_fields(data)
|
||
flash_expected, secure_expected = _collect_expected_purpose_matches(cfg)
|
||
flash_matches, flash_missing = _match_expected_keys(
|
||
key_values, flash_expected, allow_secure_boot_prefix=False
|
||
)
|
||
secure_matches, secure_missing = _match_expected_keys(
|
||
key_values, secure_expected, allow_secure_boot_prefix=True
|
||
)
|
||
|
||
flash_rd_details: list[dict] = []
|
||
secure_rd_details: list[dict] = []
|
||
flash_rd_missing = False
|
||
flash_rd_unknown = False
|
||
secure_rd_blocked = False
|
||
secure_rd_unknown = False
|
||
|
||
for matched in flash_matches:
|
||
key_name = matched["key"]
|
||
rd_ok = _rd_dis_for_key(rd_dis_value, key_name)
|
||
flash_rd_details.append({"key": key_name, "rd_protected": rd_ok})
|
||
if rd_ok is False:
|
||
flash_rd_missing = True
|
||
elif rd_ok is None:
|
||
flash_rd_unknown = True
|
||
|
||
for matched in secure_matches:
|
||
key_name = matched["key"]
|
||
rd_ok = _rd_dis_for_key(rd_dis_value, key_name)
|
||
secure_rd_details.append({"key": key_name, "rd_protected": rd_ok})
|
||
if rd_ok is True:
|
||
secure_rd_blocked = True
|
||
elif rd_ok is None:
|
||
secure_rd_unknown = True
|
||
|
||
if flash_expected and flash_missing:
|
||
errors.append("flash_key_missing")
|
||
if secure_expected and secure_missing:
|
||
errors.append("secure_boot_key_missing")
|
||
if flash_matches and flash_rd_missing:
|
||
errors.append("flash_key_rd_dis_missing")
|
||
elif flash_matches and flash_rd_unknown:
|
||
errors.append("flash_key_rd_dis_unknown")
|
||
if secure_matches and secure_rd_blocked:
|
||
errors.append("secure_boot_key_rd_dis_set")
|
||
elif secure_matches and secure_rd_unknown:
|
||
errors.append("secure_boot_key_rd_dis_unknown")
|
||
|
||
key_details["flash_enc_keys"] = flash_matches
|
||
key_details["secure_boot_keys"] = secure_matches
|
||
key_details["flash_enc_rd_protected"] = flash_rd_details
|
||
key_details["secure_boot_rd_protected"] = secure_rd_details
|
||
details["keys"] = key_details
|
||
|
||
return len(errors) == 0, details, errors
|
||
|
||
|
||
def build_flash_entries(
|
||
bin_dir: Path,
|
||
layout: Iterable[Tuple[str, str]],
|
||
app_only: bool,
|
||
app_partition_offset: str,
|
||
app_bin_name: str,
|
||
) -> list[Tuple[object, Path]]:
|
||
if app_only:
|
||
return [(app_partition_offset, bin_dir / app_bin_name)]
|
||
return [(offset, bin_dir / name) for offset, name in layout]
|
||
|
||
|
||
def recommend_mode(
|
||
summary_text: str, crypt_cnt_fields: Iterable[str]
|
||
) -> Tuple[Optional[str], Optional[str]]:
|
||
targets = tuple(crypt_cnt_fields)
|
||
if not targets:
|
||
return None, "CRYPT_CNT_NOT_CONFIGURED"
|
||
for line in summary_text.splitlines():
|
||
if "CRYPT_CNT_DISABLE" in line:
|
||
continue
|
||
if not any(target in line for target in targets):
|
||
continue
|
||
match = re.search(r"\(0b([01]+)\)", line)
|
||
if match:
|
||
ones = match.group(1).count("1")
|
||
else:
|
||
match = re.search(r"=\s*([0-9]+)", line)
|
||
if not match:
|
||
return None, "CRYPT_CNT_PARSE_FAIL"
|
||
value = int(match.group(1), 10)
|
||
ones = bin(value).count("1")
|
||
if ones == 0:
|
||
return "1", "CRYPT_CNT_ZERO"
|
||
if ones % 2 == 1:
|
||
return "2", "CRYPT_CNT_ODD"
|
||
return None, "CRYPT_CNT_UNEXPECTED"
|
||
return None, "CRYPT_CNT_NOT_FOUND"
|
||
|
||
|
||
def extract_mac_from_output(output: str) -> Optional[str]:
|
||
match = _MAC_OUTPUT_RE.search(output) or _MAC_RE.search(output)
|
||
if not match:
|
||
return None
|
||
mac = match.group(1) if match.re is _MAC_OUTPUT_RE else match.group(0)
|
||
return mac.lower()
|