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/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/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: diff --git a/builders/site.py b/builders/site.py index 2ddd16e7..6ac8b0b8 100644 --- a/builders/site.py +++ b/builders/site.py @@ -38,6 +38,12 @@ 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 +except ImportError: + from usb_profiles import artifact_sha256, write_profiles HERE = pathlib.Path(__file__).resolve().parent @@ -54,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: @@ -133,6 +154,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 +215,18 @@ 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")) + layer_revisions = {} + for layer in ("vendors", "arduino", "platformio", "other"): + 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 [], @@ -212,9 +246,20 @@ def orchestrate( "--db", str(public_dir / "boards.db"), "--out", str(public_dir / "usb-ids.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")) + 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, @@ -236,6 +281,10 @@ def orchestrate( "database": "boards.db", "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), + "source_revisions": source_revisions, "loader": "vite", "boards_root": "boards/", } diff --git a/builders/usb_profiles.py b/builders/usb_profiles.py new file mode 100644 index 00000000..11bd0639 --- /dev/null +++ b/builders/usb_profiles.py @@ -0,0 +1,221 @@ +"""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 +import hashlib +import re +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"} +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: + 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 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 () + if isinstance(values, (str, tuple)): + values = [values] + return (normalize_vidpid(v) for v in values) + + +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}") + 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"): + 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": 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)), + } + + +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 = {"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"]) + 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() or None + 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) + rec = dict(rec); rec["match"] = match + vp = key + 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=(",", ":"))) + 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_match(key)[0] != key or not isinstance(entries, list): + raise ValueError(f"invalid identity key {key!r}") + for entry in entries: + 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("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") + 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}") + 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 + + +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 new file mode 100644 index 00000000..6004cd1b --- /dev/null +++ b/tests/test_usb_profiles.py @@ -0,0 +1,51 @@ +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, 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"} + 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", "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 + assert a["boards"]["x"]["aliases"] == ["a", "z"] + 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", "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"] + + +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", "purpose": "runtime"}]}) + + +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)