v0.0.28 - Finalize Feather TinyUSB migration and remove cardputer artifacts

examples/feather_sign_event.py

@@ -0,0 +1,199 @@
+#!/usr/bin/env python3
+import hashlib
+import json
+import struct
+import sys
+import time
+
+import serial
+from coincurve import PrivateKey, PublicKeyXOnly
+
+
+def read_exact(ser: serial.Serial, n: int, timeout_s: float = 8.0) -> bytes:
+    end = time.time() + timeout_s
+    out = bytearray()
+    while len(out) < n and time.time() < end:
+        chunk = ser.read(n - len(out))
+        if chunk:
+            out.extend(chunk)
+            continue
+        time.sleep(0.005)
+    return bytes(out)
+
+
+def recv_frame(ser: serial.Serial, timeout_s: float = 10.0) -> bytes:
+    end = time.time() + timeout_s
+    window = bytearray()
+
+    while time.time() < end:
+        b = ser.read(1)
+        if not b:
+            time.sleep(0.005)
+            continue
+
+        window.extend(b)
+        if len(window) > 4:
+            window.pop(0)
+
+        if len(window) < 4:
+            continue
+
+        (n,) = struct.unpack(">I", bytes(window))
+        if n == 0 or n > 1_000_000:
+            continue
+
+        body = read_exact(ser, n, timeout_s=max(0.2, end - time.time()))
+        if len(body) != n:
+            return b""
+        if body[:1] not in (b"{", b"["):
+            continue
+
+        try:
+            json.loads(body.decode("utf-8", errors="strict"))
+        except Exception:
+            # Probably matched a false header inside stream noise; keep scanning.
+            continue
+
+        return body
+
+    return b""
+
+
+def nostr_serialize(pubkey: str, created_at: int, kind: int, tags: list, content: str) -> bytes:
+    arr = [0, pubkey, created_at, kind, tags, content]
+    return json.dumps(arr, separators=(",", ":"), ensure_ascii=False).encode("utf-8")
+
+
+def event_id_hex(event_obj: dict) -> str:
+    ser = nostr_serialize(
+        event_obj["pubkey"],
+        int(event_obj["created_at"]),
+        int(event_obj["kind"]),
+        event_obj["tags"],
+        event_obj["content"],
+    )
+    return hashlib.sha256(ser).hexdigest()
+
+
+def verify_event(event_obj: dict) -> bool:
+    ev_id = event_id_hex(event_obj)
+    if ev_id != event_obj.get("id"):
+        return False
+
+    pk = PublicKeyXOnly(bytes.fromhex(event_obj["pubkey"]))
+    sig = bytes.fromhex(event_obj["sig"])
+    return pk.verify(sig, bytes.fromhex(ev_id))
+
+
+def build_auth_envelope(method: str, params, caller_priv: bytes, label: str = "py-cli") -> dict:
+    body_json = json.dumps(params, separators=(",", ":")).encode("utf-8")
+    body_hash = hashlib.sha256(body_json).hexdigest()
+
+    sk = PrivateKey(caller_priv)
+    caller_pub_x = sk.public_key.format(compressed=True)[1:].hex()
+    created_at = int(time.time())
+    tags = [
+        ["nsigner_rpc", "1"],
+        ["nsigner_method", method],
+        ["nsigner_body_hash", body_hash],
+    ]
+    serialized = json.dumps(
+        [0, caller_pub_x, created_at, 27235, tags, label],
+        separators=(",", ":"),
+        ensure_ascii=False,
+    ).encode("utf-8")
+    event_id = hashlib.sha256(serialized).hexdigest()
+    sig = sk.sign_schnorr(bytes.fromhex(event_id), aux_randomness=b"\x00" * 32).hex()
+
+    return {
+        "id": event_id,
+        "pubkey": caller_pub_x,
+        "created_at": created_at,
+        "kind": 27235,
+        "tags": tags,
+        "content": label,
+        "sig": sig,
+    }
+
+
+def main() -> int:
+    args = [a for a in sys.argv[1:]]
+    no_auth = False
+    if "--no-auth" in args:
+        no_auth = True
+        args.remove("--no-auth")
+    use_alt_caller = "--alt-caller" in args
+    if use_alt_caller:
+        args.remove("--alt-caller")
+    port = args[0] if args else "/dev/ttyACM0"
+
+    caller_priv = bytes(range(1, 33)) if not use_alt_caller else bytes(range(33, 65))
+
+    unsigned_event = {
+        "kind": 1,
+        "created_at": int(time.time()),
+        "tags": [],
+        "content": "hello from feather phase4",
+    }
+
+    params = [unsigned_event]
+    req = {
+        "jsonrpc": "2.0",
+        "id": "2",
+        "method": "sign_event",
+        "params": params,
+    }
+    if not no_auth:
+        req["auth"] = build_auth_envelope("sign_event", params, caller_priv)
+
+    body = json.dumps(req, separators=(",", ":")).encode("utf-8")
+    frame = struct.pack(">I", len(body)) + body
+
+    ser = serial.Serial()
+    ser.port = port
+    ser.baudrate = 115200
+    ser.timeout = 0.2
+    ser.rtscts = False
+    ser.dsrdtr = False
+    # Set modem-control lines before open to avoid auto-reset pulses.
+    ser.dtr = False
+    ser.rts = False
+
+    with ser:
+        # Let board fully settle after USB open/reconnect.
+        time.sleep(3.0)
+        ser.reset_input_buffer()
+        ser.write(frame)
+        ser.flush()
+
+        resp_raw = recv_frame(ser, timeout_s=45.0)
+        if not resp_raw:
+            print("No framed response received")
+            return 1
+
+        try:
+            resp = json.loads(resp_raw.decode("utf-8", errors="strict"))
+        except Exception:
+            print("Invalid JSON response bytes:", resp_raw)
+            return 1
+        if "error" in resp:
+            print("RPC error:", json.dumps(resp["error"], separators=(",", ":")))
+            return 1
+
+        result = resp.get("result")
+        if not isinstance(result, dict):
+            print("Invalid RPC result")
+            return 1
+
+        ok = verify_event(result)
+        print(json.dumps(result, separators=(",", ":")))
+        if not ok:
+            print("signature invalid")
+            return 1
+
+        print("✓ signature valid")
+        return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())