n_signer/examples/feather_sign_event.py

#!/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())