From 233bf83d2d6705b0180023aa6ab852e853d6e5f5 Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 18:58:42 -0700 Subject: [PATCH 01/11] feat: publish semantic USB transport profiles --- builders/site.py | 7 ++ builders/usb_profiles.py | 132 +++++++++++++++++++++++++++++++++++++ tests/test_usb_profiles.py | 36 ++++++++++ 3 files changed, 175 insertions(+) create mode 100644 builders/usb_profiles.py create mode 100644 tests/test_usb_profiles.py diff --git a/builders/site.py b/builders/site.py index 2ddd16e7..c23a8291 100644 --- a/builders/site.py +++ b/builders/site.py @@ -39,6 +39,8 @@ import subprocess import sys +from usb_profiles import write_profiles + HERE = pathlib.Path(__file__).resolve().parent REPO_ROOT = HERE.parent @@ -133,6 +135,7 @@ def _clean_public(public_dir: pathlib.Path) -> None: "site.db", "usb-ids.json", "usb-vids.proto.zstd", + "usb-profiles.json", "_meta.json", ): f = public_dir / name @@ -193,6 +196,7 @@ def orchestrate( "--out", str(boards_path), ) boards_data = json.loads(boards_path.read_text(encoding="utf-8")) + other_data = json.loads((normalized / "other.json").read_text(encoding="utf-8")) # 6b: stage the per-board JSONs as Vite static assets boards_copied = _copy_board_jsons(boards_data.get("boards") or [], @@ -212,6 +216,8 @@ def orchestrate( "--db", str(public_dir / "boards.db"), "--out", str(public_dir / "usb-ids.json"), ) + write_profiles(boards_data.get("boards") or [], other_data, + public_dir / "usb-profiles.json") # 8: _meta.json (also written into public/ for Vite to serve) merged = json.loads(merged_path.read_text(encoding="utf-8")) @@ -236,6 +242,7 @@ def orchestrate( "database": "boards.db", "usb_ids_download": "usb-ids.json", "usb_vids_proto_zstd": "usb-vids.proto.zstd", + "usb_profiles": "usb-profiles.json", "loader": "vite", "boards_root": "boards/", } diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py new file mode 100644 index 00000000..7194ff4e --- /dev/null +++ b/builders/usb_profiles.py @@ -0,0 +1,132 @@ +"""Build the versioned semantic USB transport profile artifact. + +The artifact is deliberately independent of the legacy usb-ids JSON and +protobuf files: those files keep their wire shape while this adds roles and +board relationships for fbuild's resolver. +""" +from __future__ import annotations + +import json +from typing import Any, Iterable + +SCHEMA_VERSION = 1 +ROLES = {"compile", "runtime", "bootloader", "probe"} +PURPOSES = ROLES + + +def normalize_vidpid(value: Any) -> str: + if isinstance(value, (list, tuple)) and len(value) == 2: + value = f"{value[0]}:{value[1]}" + if not isinstance(value, str) or ":" not in value: + raise ValueError(f"invalid VID:PID {value!r}") + vid, pid = (x.strip().lower().removeprefix("0x") for x in value.split(":", 1)) + if not (1 <= len(vid) <= 4 and 1 <= len(pid) <= 4): + raise ValueError(f"invalid VID:PID {value!r}") + try: + vid_i, pid_i = int(vid, 16), int(pid, 16) + except ValueError as exc: + raise ValueError(f"invalid VID:PID {value!r}") from exc + return f"{vid_i:04x}:{pid_i:04x}" + + +def _pairs(values: Any) -> Iterable[str]: + if not values: + return () + if isinstance(values, (str, tuple)): + values = [values] + return (normalize_vidpid(v) for v in values) + + +def _profile_record(board: dict[str, Any], key: str, role: str, source: str) -> dict[str, Any]: + if role not in ROLES: + raise ValueError(f"invalid USB profile role {role!r}") + return { + "role": role, + "transport": str(board.get("transport", "usb")), + "reset": board.get("reset"), + "handoff": board.get("handoff"), + "platform": board.get("platform") or board.get("core"), + "family": board.get("family") or board.get("mcu"), + "generation": board.get("generation"), + "interface": board.get("interface"), + "provenance": source, + } + + +def build_profiles(boards: list[dict[str, Any]], other: Any = None) -> dict[str, Any]: + identities: dict[str, list[dict[str, Any]]] = {} + board_profiles: dict[str, dict[str, Any]] = {} + + def add(board_id: str, vp: str, role: str, record: dict[str, Any], source: str) -> None: + item = _profile_record(record, vp, role, source) + identities.setdefault(vp, []).append(item) + profile = board_profiles.setdefault(board_id, {"identities": {r: [] for r in PURPOSES}, "aliases": []}) + if vp not in profile["identities"][role]: + profile["identities"][role].append(vp) + + for board in boards: + board_id = str(board.get("board_id", "")).strip() + if not board_id: + continue + source = str(board.get("upstream_blob") or board.get("source") or board.get("layer", "upstream")) + for vp in _pairs(board.get("vidpids")): + for role in (board.get("roles") or ["runtime", "compile"]): + add(board_id, vp, role, board, source) + profile = board_profiles.setdefault(board_id, {"identities": {r: [] for r in PURPOSES}, "aliases": []}) + aliases = board.get("aliases") or [] + profile["aliases"] = sorted({str(x) for x in aliases if str(x).strip()}) + + # Curated special-role records are accepted from the `other` layer. They + # are additive, so collisions/alternates remain visible in identities. + records = (other or {}).get("usb_profiles", []) if isinstance(other, dict) else [] + for rec in records: + board_id = str(rec.get("board_id", "")).strip() + if not board_id: + continue + source = str(rec.get("provenance") or rec.get("source") or "other") + role = rec.get("role", "runtime") + for vp in _pairs(rec.get("vidpids") or rec.get("vidpid")): + add(board_id, vp, role, rec, source) + + for entries in identities.values(): + entries.sort(key=lambda x: json.dumps(x, sort_keys=True, separators=(",", ":"))) + for profile in board_profiles.values(): + for role in PURPOSES: + profile["identities"][role].sort() + return { + "schema_version": SCHEMA_VERSION, + "metadata": { + "artifact": "usb-transport-profiles", + "compatibility": {"usb_ids": "usb-ids.json", "protobuf": "usb-vids.proto.zstd"}, + }, + "identities": dict(sorted(identities.items())), + "boards": dict(sorted(board_profiles.items())), + } + + +def validate_profiles(artifact: dict[str, Any]) -> None: + if artifact.get("schema_version") != SCHEMA_VERSION or not isinstance(artifact.get("metadata"), dict): + raise ValueError("invalid USB profile schema metadata") + for key, entries in artifact.get("identities", {}).items(): + if normalize_vidpid(key) != key or not isinstance(entries, list): + raise ValueError(f"invalid identity key {key!r}") + for entry in entries: + if entry.get("role") not in ROLES or not entry.get("provenance"): + raise ValueError("identity requires a valid role and provenance") + for board_id, profile in artifact.get("boards", {}).items(): + if not board_id or not isinstance(profile.get("identities"), dict): + raise ValueError(f"invalid board profile {board_id!r}") + for role, keys in profile["identities"].items(): + if role not in PURPOSES: + raise ValueError(f"invalid board purpose {role!r}") + for key in keys: + if key not in artifact["identities"]: + raise ValueError(f"board references unknown identity {key}") + + +def write_profiles(boards: list[dict[str, Any]], other: Any, output) -> dict[str, Any]: + artifact = build_profiles(boards, other) + validate_profiles(artifact) + output.parent.mkdir(parents=True, exist_ok=True) + output.write_text(json.dumps(artifact, indent=2, ensure_ascii=False, sort_keys=True) + "\n", encoding="utf-8") + return artifact diff --git a/tests/test_usb_profiles.py b/tests/test_usb_profiles.py new file mode 100644 index 00000000..5f31c966 --- /dev/null +++ b/tests/test_usb_profiles.py @@ -0,0 +1,36 @@ +import json + +import pytest + +from builders.usb_profiles import build_profiles, normalize_vidpid, validate_profiles + + +def test_normalization_and_collision_preserve_provenance(): + artifact = build_profiles([ + {"board_id": "pico", "aliases": ["Pico2040"], "vidpids": [["0x2E8A", "000A"]], "upstream_blob": "pio.json"}, + {"board_id": "other-pico", "vidpids": [["2e8a", "a"]], "upstream_blob": "arduino.json"}, + ]) + assert normalize_vidpid("2E8A:000A") == "2e8a:000a" + assert len(artifact["identities"]["2e8a:000a"]) == 4 + assert {x["provenance"] for x in artifact["identities"]["2e8a:000a"]} == {"pio.json", "arduino.json"} + validate_profiles(artifact) + + +def test_curated_roles_aliases_and_determinism(): + boards = [{"board_id": "x", "aliases": ["z", "a"], "vidpids": ["303a:1001"]}] + other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader", "provenance": "curated#1"}]} + a = build_profiles(boards, other) + b = build_profiles(list(reversed(boards)), other) + assert a == b + assert a["boards"]["x"]["aliases"] == ["a", "z"] + assert a["boards"]["x"]["identities"]["bootloader"] == ["303a:0002"] + + +def test_invalid_role_rejected(): + with pytest.raises(ValueError, match="invalid USB profile role"): + build_profiles([], {"usb_profiles": [{"board_id": "x", "vidpid": "1:2", "role": "bogus"}]}) + + +def test_validation_rejects_unknown_identity(): + with pytest.raises(ValueError): + validate_profiles({"schema_version": 1, "metadata": {}, "identities": {}, "boards": {"x": {"identities": {"runtime": ["1:2"]}}}}) From cff25b5267ce3e5a865262ab834a922f06bbba1c Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 18:59:32 -0700 Subject: [PATCH 02/11] feat: carry curated USB profiles from other data --- builders/extract_other.py | 10 ++++++++-- 1 file changed, 8 insertions(+), 2 deletions(-) diff --git a/builders/extract_other.py b/builders/extract_other.py index 1099f881..7f4011b5 100644 --- a/builders/extract_other.py +++ b/builders/extract_other.py @@ -122,13 +122,15 @@ def extract(root: pathlib.Path) -> dict[str, Any]: vid_overrides: dict[str, str] = {} vidpid_overrides: dict[str, str] = {} vidpid_board_links: list[dict] = [] + usb_profiles: list[dict] = [] if not root.is_dir(): print(f"other: input dir {root} missing; emitting empty record set", file=sys.stderr) return {"layer": "other", "vendors": vendors, "products": products, "vid_overrides": vid_overrides, "vidpid_overrides": vidpid_overrides, - "vidpid_board_links": vidpid_board_links} + "vidpid_board_links": vidpid_board_links, + "usb_profiles": usb_profiles} ovr_path = root / "overrides.json" if ovr_path.is_file(): @@ -157,6 +159,9 @@ def extract(root: pathlib.Path) -> dict[str, Any]: else: print(f"other[skip vidpid_board_link rule #{i}]: malformed", file=sys.stderr) + for rec in ovr.get("usb_profiles") or []: + if isinstance(rec, dict): + usb_profiles.append(rec) print(f"other: {len(vid_overrides)} vid overrides, " f"{len(vidpid_overrides)} vidpid overrides, " f"{len(vidpid_board_links)} board-link rules", @@ -195,7 +200,8 @@ def extract(root: pathlib.Path) -> dict[str, Any]: file=sys.stderr) return {"layer": "other", "vendors": vendors, "products": products, "vid_overrides": vid_overrides, "vidpid_overrides": vidpid_overrides, - "vidpid_board_links": vidpid_board_links} + "vidpid_board_links": vidpid_board_links, + "usb_profiles": usb_profiles} def main() -> int: From f066420f5ca837fc0ee8b758fc952c8f5745ee5f Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:00:56 -0700 Subject: [PATCH 03/11] feat: refine USB profile purposes and semantic roles --- builders/extract_boards.py | 10 +++++++++ builders/site.py | 8 ++++--- builders/usb_profiles.py | 43 +++++++++++++++++++++++++++++--------- tests/test_usb_profiles.py | 14 +++++++++---- 4 files changed, 58 insertions(+), 17 deletions(-) diff --git a/builders/extract_boards.py b/builders/extract_boards.py index ba2f2994..04d6303f 100644 --- a/builders/extract_boards.py +++ b/builders/extract_boards.py @@ -470,12 +470,20 @@ def _extract_platformio(root: pathlib.Path) -> list[dict]: debug = b.get("debug") or {} vidpids: list[list[str]] = [] + identity_purposes: dict[str, list[str]] = {} for entry in (build.get("hwids") or []): if isinstance(entry, (list, tuple)) and len(entry) >= 2: vid = _norm_hex(entry[0], 4) pid = _norm_hex(entry[1], 4) if vid and pid: vidpids.append([vid, pid]) + identity_purposes[f"{vid}:{pid}"] = ["runtime"] + bvid = _norm_hex(build.get("vid", ""), 4) + bpid = _norm_hex(build.get("pid", ""), 4) + if bvid and bpid: + if [bvid, bpid] not in vidpids: + vidpids.append([bvid, bpid]) + identity_purposes.setdefault(f"{bvid}:{bpid}", []).append("compile") mcu_str = _str_or_none(build.get("mcu")) arch, bits = _derive_arch_bits(mcu_str) @@ -507,6 +515,7 @@ def _extract_platformio(root: pathlib.Path) -> list[dict]: "aliases": ",".join(aliases) if aliases else None, "keywords": " ".join(sorted(kw_sink)) if kw_sink else None, "vidpids": vidpids, + "identity_purposes": identity_purposes, "upstream_repo": repo, "upstream_blob": blob, # Source path under the layer's data dir — site.py uses this to @@ -587,6 +596,7 @@ def _extract_arduino(root: pathlib.Path) -> list[dict]: "aliases": ",".join(aliases) if aliases else None, "keywords": " ".join(sorted(kw_sink)) if kw_sink else None, "vidpids": _arduino_vidpids(b), + "identity_purposes": {f"{v}:{p}": ["compile", "runtime"] for v, p in _arduino_vidpids(b)}, "upstream_repo": upstream, "upstream_blob": upstream, "src_relpath": f"{core}/boards/{board_id}.json", diff --git a/builders/site.py b/builders/site.py index c23a8291..f061fc2b 100644 --- a/builders/site.py +++ b/builders/site.py @@ -39,7 +39,7 @@ import subprocess import sys -from usb_profiles import write_profiles +from usb_profiles import artifact_sha256, write_profiles HERE = pathlib.Path(__file__).resolve().parent @@ -216,8 +216,8 @@ def orchestrate( "--db", str(public_dir / "boards.db"), "--out", str(public_dir / "usb-ids.json"), ) - write_profiles(boards_data.get("boards") or [], other_data, - public_dir / "usb-profiles.json") + usb_profiles = write_profiles(boards_data.get("boards") or [], other_data, + public_dir / "usb-profiles.json") # 8: _meta.json (also written into public/ for Vite to serve) merged = json.loads(merged_path.read_text(encoding="utf-8")) @@ -243,6 +243,8 @@ def orchestrate( "usb_ids_download": "usb-ids.json", "usb_vids_proto_zstd": "usb-vids.proto.zstd", "usb_profiles": "usb-profiles.json", + "usb_profiles_schema_version": usb_profiles["schema_version"], + "usb_profiles_sha256": artifact_sha256(usb_profiles), "loader": "vite", "boards_root": "boards/", } diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py index 7194ff4e..01401be6 100644 --- a/builders/usb_profiles.py +++ b/builders/usb_profiles.py @@ -7,11 +7,15 @@ from __future__ import annotations import json +import hashlib from typing import Any, Iterable SCHEMA_VERSION = 1 ROLES = {"compile", "runtime", "bootloader", "probe"} PURPOSES = ROLES +DEVICE_ROLES = {"runtime_cdc", "usb_uart_bridge", "bootloader_msc", "bootloader_hid", "bootloader_dfu", "bootloader_uf2", "debug_probe", "recovery_transport"} +RESETS = {None, "touch-1200", "hardware", "software", "manual"} +HANDOFFS = {None, "reconnect", "reset", "bootloader", "none"} def normalize_vidpid(value: Any) -> str: @@ -37,10 +41,16 @@ def _pairs(values: Any) -> Iterable[str]: return (normalize_vidpid(v) for v in values) -def _profile_record(board: dict[str, Any], key: str, role: str, source: str) -> dict[str, Any]: - if role not in ROLES: +def _profile_record(board: dict[str, Any], key: str, purpose: str, source: str) -> dict[str, Any]: + if purpose not in PURPOSES: + raise ValueError(f"invalid USB profile purpose {purpose!r}") + role = board.get("role") or {"runtime": "runtime_cdc", "compile": "runtime_cdc", "bootloader": "bootloader_uf2", "probe": "debug_probe"}[purpose] + if role not in DEVICE_ROLES: raise ValueError(f"invalid USB profile role {role!r}") + if board.get("reset") not in RESETS or board.get("handoff") not in HANDOFFS: + raise ValueError("invalid USB reset or handoff") return { + "purpose": purpose, "role": role, "transport": str(board.get("transport", "usb")), "reset": board.get("reset"), @@ -69,24 +79,32 @@ def add(board_id: str, vp: str, role: str, record: dict[str, Any], source: str) if not board_id: continue source = str(board.get("upstream_blob") or board.get("source") or board.get("layer", "upstream")) + purposes = board.get("identity_purposes") or {} for vp in _pairs(board.get("vidpids")): - for role in (board.get("roles") or ["runtime", "compile"]): - add(board_id, vp, role, board, source) + key_purposes = purposes.get(vp, ["runtime", "compile"]) + for purpose in key_purposes: + add(board_id, vp, purpose, board, source) profile = board_profiles.setdefault(board_id, {"identities": {r: [] for r in PURPOSES}, "aliases": []}) aliases = board.get("aliases") or [] + if isinstance(aliases, str): + aliases = aliases.split(",") profile["aliases"] = sorted({str(x) for x in aliases if str(x).strip()}) # Curated special-role records are accepted from the `other` layer. They # are additive, so collisions/alternates remain visible in identities. records = (other or {}).get("usb_profiles", []) if isinstance(other, dict) else [] for rec in records: - board_id = str(rec.get("board_id", "")).strip() - if not board_id: - continue + board_id = str(rec.get("board_id", "")).strip() or None source = str(rec.get("provenance") or rec.get("source") or "other") - role = rec.get("role", "runtime") + purpose = rec.get("purpose", "runtime") for vp in _pairs(rec.get("vidpids") or rec.get("vidpid")): - add(board_id, vp, role, rec, source) + if board_id: + add(board_id, vp, purpose, rec, source) + else: + # Generic bridge/probe identities remain addressable even + # without a board profile. + item = _profile_record(rec, vp, purpose, source) + identities.setdefault(vp, []).append(item) for entries in identities.values(): entries.sort(key=lambda x: json.dumps(x, sort_keys=True, separators=(",", ":"))) @@ -111,7 +129,7 @@ def validate_profiles(artifact: dict[str, Any]) -> None: if normalize_vidpid(key) != key or not isinstance(entries, list): raise ValueError(f"invalid identity key {key!r}") for entry in entries: - if entry.get("role") not in ROLES or not entry.get("provenance"): + if entry.get("purpose") not in PURPOSES or entry.get("role") not in DEVICE_ROLES or not entry.get("provenance"): raise ValueError("identity requires a valid role and provenance") for board_id, profile in artifact.get("boards", {}).items(): if not board_id or not isinstance(profile.get("identities"), dict): @@ -130,3 +148,8 @@ def write_profiles(boards: list[dict[str, Any]], other: Any, output) -> dict[str output.parent.mkdir(parents=True, exist_ok=True) output.write_text(json.dumps(artifact, indent=2, ensure_ascii=False, sort_keys=True) + "\n", encoding="utf-8") return artifact + + +def artifact_sha256(artifact: dict[str, Any]) -> str: + raw = json.dumps(artifact, ensure_ascii=False, sort_keys=True, separators=(",", ":")).encode() + return hashlib.sha256(raw).hexdigest() diff --git a/tests/test_usb_profiles.py b/tests/test_usb_profiles.py index 5f31c966..a2adf736 100644 --- a/tests/test_usb_profiles.py +++ b/tests/test_usb_profiles.py @@ -7,18 +7,18 @@ def test_normalization_and_collision_preserve_provenance(): artifact = build_profiles([ - {"board_id": "pico", "aliases": ["Pico2040"], "vidpids": [["0x2E8A", "000A"]], "upstream_blob": "pio.json"}, + {"board_id": "pico", "aliases": "Pico2040, pico-alias", "vidpids": [["0x2E8A", "000A"]], "identity_purposes": {"2e8a:000a": ["runtime"]}, "upstream_blob": "pio.json"}, {"board_id": "other-pico", "vidpids": [["2e8a", "a"]], "upstream_blob": "arduino.json"}, ]) assert normalize_vidpid("2E8A:000A") == "2e8a:000a" - assert len(artifact["identities"]["2e8a:000a"]) == 4 + assert len(artifact["identities"]["2e8a:000a"]) == 3 assert {x["provenance"] for x in artifact["identities"]["2e8a:000a"]} == {"pio.json", "arduino.json"} validate_profiles(artifact) def test_curated_roles_aliases_and_determinism(): boards = [{"board_id": "x", "aliases": ["z", "a"], "vidpids": ["303a:1001"]}] - other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader", "provenance": "curated#1"}]} + other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader_uf2", "purpose": "bootloader", "provenance": "curated#1"}]} a = build_profiles(boards, other) b = build_profiles(list(reversed(boards)), other) assert a == b @@ -26,9 +26,15 @@ def test_curated_roles_aliases_and_determinism(): assert a["boards"]["x"]["identities"]["bootloader"] == ["303a:0002"] +def test_generic_bridge_identity_without_board(): + artifact = build_profiles([], {"usb_profiles": [{"vidpid": "1d50:6018", "role": "usb_uart_bridge", "purpose": "runtime", "provenance": "curated"}]}) + assert "1d50:6018" in artifact["identities"] + assert not artifact["boards"] + + def test_invalid_role_rejected(): with pytest.raises(ValueError, match="invalid USB profile role"): - build_profiles([], {"usb_profiles": [{"board_id": "x", "vidpid": "1:2", "role": "bogus"}]}) + build_profiles([], {"usb_profiles": [{"board_id": "x", "vidpid": "1:2", "role": "bogus", "purpose": "runtime"}]}) def test_validation_rejects_unknown_identity(): From 8426e5d1950df2c13fb4bcc59a374ea1f6b9a189 Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:01:21 -0700 Subject: [PATCH 04/11] fix: support package import for site builder --- builders/site.py | 5 ++++- 1 file changed, 4 insertions(+), 1 deletion(-) diff --git a/builders/site.py b/builders/site.py index f061fc2b..311f0e10 100644 --- a/builders/site.py +++ b/builders/site.py @@ -39,7 +39,10 @@ import subprocess import sys -from usb_profiles import artifact_sha256, write_profiles +try: # direct script execution and package imports are both supported + from .usb_profiles import artifact_sha256, write_profiles +except ImportError: + from usb_profiles import artifact_sha256, write_profiles HERE = pathlib.Path(__file__).resolve().parent From 9f7284ce3723ab0865e7a359f98a03b6f0f08e4e Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:05:04 -0700 Subject: [PATCH 05/11] feat: type USB identity matches and integrity metadata --- builders/usb_profiles.py | 50 ++++++++++++++++++++++++++++++++------ tests/test_usb_profiles.py | 6 ++--- 2 files changed, 45 insertions(+), 11 deletions(-) diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py index 01401be6..18bd2f74 100644 --- a/builders/usb_profiles.py +++ b/builders/usb_profiles.py @@ -33,6 +33,23 @@ def normalize_vidpid(value: Any) -> str: return f"{vid_i:04x}:{pid_i:04x}" +def normalize_match(value: Any) -> tuple[str, dict[str, Any]]: + if isinstance(value, dict): + vid = normalize_vidpid(f"{value.get('vid', '')}:0").split(":")[0] + pid = value.get("pid") + mask = value.get("pid_mask") + if pid in (None, "", "*"): + return f"{vid}:*", {"vid": vid, "pid": None, "pid_mask": mask} + key = normalize_vidpid(f"{vid}:{pid}") + return key, {"vid": vid, "pid": key.split(":")[1], "pid_mask": mask} + if isinstance(value, str) and value.endswith(":*"): + vid = normalize_vidpid(value[:-2] + ":0").split(":")[0] + return f"{vid}:*", {"vid": vid, "pid": None, "pid_mask": None} + key = normalize_vidpid(value) + vid, pid = key.split(":") + return key, {"vid": vid, "pid": pid, "pid_mask": None} + + def _pairs(values: Any) -> Iterable[str]: if not values: return () @@ -41,25 +58,34 @@ def _pairs(values: Any) -> Iterable[str]: return (normalize_vidpid(v) for v in values) -def _profile_record(board: dict[str, Any], key: str, purpose: str, source: str) -> dict[str, Any]: +def _profile_record(board: dict[str, Any], key: str, purpose: str, source: Any) -> dict[str, Any]: if purpose not in PURPOSES: raise ValueError(f"invalid USB profile purpose {purpose!r}") role = board.get("role") or {"runtime": "runtime_cdc", "compile": "runtime_cdc", "bootloader": "bootloader_uf2", "probe": "debug_probe"}[purpose] if role not in DEVICE_ROLES: raise ValueError(f"invalid USB profile role {role!r}") - if board.get("reset") not in RESETS or board.get("handoff") not in HANDOFFS: + reset = board.get("reset", "unknown") + handoff = board.get("handoff", "unknown") + if reset not in RESETS | {"unknown"} or handoff not in HANDOFFS | {"unknown"}: raise ValueError("invalid USB reset or handoff") + if role != "runtime_cdc" and (reset == "unknown" or handoff == "unknown"): + raise ValueError("curated deploy role requires reset and handoff") + if not isinstance(source, dict): + source = {"source_url": str(source), "source_revision": None, "source_class": "upstream"} return { + "match": board.get("match") or {"vid": key.split(":")[0], "pid": key.split(":")[1] if ":" in key and key.split(":")[1] != "*" else None, "pid_mask": None}, "purpose": purpose, "role": role, "transport": str(board.get("transport", "usb")), - "reset": board.get("reset"), - "handoff": board.get("handoff"), + "reset": reset, + "handoff": handoff, "platform": board.get("platform") or board.get("core"), "family": board.get("family") or board.get("mcu"), "generation": board.get("generation"), "interface": board.get("interface"), "provenance": source, + "priority": int(board.get("priority", 0)), + "allow_ambiguous": bool(board.get("allow_ambiguous", source.get("source_class") == "upstream" if isinstance(source, dict) else False)), } @@ -78,7 +104,7 @@ def add(board_id: str, vp: str, role: str, record: dict[str, Any], source: str) board_id = str(board.get("board_id", "")).strip() if not board_id: continue - source = str(board.get("upstream_blob") or board.get("source") or board.get("layer", "upstream")) + source = {"source_url": board.get("upstream_blob"), "source_revision": board.get("source_revision"), "source_class": board.get("layer", "upstream")} purposes = board.get("identity_purposes") or {} for vp in _pairs(board.get("vidpids")): key_purposes = purposes.get(vp, ["runtime", "compile"]) @@ -95,9 +121,12 @@ def add(board_id: str, vp: str, role: str, record: dict[str, Any], source: str) records = (other or {}).get("usb_profiles", []) if isinstance(other, dict) else [] for rec in records: board_id = str(rec.get("board_id", "")).strip() or None - source = str(rec.get("provenance") or rec.get("source") or "other") + source = rec.get("provenance") or {"source_url": rec.get("source_url"), "source_revision": rec.get("source_revision"), "source_class": "other"} purpose = rec.get("purpose", "runtime") - for vp in _pairs(rec.get("vidpids") or rec.get("vidpid")): + for raw in (rec.get("vidpids") or [rec.get("vidpid")]): + key, match = normalize_match(raw) + rec = dict(rec); rec["match"] = match + vp = key if board_id: add(board_id, vp, purpose, rec, source) else: @@ -129,8 +158,13 @@ def validate_profiles(artifact: dict[str, Any]) -> None: if normalize_vidpid(key) != key or not isinstance(entries, list): raise ValueError(f"invalid identity key {key!r}") for entry in entries: - if entry.get("purpose") not in PURPOSES or entry.get("role") not in DEVICE_ROLES or not entry.get("provenance"): + if entry.get("purpose") not in PURPOSES or entry.get("role") not in DEVICE_ROLES or not isinstance(entry.get("provenance"), dict): raise ValueError("identity requires a valid role and provenance") + if entry.get("priority") is None: + raise ValueError("identity requires priority") + priorities = [(e.get("priority"), e.get("purpose"), e.get("role")) for e in entries] + if len(priorities) != len(set(priorities)) and not any(e.get("allow_ambiguous") for e in entries): + raise ValueError(f"ambiguous identity records for {key}") for board_id, profile in artifact.get("boards", {}).items(): if not board_id or not isinstance(profile.get("identities"), dict): raise ValueError(f"invalid board profile {board_id!r}") diff --git a/tests/test_usb_profiles.py b/tests/test_usb_profiles.py index a2adf736..51f9317e 100644 --- a/tests/test_usb_profiles.py +++ b/tests/test_usb_profiles.py @@ -12,13 +12,13 @@ def test_normalization_and_collision_preserve_provenance(): ]) assert normalize_vidpid("2E8A:000A") == "2e8a:000a" assert len(artifact["identities"]["2e8a:000a"]) == 3 - assert {x["provenance"] for x in artifact["identities"]["2e8a:000a"]} == {"pio.json", "arduino.json"} + assert {x["provenance"]["source_url"] for x in artifact["identities"]["2e8a:000a"]} == {"pio.json", "arduino.json"} validate_profiles(artifact) def test_curated_roles_aliases_and_determinism(): boards = [{"board_id": "x", "aliases": ["z", "a"], "vidpids": ["303a:1001"]}] - other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader_uf2", "purpose": "bootloader", "provenance": "curated#1"}]} + other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader_uf2", "purpose": "bootloader", "reset": "touch-1200", "handoff": "bootloader", "provenance": {"source_url": "curated#1", "source_revision": "r1", "source_class": "other"}}]} a = build_profiles(boards, other) b = build_profiles(list(reversed(boards)), other) assert a == b @@ -27,7 +27,7 @@ def test_curated_roles_aliases_and_determinism(): def test_generic_bridge_identity_without_board(): - artifact = build_profiles([], {"usb_profiles": [{"vidpid": "1d50:6018", "role": "usb_uart_bridge", "purpose": "runtime", "provenance": "curated"}]}) + artifact = build_profiles([], {"usb_profiles": [{"vidpid": "1d50:6018", "role": "usb_uart_bridge", "purpose": "runtime", "reset": "touch-1200", "handoff": "reconnect", "provenance": {"source_url": "curated", "source_revision": "r1", "source_class": "other"}}]}) assert "1d50:6018" in artifact["identities"] assert not artifact["boards"] From 2fd5364343aa440d6e02f2e6c7b0040a7cb8b8f4 Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:05:26 -0700 Subject: [PATCH 06/11] docs: specify USB profile integrity handshake --- README.md | 6 ++++++ builders/site.py | 10 ++++++++++ 2 files changed, 16 insertions(+) diff --git a/README.md b/README.md index e2380e9a..f6d36c53 100644 --- a/README.md +++ b/README.md @@ -141,6 +141,12 @@ client-side in sql.js (WASM). Zero server, fully cacheable. ### From curl / scripts +Consumers of `usb-profiles.json` must fetch `_meta.json` first, require the +advertised `usb_profiles_schema_version`, then fetch the artifact and verify +its SHA-256 equals `usb_profiles_sha256`. A failed or mismatched second fetch +must be treated as stale/corrupt data; never mix metadata from one build with +an artifact from another. + Once branches exist, raw JSON is available via: ``` diff --git a/builders/site.py b/builders/site.py index 311f0e10..410361ad 100644 --- a/builders/site.py +++ b/builders/site.py @@ -224,6 +224,15 @@ def orchestrate( # 8: _meta.json (also written into public/ for Vite to serve) merged = json.loads(merged_path.read_text(encoding="utf-8")) + source_revisions = {} + for layer in ("vendors", "arduino", "platformio", "other"): + for candidate in (data_root / layer / "_meta.json", data_root / layer / "data" / "_meta.json"): + if candidate.is_file(): + try: + source_revisions[layer] = json.loads(candidate.read_text(encoding="utf-8")) + except json.JSONDecodeError: + pass + break stats = merged.get("stats", {}) meta = { "schema_version": 5, @@ -248,6 +257,7 @@ def orchestrate( "usb_profiles": "usb-profiles.json", "usb_profiles_schema_version": usb_profiles["schema_version"], "usb_profiles_sha256": artifact_sha256(usb_profiles), + "source_revisions": source_revisions, "loader": "vite", "boards_root": "boards/", } From c02537b602c87ec76507e8f2b1ddee76423d1422 Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:05:42 -0700 Subject: [PATCH 07/11] fix: validate wildcard identity keys --- builders/usb_profiles.py | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py index 18bd2f74..57d59dd2 100644 --- a/builders/usb_profiles.py +++ b/builders/usb_profiles.py @@ -155,7 +155,7 @@ def validate_profiles(artifact: dict[str, Any]) -> None: if artifact.get("schema_version") != SCHEMA_VERSION or not isinstance(artifact.get("metadata"), dict): raise ValueError("invalid USB profile schema metadata") for key, entries in artifact.get("identities", {}).items(): - if normalize_vidpid(key) != key or not isinstance(entries, list): + if normalize_match(key)[0] != key or not isinstance(entries, list): raise ValueError(f"invalid identity key {key!r}") for entry in entries: if entry.get("purpose") not in PURPOSES or entry.get("role") not in DEVICE_ROLES or not isinstance(entry.get("provenance"), dict): From 37ff7415d6cfb19ec2835d39aa2e85b54d1282e0 Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:07:52 -0700 Subject: [PATCH 08/11] fix: validate profile matches and provenance revisions --- builders/usb_profiles.py | 24 ++++++++++++++++++++---- 1 file changed, 20 insertions(+), 4 deletions(-) diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py index 57d59dd2..73381f08 100644 --- a/builders/usb_profiles.py +++ b/builders/usb_profiles.py @@ -16,6 +16,8 @@ DEVICE_ROLES = {"runtime_cdc", "usb_uart_bridge", "bootloader_msc", "bootloader_hid", "bootloader_dfu", "bootloader_uf2", "debug_probe", "recovery_transport"} RESETS = {None, "touch-1200", "hardware", "software", "manual"} HANDOFFS = {None, "reconnect", "reset", "bootloader", "none"} +TRANSPORTS = {"usb", "serial", "hid", "msc", "dfu", "uf2", "jtag", "swd"} +INTERFACES = {None, "cdc", "uart", "msc", "hid", "dfu", "uf2", "jtag", "swd", "unknown"} def normalize_vidpid(value: Any) -> str: @@ -66,6 +68,8 @@ def _profile_record(board: dict[str, Any], key: str, purpose: str, source: Any) raise ValueError(f"invalid USB profile role {role!r}") reset = board.get("reset", "unknown") handoff = board.get("handoff", "unknown") + if board.get("transport", "usb") not in TRANSPORTS or board.get("interface") not in INTERFACES: + raise ValueError("invalid USB transport or interface") if reset not in RESETS | {"unknown"} or handoff not in HANDOFFS | {"unknown"}: raise ValueError("invalid USB reset or handoff") if role != "runtime_cdc" and (reset == "unknown" or handoff == "unknown"): @@ -104,7 +108,7 @@ def add(board_id: str, vp: str, role: str, record: dict[str, Any], source: str) board_id = str(board.get("board_id", "")).strip() if not board_id: continue - source = {"source_url": board.get("upstream_blob"), "source_revision": board.get("source_revision"), "source_class": board.get("layer", "upstream")} + source = {"source_url": board.get("upstream_blob"), "source_revision": board.get("source_revision") or "unknown", "source_class": board.get("layer", "upstream")} purposes = board.get("identity_purposes") or {} for vp in _pairs(board.get("vidpids")): key_purposes = purposes.get(vp, ["runtime", "compile"]) @@ -121,7 +125,7 @@ def add(board_id: str, vp: str, role: str, record: dict[str, Any], source: str) records = (other or {}).get("usb_profiles", []) if isinstance(other, dict) else [] for rec in records: board_id = str(rec.get("board_id", "")).strip() or None - source = rec.get("provenance") or {"source_url": rec.get("source_url"), "source_revision": rec.get("source_revision"), "source_class": "other"} + source = rec.get("provenance") or {"source_url": rec.get("source_url"), "source_revision": rec.get("source_revision") or "unknown", "source_class": "other"} purpose = rec.get("purpose", "runtime") for raw in (rec.get("vidpids") or [rec.get("vidpid")]): key, match = normalize_match(raw) @@ -158,10 +162,22 @@ def validate_profiles(artifact: dict[str, Any]) -> None: if normalize_match(key)[0] != key or not isinstance(entries, list): raise ValueError(f"invalid identity key {key!r}") for entry in entries: - if entry.get("purpose") not in PURPOSES or entry.get("role") not in DEVICE_ROLES or not isinstance(entry.get("provenance"), dict): + match = entry.get("match") + if (entry.get("purpose") not in PURPOSES or entry.get("role") not in DEVICE_ROLES + or not isinstance(entry.get("provenance"), dict) + or not entry["provenance"].get("source_revision") + or not isinstance(match, dict) or not isinstance(match.get("vid"), str)): raise ValueError("identity requires a valid role and provenance") - if entry.get("priority") is None: + if entry.get("priority") is None or not isinstance(entry.get("priority"), int) or not 0 <= entry["priority"] <= 1000: raise ValueError("identity requires priority") + if match.get("pid_mask") is not None: + try: + if len(str(match["pid_mask"])) != 4 or int(str(match["pid_mask"]), 16) == 0: + raise ValueError + except ValueError as exc: + raise ValueError("invalid pid_mask") from exc + if not isinstance(entry.get("allow_ambiguous"), bool): + raise ValueError("allow_ambiguous must be boolean") priorities = [(e.get("priority"), e.get("purpose"), e.get("role")) for e in entries] if len(priorities) != len(set(priorities)) and not any(e.get("allow_ambiguous") for e in entries): raise ValueError(f"ambiguous identity records for {key}") From 7f8e0e69e8bb5135a3b3b2078fe024b8933c5dab Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:08:04 -0700 Subject: [PATCH 09/11] fix: propagate data layer revisions into profiles --- builders/site.py | 11 +++++++++++ 1 file changed, 11 insertions(+) diff --git a/builders/site.py b/builders/site.py index 410361ad..7e18ac28 100644 --- a/builders/site.py +++ b/builders/site.py @@ -200,6 +200,17 @@ def orchestrate( ) boards_data = json.loads(boards_path.read_text(encoding="utf-8")) other_data = json.loads((normalized / "other.json").read_text(encoding="utf-8")) + layer_revisions = {} + for layer in ("vendors", "arduino", "platformio", "other"): + for candidate in (data_root / layer / "_meta.json", data_root / layer / "data" / "_meta.json"): + if candidate.is_file(): + try: + layer_revisions[layer] = json.loads(candidate.read_text(encoding="utf-8")).get("source_revision") or json.loads(candidate.read_text(encoding="utf-8")).get("commit") or "unknown" + except json.JSONDecodeError: + layer_revisions[layer] = "unknown" + break + for board in boards_data.get("boards") or []: + board.setdefault("source_revision", layer_revisions.get(board.get("layer"), "unknown")) # 6b: stage the per-board JSONs as Vite static assets boards_copied = _copy_board_jsons(boards_data.get("boards") or [], From 613f54b9664e55db582da068471a1154c94a35ae Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:09:50 -0700 Subject: [PATCH 10/11] fix: require immutable data layer revisions --- builders/site.py | 30 +++++++++++++++++++++++------- builders/usb_profiles.py | 4 ++++ tests/test_usb_profiles.py | 9 +++++---- 3 files changed, 32 insertions(+), 11 deletions(-) diff --git a/builders/site.py b/builders/site.py index 7e18ac28..6ac8b0b8 100644 --- a/builders/site.py +++ b/builders/site.py @@ -38,6 +38,7 @@ import shutil import subprocess import sys +import subprocess as _subprocess try: # direct script execution and package imports are both supported from .usb_profiles import artifact_sha256, write_profiles @@ -59,6 +60,21 @@ def _run_script(script: pathlib.Path, *args: str) -> None: def _now_iso() -> str: return _dt.datetime.now(_dt.timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ") +def _layer_revision(path: pathlib.Path) -> str: + for candidate in (path / "_meta.json", path / "data" / "_meta.json"): + if candidate.is_file(): + try: + meta = json.loads(candidate.read_text(encoding="utf-8")) + value = meta.get("source_revision") or meta.get("commit") or meta.get("sha") + if value: + return str(value) + except json.JSONDecodeError: + pass + try: + return _subprocess.check_output(["git", "-C", str(path), "rev-parse", "HEAD"], text=True, stderr=_subprocess.DEVNULL).strip() + except (OSError, _subprocess.CalledProcessError): + return "unknown" + def _copy_board_jsons(boards: list[dict], data_root: pathlib.Path, public_dir: pathlib.Path) -> int: @@ -202,15 +218,15 @@ def orchestrate( other_data = json.loads((normalized / "other.json").read_text(encoding="utf-8")) layer_revisions = {} for layer in ("vendors", "arduino", "platformio", "other"): - for candidate in (data_root / layer / "_meta.json", data_root / layer / "data" / "_meta.json"): - if candidate.is_file(): - try: - layer_revisions[layer] = json.loads(candidate.read_text(encoding="utf-8")).get("source_revision") or json.loads(candidate.read_text(encoding="utf-8")).get("commit") or "unknown" - except json.JSONDecodeError: - layer_revisions[layer] = "unknown" - break + layer_revisions[layer] = _layer_revision(data_root / layer) + if layer_revisions[layer] == "unknown": + raise RuntimeError(f"cannot determine immutable revision for data layer {layer}") for board in boards_data.get("boards") or []: board.setdefault("source_revision", layer_revisions.get(board.get("layer"), "unknown")) + for record in other_data.get("usb_profiles") or []: + provenance = record.get("provenance") + if isinstance(provenance, dict): + provenance.setdefault("source_revision", layer_revisions["other"]) # 6b: stage the per-board JSONs as Vite static assets boards_copied = _copy_board_jsons(boards_data.get("boards") or [], diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py index 73381f08..6422e587 100644 --- a/builders/usb_profiles.py +++ b/builders/usb_profiles.py @@ -8,6 +8,7 @@ import json import hashlib +import re from typing import Any, Iterable SCHEMA_VERSION = 1 @@ -168,6 +169,9 @@ def validate_profiles(artifact: dict[str, Any]) -> None: or not entry["provenance"].get("source_revision") or not isinstance(match, dict) or not isinstance(match.get("vid"), str)): raise ValueError("identity requires a valid role and provenance") + revision = str(entry["provenance"].get("source_revision", "")) + if revision.lower() in {"unknown", "main", "master", "head"} or not re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", revision): + raise ValueError("provenance revision must be an immutable commit") if entry.get("priority") is None or not isinstance(entry.get("priority"), int) or not 0 <= entry["priority"] <= 1000: raise ValueError("identity requires priority") if match.get("pid_mask") is not None: diff --git a/tests/test_usb_profiles.py b/tests/test_usb_profiles.py index 51f9317e..96399f73 100644 --- a/tests/test_usb_profiles.py +++ b/tests/test_usb_profiles.py @@ -7,9 +7,10 @@ def test_normalization_and_collision_preserve_provenance(): artifact = build_profiles([ - {"board_id": "pico", "aliases": "Pico2040, pico-alias", "vidpids": [["0x2E8A", "000A"]], "identity_purposes": {"2e8a:000a": ["runtime"]}, "upstream_blob": "pio.json"}, - {"board_id": "other-pico", "vidpids": [["2e8a", "a"]], "upstream_blob": "arduino.json"}, + {"board_id": "pico", "aliases": "Pico2040, pico-alias", "vidpids": [["0x2E8A", "000A"]], "identity_purposes": {"2e8a:000a": ["runtime"]}, "source_revision": "a"*40, "upstream_blob": "pio.json"}, + {"board_id": "other-pico", "vidpids": [["2e8a", "a"]], "source_revision": "b"*40, "upstream_blob": "arduino.json"}, ]) + for board in []: pass assert normalize_vidpid("2E8A:000A") == "2e8a:000a" assert len(artifact["identities"]["2e8a:000a"]) == 3 assert {x["provenance"]["source_url"] for x in artifact["identities"]["2e8a:000a"]} == {"pio.json", "arduino.json"} @@ -18,7 +19,7 @@ def test_normalization_and_collision_preserve_provenance(): def test_curated_roles_aliases_and_determinism(): boards = [{"board_id": "x", "aliases": ["z", "a"], "vidpids": ["303a:1001"]}] - other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader_uf2", "purpose": "bootloader", "reset": "touch-1200", "handoff": "bootloader", "provenance": {"source_url": "curated#1", "source_revision": "r1", "source_class": "other"}}]} + other = {"usb_profiles": [{"board_id": "x", "vidpid": "303a:0002", "role": "bootloader_uf2", "purpose": "bootloader", "reset": "touch-1200", "handoff": "bootloader", "provenance": {"source_url": "curated#1", "source_revision": "c"*40, "source_class": "other"}}]} a = build_profiles(boards, other) b = build_profiles(list(reversed(boards)), other) assert a == b @@ -27,7 +28,7 @@ def test_curated_roles_aliases_and_determinism(): def test_generic_bridge_identity_without_board(): - artifact = build_profiles([], {"usb_profiles": [{"vidpid": "1d50:6018", "role": "usb_uart_bridge", "purpose": "runtime", "reset": "touch-1200", "handoff": "reconnect", "provenance": {"source_url": "curated", "source_revision": "r1", "source_class": "other"}}]}) + artifact = build_profiles([], {"usb_profiles": [{"vidpid": "1d50:6018", "role": "usb_uart_bridge", "purpose": "runtime", "reset": "touch-1200", "handoff": "reconnect", "provenance": {"source_url": "curated", "source_revision": "d"*40, "source_class": "other"}}]}) assert "1d50:6018" in artifact["identities"] assert not artifact["boards"] From 446d98d605691c680ec62ebb3ab851a5e486b480 Mon Sep 17 00:00:00 2001 From: zackees Date: Tue, 14 Jul 2026 19:10:36 -0700 Subject: [PATCH 11/11] fix: validate downloaded USB profile semantics --- builders/usb_profiles.py | 12 ++++++++++++ tests/test_usb_profiles.py | 8 ++++++++ 2 files changed, 20 insertions(+) diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py index 6422e587..11bd0639 100644 --- a/builders/usb_profiles.py +++ b/builders/usb_profiles.py @@ -169,6 +169,18 @@ def validate_profiles(artifact: dict[str, Any]) -> None: or not entry["provenance"].get("source_revision") or not isinstance(match, dict) or not isinstance(match.get("vid"), str)): raise ValueError("identity requires a valid role and provenance") + if entry.get("transport") not in TRANSPORTS or entry.get("interface") not in INTERFACES: + raise ValueError("invalid transport/interface") + if entry.get("reset") not in RESETS | {"unknown"} or entry.get("handoff") not in HANDOFFS | {"unknown"}: + raise ValueError("invalid reset/handoff") + if entry.get("role") != "runtime_cdc" and (entry.get("reset") == "unknown" or entry.get("handoff") == "unknown"): + raise ValueError("deploy-ready identity requires reset/handoff") + vid = match.get("vid", "") + pid = match.get("pid") + if not re.fullmatch(r"[0-9a-fA-F]{4}", vid): + raise ValueError("invalid match VID") + if pid is not None and not re.fullmatch(r"[0-9a-fA-F]{4}", str(pid)): + raise ValueError("invalid match PID") revision = str(entry["provenance"].get("source_revision", "")) if revision.lower() in {"unknown", "main", "master", "head"} or not re.fullmatch(r"[0-9a-fA-F]{40}|[0-9a-fA-F]{64}", revision): raise ValueError("provenance revision must be an immutable commit") diff --git a/tests/test_usb_profiles.py b/tests/test_usb_profiles.py index 96399f73..6004cd1b 100644 --- a/tests/test_usb_profiles.py +++ b/tests/test_usb_profiles.py @@ -41,3 +41,11 @@ def test_invalid_role_rejected(): def test_validation_rejects_unknown_identity(): with pytest.raises(ValueError): validate_profiles({"schema_version": 1, "metadata": {}, "identities": {}, "boards": {"x": {"identities": {"runtime": ["1:2"]}}}}) + + +def test_validation_rejects_tampered_transport_and_match(): + artifact = build_profiles([{"board_id": "x", "vidpids": ["1:2"], "source_revision": "e"*40}]) + entry = artifact["identities"]["0001:0002"][0] + entry["transport"] = "bogus" + with pytest.raises(ValueError): + validate_profiles(artifact)