Codex Local
bf1233d5df
ci / verify (push) Canceled after 0s
Add customer and operator login, BLE provisioning, device fleet controls, call memory, memoir generation, and refreshed family and maintenance portals. Include API contracts, migrations, firmware sources, tests, assets, and deployment documentation.
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())
|