163 lines
7.7 KiB
Python
163 lines
7.7 KiB
Python
|
"""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())
|