From 6da3698e73e9c8be715afbc1a70fdd9a8451ba98 Mon Sep 17 00:00:00 2001 From: Daniel Frenkel Date: Fri, 25 Sep 2026 08:12:16 -0700 Subject: [PATCH] Bench mode is gated by the provisioning-token file; DEVICE_KEY_SECRET is gone Keys are minted by the Worker (valar-scopes docs/provisioning-mint.md), so the bench holds a PROVISION_TOKEN in ~/.config/valar-flasher/provision-token instead of the fleet secret. Bench mode refuses in one sentence without the file, checks presence only, and passes the file's PATH to the provisioner with --token-file (one path, never its contents). A test asserts the flasher never reads DEVICE_KEY_SECRET, with a control. Also: bench-loop tests get a 20 s deadline. The first sabotage run (drop --token-file) HUNG instead of going red -- a loop waiting for a count no failing board can reach. With the deadline: 8 red, as predicted; restored, 41 pass. Co-Authored-By: Claude Opus 5.5 (1M context) --- README.md | 14 +++++--- products.json | 2 +- products.local.example.json | 2 +- tests/fake_provisioner.py | 7 +++- tests/test_flasher.py | 68 ++++++++++++++++++++++++++----------- valar_flasher.py | 35 +++++++++++++------ 6 files changed, 91 insertions(+), 37 deletions(-) diff --git a/README.md b/README.md index fd5779d..3f55708 100644 --- a/README.md +++ b/README.md @@ -128,9 +128,15 @@ warning everywhere it is used, and it is scanned like everything else. For the one machine that provisions new boards. Copy `products.local.example.json` to `products.local.json` (gitignored) and point `provisioner.command` at Blipscope's `scripts/provision_one.py`. **Bench mode** -then appears; it refuses to start unless `DEVICE_KEY_SECRET` is set in the -environment the flasher was started from — it is never typed into the window or -stored in any file, and the flasher only checks that it is set. +then appears. + +**Keys are minted by the Worker**, not derived on the bench: nobody holds +`DEVICE_KEY_SECRET`, and nothing here asks for it. What the bench holds is a +**provisioning token**. Put it once, from the password manager, in +`~/.config/valar-flasher/provision-token` +(`%USERPROFILE%\.config\valar-flasher\provision-token` on Windows). Bench mode +refuses to start without it. The flasher only checks that the file exists and hands +its path to the provisioner; the token is never shown in the window or printed. Load a powered hub. Each board gets a tile, keyed by its MAC: **waiting → flashing → provisioning → verifying → DONE**, or **FAILED (reason)**, @@ -140,7 +146,7 @@ is in `provisioned.csv`. The header counts *N of --count this session*, and ever failure is listed again at the end. Console form: ``` -python valar_flasher.py --product Blipscope --bench --count 50 +python valar_flasher.py --product Blipscope --bench --count 50 --ports COM18 ``` **Other boards attached?** Every Valar board shows up with the same Espressif USB diff --git a/products.json b/products.json index c6a0b93..bc09da8 100644 --- a/products.json +++ b/products.json @@ -33,7 +33,7 @@ "anchor": "[build] env=" } }, - "provisioner": { "env": "DEVICE_KEY_SECRET" } + "provisioner": { "auth": "provision-token" } } } } \ No newline at end of file diff --git a/products.local.example.json b/products.local.example.json index 5d7bfcf..9086b51 100644 --- a/products.local.example.json +++ b/products.local.example.json @@ -1,5 +1,5 @@ { - "_comment": "Copy to products.local.json (gitignored) on the ONE machine that provisions boards. It is merged over products.json. Nothing here is a secret: DEVICE_KEY_SECRET is read by the provisioner from the environment and is never written in any file.", + "_comment": "Copy to products.local.json (gitignored) on the ONE machine that provisions boards. It is merged over products.json. Nothing here is a secret. Keys are minted by the Worker; the bench's provisioning token lives in ~/.config/valar-flasher/provision-token (never in this file), and nobody holds DEVICE_KEY_SECRET.", "products": { "Blipscope": { "provisioner": { diff --git a/tests/fake_provisioner.py b/tests/fake_provisioner.py index dfbbf97..54a81da 100644 --- a/tests/fake_provisioner.py +++ b/tests/fake_provisioner.py @@ -13,11 +13,16 @@ port, mac, rest = sys.argv[1], sys.argv[2], sys.argv[3:] log = rest[rest.index("--log") + 1] if "--log" in rest else None +# The real provisioner needs the token FILE's path from the flasher; refuse without it, +# so every DONE in the flasher's tests also proves the flasher passed it. +if "--token-file" not in rest or not os.path.exists(rest[rest.index("--token-file") + 1]): + print("RESULT FAIL no --token-file passed (or the file is missing)") + sys.exit(3) print("STEP write", flush=True) print("STEP verify", flush=True) code = int(os.environ.get("FAKE_PROV_EXIT", "0")) if code != 0: - print("RESULT FAIL key REJECTED (401) -- DEVICE_KEY_SECRET does not match the Worker's") + print("RESULT FAIL 02:00:00:00:00:01: PROVISION_TOKEN rejected (403) -- the token file does not match the Worker's") sys.exit(code) if os.environ.get("FAKE_PROV_LIE") == "1": sys.exit(0) diff --git a/tests/test_flasher.py b/tests/test_flasher.py index 6042fc0..65795e1 100644 --- a/tests/test_flasher.py +++ b/tests/test_flasher.py @@ -6,6 +6,7 @@ """ import hashlib import json +import re import os import secrets import shutil @@ -62,7 +63,24 @@ class Rig(unittest.TestCase): def setUp(self): self.td = tempfile.mkdtemp() self.saved = (vf.FW_ROOT, vf.ESPTOOL, vf.KNOWN_PUBLIC_FILE, vf.PROTECTED_FILE, - vf._port_serials, vf.PORT_ALLOW) + vf._port_serials, vf.PORT_ALLOW, vf.PROVISION_TOKEN_FILE) + # A bench loop that never reaches its count must FAIL, not hang: in production it + # runs until Ctrl-C, so every test gets a deadline that stops it. (Found when a + # sabotage that made every board fail hung the suite instead of turning it red.) + self._orig_loop = vf.bench_loop + + def _deadline_loop(s, stop, *a, **k): + t = threading.Timer(20, stop.set) + t.start() + try: + return self._orig_loop(s, stop, *a, **k) + finally: + t.cancel() + vf.bench_loop = _deadline_loop + # Never this machine's real token file: a test one, present by default. + vf.PROVISION_TOKEN_FILE = os.path.join(self.td, "provision-token") + with open(vf.PROVISION_TOKEN_FILE, "w") as f: + f.write("test-provision-token\n") # Never this machine's real protected list or real ports. vf.PROTECTED_FILE = os.path.join(self.td, "protected-macs.json") vf._port_serials = lambda: {} @@ -82,7 +100,8 @@ def setUp(self): def tearDown(self): (vf.FW_ROOT, vf.ESPTOOL, vf.KNOWN_PUBLIC_FILE, vf.PROTECTED_FILE, - vf._port_serials, vf.PORT_ALLOW) = self.saved + vf._port_serials, vf.PORT_ALLOW, vf.PROVISION_TOKEN_FILE) = self.saved + vf.bench_loop = self._orig_loop shutil.rmtree(self.td, ignore_errors=True) def calls(self): @@ -206,14 +225,9 @@ def setUp(self): super().setUp() self.csv = os.path.join(self.td, "provisioned.csv") self.pcfg = {"select": "variant", "variants": {VNAME: self.vcfg}, - "provisioner": {"env": "DEVICE_KEY_SECRET", + "provisioner": { "command": os.path.join(HERE, "fake_provisioner.py"), "log": self.csv}} - os.environ["DEVICE_KEY_SECRET"] = "test-only" - - def tearDown(self): - os.environ.pop("DEVICE_KEY_SECRET", None) - super().tearDown() def session(self, count=0): return vf.BenchSession("Blipscope", VNAME, self.vcfg, self.pcfg, count=count) @@ -233,7 +247,7 @@ def test_provisioner_exit_nonzero_makes_the_tile_red(self): t = s.tiles["00:00:00:00:00:01"] self.assertEqual(t["state"], "FAILED") self.assertEqual(vf.TILE_COLOR[t["state"]], "#e5534b") - self.assertIn("REJECTED", t["detail"]) + self.assertIn("PROVISION_TOKEN rejected (403)", t["detail"]) self.assertEqual(s.done, 0) self.assertEqual(len(s.failures), 1) @@ -249,15 +263,34 @@ def test_a_mac_already_in_the_csv_is_skipped_untouched(self): self.assertEqual(vf.bench_board("COMX", s), "SKIPPED") self.assertEqual(self.writes(), []) - def test_bench_refuses_in_one_sentence_without_the_env_var(self): - os.environ.pop("DEVICE_KEY_SECRET") + def test_bench_refuses_in_one_sentence_without_the_token_file(self): + self.assertIsNone(vf.bench_refusal(self.pcfg), "CONTROL: with the file, bench mode may start") + os.remove(vf.PROVISION_TOKEN_FILE) why = vf.bench_refusal(self.pcfg) self.assertIsNotNone(why) self.assertEqual(why.count(". "), 0, "one sentence") - self.assertIn("DEVICE_KEY_SECRET", why) + self.assertIn("provisioning token", why) + self.assertIn(vf.PROVISION_TOKEN_FILE, why) + + def test_an_empty_token_file_is_no_token(self): + with open(vf.PROVISION_TOKEN_FILE, "w") as f: + f.write(" \n") + self.assertIsNotNone(vf.bench_refusal(self.pcfg)) + + def test_the_provisioner_gets_the_token_file_path_never_its_contents(self): + argv = vf.provisioner_argv(self.pcfg, "COMX", "00:00:00:00:00:01") + i = argv.index("--token-file") + self.assertEqual(argv[i + 1], vf.PROVISION_TOKEN_FILE) + self.assertNotIn("test-provision-token", " ".join(argv)) + + def test_the_flasher_never_reads_the_device_key_secret(self): + src = open(os.path.join(ROOT, "valar_flasher.py"), encoding="utf-8").read() + read = re.compile(r"(environ|getenv)[^\n]{0,40}DEVICE_KEY_SECRET") + self.assertIsNone(read.search(src)) + self.assertIsNotNone(read.search('os.environ.get("DEVICE_KEY_SECRET")'), "CONTROL: the pattern finds a read") def test_bench_is_not_offered_without_a_local_command(self): - self.assertFalse(vf.bench_available({"provisioner": {"env": "DEVICE_KEY_SECRET"}})) + self.assertFalse(vf.bench_available({"provisioner": {"auth": "provision-token"}})) shipped = json.load(open(os.path.join(ROOT, "products.json")))["products"]["Blipscope"] self.assertNotIn("command", shipped.get("provisioner", {}), "the shipped products.json names no command") @@ -294,7 +327,7 @@ def test_a_bench_run_touches_only_the_named_port(self): vf.PORT_ALLOW = {"COM18"} csv_path = os.path.join(self.td, "provisioned.csv") pcfg = {"select": "variant", "variants": {VNAME: self.vcfg}, - "provisioner": {"env": "DEVICE_KEY_SECRET", + "provisioner": { "command": os.path.join(HERE, "fake_provisioner.py"), "log": csv_path}} s = vf.BenchSession("Blipscope", VNAME, self.vcfg, pcfg, count=1) vf.bench_loop(s, threading.Event(), ports_fn=vf.list_ports, poll=0.01) @@ -314,14 +347,9 @@ def setUp(self): with open(vf.PROTECTED_FILE, "w") as f: json.dump({"macs": {COM6_MAC: "COM6 bench unit", "90:70:69:31:e9:d8": "COM15 Missileer"}}, f) self.pcfg = {"select": "variant", "variants": {VNAME: self.vcfg}, - "provisioner": {"env": "DEVICE_KEY_SECRET", + "provisioner": { "command": os.path.join(HERE, "fake_provisioner.py"), "log": os.path.join(self.td, "provisioned.csv")}} - os.environ["DEVICE_KEY_SECRET"] = "test-only" - - def tearDown(self): - os.environ.pop("DEVICE_KEY_SECRET", None) - super().tearDown() def run_board(self, port): s = vf.BenchSession("Blipscope", VNAME, self.vcfg, self.pcfg) diff --git a/valar_flasher.py b/valar_flasher.py index 7fdd520..be76943 100644 --- a/valar_flasher.py +++ b/valar_flasher.py @@ -13,7 +13,7 @@ "select": "variant" (Blipscope) {"variants": {"": {"chip", "repo", "asset", "manifest", "anchor"}}, - "provisioner": {"env": "DEVICE_KEY_SECRET"}} + "provisioner": {"auth": "provision-token"}} Every Blipscope SKU is the same ESP32-S3, so chip detection CANNOT tell them apart and a wrong pick flashes the wrong board's image. The operator picks the variant, and the chosen variant is shown in large type throughout. @@ -31,9 +31,11 @@ Bench mode (provisioning) appears only for a product whose provisioner is configured in products.local.json (never in the shipped products.json), and it -refuses unless the provisioner's env var (DEVICE_KEY_SECRET) is set in the -environment this program was started from. The secret is never typed, stored or -read here beyond "is it set"; the provisioner reads it itself. +refuses unless the bench's provisioning token is in its file +(~/.config/valar-flasher/provision-token). Keys are MINTED BY THE WORKER; nobody on +the bench holds DEVICE_KEY_SECRET and nothing here reads it. The token is never +typed into the window and never printed; the flasher checks only that the file +exists, and hands its PATH (not its contents) to the provisioner. For each product it checks the repo's latest GitHub release on launch and caches the download (offline? it uses whatever is cached). Runs a small Tk window; falls @@ -568,7 +570,19 @@ def flash_variant(port, product, vname, vcfg, factory_reset, log): # ---- bench mode (provisioning) --------------------------------------------- -BENCH_ENV_DEFAULT = "DEVICE_KEY_SECRET" +# The bench's PROVISION_TOKEN for the Worker's mint route (Blipscope +# docs/provisioning-mint.md). One path, read by the flasher (presence only) and +# passed to the provisioner with --token-file, so the two can never disagree. +PROVISION_TOKEN_FILE = os.environ.get("VALAR_FLASHER_PROVISION_TOKEN") or os.path.join( + os.path.expanduser("~"), ".config", "valar-flasher", "provision-token") + + +def provision_token_present(): + try: + with open(PROVISION_TOKEN_FILE, encoding="utf-8") as f: + return bool(f.read().strip()) + except OSError: + return False def bench_available(pcfg): @@ -579,12 +593,11 @@ def bench_available(pcfg): def bench_refusal(pcfg): """None if bench mode may start, else the one sentence that says why not.""" prov = pcfg.get("provisioner") or {} - env = prov.get("env", BENCH_ENV_DEFAULT) if not prov.get("command"): return "This product has no provisioner configured on this machine (products.local.json)." - if not os.environ.get(env): - return (f"Bench mode needs {env} set in the environment this program was started from; " - f"it is never typed or stored here.") + if not provision_token_present(): + return (f"Bench mode needs the provisioning token in {PROVISION_TOKEN_FILE} -- " + f"put it there once from the password manager.") if prov["command"].endswith(".py") and not os.path.exists(prov["command"]): return f"The provisioner configured in products.local.json does not exist: {prov['command']}" try: @@ -654,7 +667,9 @@ def provisioner_argv(pcfg, port, mac): prov = pcfg.get("provisioner") or {} cmd = prov["command"] head = [PY, cmd] if cmd.endswith(".py") else [cmd] - return head + [port, mac] + list(prov.get("args") or []) + (["--log", prov["log"]] if prov.get("log") else []) + return (head + [port, mac] + list(prov.get("args") or []) + + ["--token-file", PROVISION_TOKEN_FILE] + + (["--log", prov["log"]] if prov.get("log") else [])) def bench_board(port, s):