"""Convert the online burn-in JSON to a private EchoEar factory NVS image. No network or serial access. No cloud credentials or Wi-Fi belong in this image. """ import argparse import csv import hashlib import json import os from pathlib import Path import re import secrets import subprocess import sys from urllib.parse import urlsplit from ble_factory import security_material FACTORY_OFFSET = 0xE4E000 FACTORY_SIZE = 0x6000 def normalize_mac(value): compact = value.replace(":", "").replace("-", "") if not re.fullmatch(r"[0-9a-fA-F]{12}", compact) or compact.lower() in ("000000000000", "ffffffffffff"): raise ValueError("Expected MAC must contain exactly 12 hex digits") return ":".join(compact[i:i + 2] for i in range(0, 12, 2)).lower() def read_credential(path): if path.stat().st_size > 16384: raise ValueError("Credential JSON exceeds 16 KiB") try: item = json.loads(path.read_text(encoding="utf-8-sig")) except (ValueError, UnicodeError): raise ValueError("Invalid credential JSON; download the burn-in file again") from None if not isinstance(item, dict): raise ValueError("Credential must be a JSON object") factory_id = item.get("factory_id", item.get("device_id")) if item.get("factory_id") and item.get("device_id") and item["factory_id"] != item["device_id"]: raise ValueError("Conflicting factory_id and device_id") if not isinstance(factory_id, str) or not re.fullmatch(r"KP01-[0-9A-F]{12}", factory_id): raise ValueError("Invalid factory_id; use the online burn-in JSON") # The cloud ID is random: it is deliberately NOT compared to the eFuse MAC. model = item.get("model") if model != "ECHOEAR-S3": raise ValueError("This tool supports model ECHOEAR-S3 only") key = item.get("device_key_hex") if not isinstance(key, str) or not re.fullmatch(r"[0-9a-fA-F]{64}", key): raise ValueError("device_key_hex must be 32 bytes in hexadecimal") version = item.get("key_version") if type(version) is not int or not 1 <= version <= 0xFFFFFFFF: raise ValueError("Invalid key_version") origin = item.get("origin") if not isinstance(origin, str) or not origin.isascii() or len(origin) > 150 or any(c.isspace() for c in origin): raise ValueError("Invalid HTTPS origin") try: url = urlsplit(origin) port = url.port except ValueError: raise ValueError("Invalid HTTPS origin") from None if (url.scheme != "https" or not url.hostname or url.username or url.password or url.path or url.query or url.fragment or "\\" in origin or not re.fullmatch(r"[A-Za-z0-9.:-]+", url.netloc) or (port is not None and not 1 <= port <= 65535)): raise ValueError("origin must be HTTPS scheme + host + optional port, without a path") return {"device_id": factory_id, "model": model, "origin": origin, "key_version": version, "device_key_hex": key.lower()} def private_directory(path): """Create a new directory and restrict access before writing secrets.""" path.mkdir(parents=True, exist_ok=False, mode=0o700) if os.name == "nt": who = subprocess.run(["whoami", "/user", "/fo", "csv", "/nh"], capture_output=True, check=True) sid = re.search(rb"S-1-[0-9-]+", who.stdout) if not sid: raise ValueError("Cannot determine the current Windows user SID") subprocess.run(["icacls", str(path), "/inheritance:r", "/grant:r", "*" + sid.group().decode("ascii") + ":(OI)(CI)F"], capture_output=True, check=True) else: os.chmod(path, 0o700) def private_json(path, value): with path.open("x", encoding="utf-8", newline="\n") as stream: json.dump(value, stream, ensure_ascii=False, indent=2) stream.write("\n") os.chmod(path, 0o600) def prepare(credential, output, expected_mac): item = read_credential(credential) expected_mac = normalize_mac(expected_mac) output = output.resolve() # Fail on a missing generator before creating a partially populated directory. import importlib.util if importlib.util.find_spec("esp_idf_nvs_partition_gen") is None: raise ValueError("Missing esp-idf-nvs-partition-gen; use the ESP-IDF Python environment") private_directory(output) password, salt = secrets.token_urlsafe(16), secrets.token_bytes(16) verifier = security_material(item["device_id"], password, salt) csv_path = output / "factory.csv" rows = [("identity", "namespace", "", ""), ("factory_id", "data", "string", item["device_id"]), ("model", "data", "string", item["model"]), ("origin", "data", "string", item["origin"]), ("key_version", "data", "u32", item["key_version"]), ("device_key", "data", "hex2bin", item["device_key_hex"]), ("salt", "data", "hex2bin", salt.hex()), ("verifier", "data", "hex2bin", verifier.hex())] with csv_path.open("x", encoding="utf-8", newline="") as stream: writer = csv.writer(stream) writer.writerow(("key", "type", "encoding", "value")) writer.writerows(rows) os.chmod(csv_path, 0o600) image_path = output / "ble_factory.bin" # Never print generator output: failures could contain a CSV value/secret. generated = subprocess.run([sys.executable, "-m", "esp_idf_nvs_partition_gen", "generate", str(csv_path), str(image_path), hex(FACTORY_SIZE)], capture_output=True) if generated.returncode: raise ValueError("NVS generation failed; private partial output was kept, choose a new output directory on retry") image = image_path.read_bytes() os.chmod(image_path, 0o600) if len(image) != FACTORY_SIZE: raise ValueError("Generator returned an unexpected factory partition size") private_json(output / "setup-label.json", {"device_id": item["device_id"], "model": item["model"], "setup_password": password}) manifest = {"schema_version": 1, "chip": "esp32s3", "flash_size": 0x1000000, "device_id": item["device_id"], "model": item["model"], "origin": item["origin"], "key_version": item["key_version"], "expected_mac": expected_mac, "image": {"file": "ble_factory.bin", "offset": FACTORY_OFFSET, "size": FACTORY_SIZE, "sha256": hashlib.sha256(image).hexdigest()}} private_json(output / "manifest.json", manifest) return manifest def main(): parser = argparse.ArgumentParser(description=__doc__) parser.add_argument("--credential", required=True, type=Path, help="Online *-burn-in-vN.json") parser.add_argument("--output", required=True, type=Path, help="NEW private directory (refuses overwrite)") parser.add_argument("--expect-mac", required=True, help="Physical board eFuse MAC, not the cloud factory ID") args = parser.parse_args() try: manifest = prepare(args.credential, args.output, args.expect_mac) except ValueError as exc: # Validation messages are constant strings; they never embed supplied values. print("Preparation failed: " + str(exc), file=sys.stderr) return 1 except (OSError, subprocess.SubprocessError): print("Preparation failed. Use ESP-IDF Python and a NEW writable output folder; check file permissions. Existing private files were kept.", file=sys.stderr) return 1 print(json.dumps({"result": "prepared_not_flashed", "device_id": manifest["device_id"], "origin": manifest["origin"], "expected_mac": manifest["expected_mac"], "output": str(args.output.resolve())}, ensure_ascii=False, indent=2)) return 0 if __name__ == "__main__": raise SystemExit(main())