137 lines
3.6 KiB
Python
137 lines
3.6 KiB
Python
#!/usr/bin/env python3
|
|
import hashlib
|
|
import json
|
|
import os
|
|
import struct
|
|
import sys
|
|
import time
|
|
|
|
import serial
|
|
from coincurve import PrivateKey
|
|
|
|
|
|
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))
|
|
# Keep this cap small to avoid getting stuck on false headers in stream noise.
|
|
if n == 0 or n > 64 * 1024:
|
|
continue
|
|
body = read_exact(ser, n, timeout_s=max(0.2, end - time.time()))
|
|
if len(body) != n:
|
|
continue
|
|
if body[:1] not in (b"{", b"["):
|
|
continue
|
|
try:
|
|
json.loads(body.decode("utf-8", errors="strict"))
|
|
except Exception:
|
|
continue
|
|
return body
|
|
|
|
return b""
|
|
|
|
|
|
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:
|
|
port = sys.argv[1] if len(sys.argv) > 1 else "/dev/ttyACM0"
|
|
|
|
caller_priv = bytes(range(1, 33))
|
|
params = []
|
|
|
|
req = {
|
|
"jsonrpc": "2.0",
|
|
"id": "1",
|
|
"method": "get_public_key",
|
|
"params": params,
|
|
"auth": build_auth_envelope("get_public_key", 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 the board settle if OS still caused a reconnect.
|
|
time.sleep(3.0)
|
|
|
|
for attempt in range(1, 4):
|
|
ser.reset_input_buffer()
|
|
ser.write(frame)
|
|
ser.flush()
|
|
|
|
resp_body = recv_frame(ser, timeout_s=10.0)
|
|
if resp_body:
|
|
print(resp_body.decode("utf-8", errors="replace"))
|
|
return 0
|
|
|
|
print(f"No framed response received (attempt {attempt}/3)")
|
|
time.sleep(0.5)
|
|
|
|
return 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|