389 lines
14 KiB
Python
389 lines
14 KiB
Python
#!/usr/bin/env python3
|
|
"""Full test suite for the Teensy 4.1 n_signer over USB CDC.
|
|
|
|
Tests all verbs: get_info, get_public_key (all 6 algorithms), sign (secp256k1,
|
|
ed25519, ml-dsa-65, slh-dsa-128s), verify, encrypt/decrypt (OTP pad),
|
|
encapsulate/decapsulate (ML-KEM-768), derive_shared_secret (X25519), derive,
|
|
nostr_get_public_key, nostr_sign_event, nostr_nip04_encrypt/decrypt,
|
|
nostr_nip44_encrypt/decrypt.
|
|
|
|
Ordering note: the PQ keygens (ml-dsa-65, slh-dsa-128s, ml-kem-768) are
|
|
heap-heavy on the Teensy 4.1 (~140 KB free heap). Running all three back to
|
|
back can exhaust/fragment the heap and crash the device. To keep the suite
|
|
useful, the classical + nostr verbs (the common case) are tested FIRST on a
|
|
fresh heap, then the PQ verbs are tested LAST. If a PQ verb crashes the
|
|
device, the earlier results still stand.
|
|
|
|
Sends requests in the canonical n_signer wire format (see api.md §4.3):
|
|
params is a JSON ARRAY of positional arguments, with the options object
|
|
(containing "algorithm", "index", "scheme", ...) as the LAST element.
|
|
Binary payloads (messages, pubkeys, ciphertexts) are hex-encoded.
|
|
|
|
Usage:
|
|
python3 firmware/teensy41/test_signer.py [--port /dev/ttyACM0]
|
|
"""
|
|
|
|
import serial
|
|
import struct
|
|
import json
|
|
import time
|
|
import sys
|
|
import argparse
|
|
import hashlib
|
|
import os
|
|
import base64
|
|
|
|
DEFAULT_PORT = "/dev/ttyACM0"
|
|
BAUD = 115200
|
|
|
|
|
|
def send_request(ser, req: dict) -> dict:
|
|
"""Send a JSON-RPC request with 4-byte big-endian length prefix, read response."""
|
|
payload = json.dumps(req).encode("utf-8")
|
|
header = struct.pack(">I", len(payload))
|
|
ser.write(header + payload)
|
|
ser.flush()
|
|
|
|
resp_header = b""
|
|
deadline = time.time() + 30.0
|
|
while len(resp_header) < 4 and time.time() < deadline:
|
|
chunk = ser.read(4 - len(resp_header))
|
|
if chunk:
|
|
resp_header += chunk
|
|
else:
|
|
time.sleep(0.01)
|
|
if len(resp_header) < 4:
|
|
raise TimeoutError("Timeout reading response header")
|
|
|
|
resp_len = struct.unpack(">I", resp_header)[0]
|
|
if resp_len == 0 or resp_len > 65536:
|
|
raise ValueError(f"Invalid response length: {resp_len}")
|
|
|
|
resp_payload = b""
|
|
while len(resp_payload) < resp_len:
|
|
chunk = ser.read(resp_len - len(resp_payload))
|
|
if chunk:
|
|
resp_payload += chunk
|
|
else:
|
|
time.sleep(0.01)
|
|
return json.loads(resp_payload.decode("utf-8"))
|
|
|
|
|
|
def test_verb(ser, name, params=None, id_counter=[0]):
|
|
id_counter[0] += 1
|
|
req = {"jsonrpc": "2.0", "id": id_counter[0], "method": name}
|
|
if params is not None:
|
|
req["params"] = params
|
|
|
|
print(f"\n--- {name} ---")
|
|
print(f" params: {json.dumps(params, indent=2) if params else '(none)'}")
|
|
|
|
try:
|
|
resp = send_request(ser, req)
|
|
except Exception as e:
|
|
print(f" ❌ FAIL: {e}")
|
|
return None
|
|
|
|
if "error" in resp:
|
|
print(f" ❌ FAIL: error code={resp['error'].get('code')} message={resp['error'].get('message')}")
|
|
return resp
|
|
elif "result" in resp:
|
|
result = resp["result"]
|
|
display = json.dumps(result, indent=2)
|
|
if len(display) > 500:
|
|
display = display[:500] + "... (truncated)"
|
|
print(f" ✅ PASS: {display}")
|
|
return resp
|
|
else:
|
|
print(f" ❌ FAIL: no result or error in response: {resp}")
|
|
return resp
|
|
|
|
|
|
def main():
|
|
parser = argparse.ArgumentParser(description="Test Teensy 4.1 n_signer")
|
|
parser.add_argument("--port", default=DEFAULT_PORT, help="Serial port (default: /dev/ttyACM0)")
|
|
args = parser.parse_args()
|
|
|
|
print(f"Connecting to {args.port}...")
|
|
ser = serial.Serial(args.port, BAUD, timeout=2.0)
|
|
time.sleep(5) # wait for auto-generate boot + key derivation
|
|
|
|
# Drain any boot messages
|
|
while ser.in_waiting:
|
|
boot_msg = ser.read(ser.in_waiting).decode("utf-8", errors="replace")
|
|
print(f"Boot: {boot_msg}", end="")
|
|
print()
|
|
|
|
passed = 0
|
|
failed = 0
|
|
pubkeys = {}
|
|
|
|
# 1. get_info
|
|
r = test_verb(ser, "get_info")
|
|
if r and "result" in r: passed += 1
|
|
else: failed += 1
|
|
|
|
# 2. get_public_key for classical algorithms (lightweight, tested first)
|
|
classical_algs = ["secp256k1", "ed25519", "x25519"]
|
|
for alg in classical_algs:
|
|
r = test_verb(ser, "get_public_key", [{"algorithm": alg, "index": 0}])
|
|
if r and "result" in r:
|
|
passed += 1
|
|
pubkeys[alg] = r["result"].get("public_key", "")
|
|
else:
|
|
failed += 1
|
|
|
|
# 3. sign + verify (secp256k1 schnorr)
|
|
msg32 = hashlib.sha256(b"test message for signing").digest()
|
|
msg_hex = msg32.hex()
|
|
r = test_verb(ser, "sign", [msg_hex, {"algorithm": "secp256k1", "index": 0, "scheme": "schnorr"}])
|
|
sig_secp = None
|
|
if r and "result" in r:
|
|
passed += 1
|
|
sig_secp = r["result"].get("signature", "")
|
|
else:
|
|
failed += 1
|
|
|
|
if sig_secp and "secp256k1" in pubkeys:
|
|
r = test_verb(ser, "verify", [msg_hex, sig_secp, {"algorithm": "secp256k1", "index": 0, "scheme": "schnorr"}])
|
|
if r and "result" in r and r["result"].get("valid") in (True, "true"):
|
|
passed += 1
|
|
else:
|
|
failed += 1
|
|
else:
|
|
print("\n--- verify (secp256k1) ---")
|
|
print(" ⏭️ SKIP: no signature or pubkey")
|
|
failed += 1
|
|
|
|
# 4. sign (ed25519)
|
|
msg_raw = b"test ed25519 message"
|
|
msg_raw_hex = msg_raw.hex()
|
|
r = test_verb(ser, "sign", [msg_raw_hex, {"algorithm": "ed25519", "index": 0}])
|
|
sig_ed = None
|
|
if r and "result" in r:
|
|
passed += 1
|
|
sig_ed = r["result"].get("signature", "")
|
|
else:
|
|
failed += 1
|
|
|
|
if sig_ed and "ed25519" in pubkeys:
|
|
r = test_verb(ser, "verify", [msg_raw_hex, sig_ed, {"algorithm": "ed25519", "index": 0}])
|
|
if r and "result" in r and r["result"].get("valid") in (True, "true"):
|
|
passed += 1
|
|
else:
|
|
failed += 1
|
|
else:
|
|
print("\n--- verify (ed25519) ---")
|
|
print(" ⏭️ SKIP: no signature or pubkey")
|
|
failed += 1
|
|
|
|
# 5. encrypt + decrypt (OTP pad)
|
|
plaintext = b"Hello, OTP pad encryption test!"
|
|
pt_b64 = base64.b64encode(plaintext).decode()
|
|
r = test_verb(ser, "encrypt", [pt_b64, {"algorithm": "otp"}])
|
|
ct_b64 = None
|
|
pad_off_before = None
|
|
if r and "result" in r:
|
|
passed += 1
|
|
ct_b64 = r["result"].get("result", "")
|
|
# The pad offset where this ciphertext's pad slice begins. decrypt
|
|
# must rewind to this offset so the same pad bytes are reused.
|
|
pad_off_before = r["result"].get("pad_offset_before")
|
|
else:
|
|
failed += 1
|
|
|
|
if ct_b64 and pad_off_before is not None:
|
|
r = test_verb(ser, "decrypt",
|
|
[ct_b64, {"algorithm": "otp", "pad_offset": int(pad_off_before)}])
|
|
if r and "result" in r:
|
|
pt_result = base64.b64decode(r["result"].get("result", ""))
|
|
if pt_result == plaintext:
|
|
print(f" ✅ decrypt matches original plaintext")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ decrypt mismatch: expected {plaintext}, got {pt_result}")
|
|
failed += 1
|
|
else:
|
|
failed += 1
|
|
else:
|
|
print("\n--- decrypt (otp) ---")
|
|
print(" ⏭️ SKIP: no ciphertext")
|
|
failed += 1
|
|
|
|
# 6. derive_shared_secret (X25519)
|
|
try:
|
|
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
|
|
peer_priv = X25519PrivateKey.generate()
|
|
peer_pub = peer_priv.public_key()
|
|
peer_pub_hex = peer_pub.public_bytes_raw().hex()
|
|
r = test_verb(ser, "derive_shared_secret", [peer_pub_hex, {"algorithm": "x25519", "index": 0}])
|
|
if r and "result" in r: passed += 1
|
|
else: failed += 1
|
|
except ImportError:
|
|
print("\n--- derive_shared_secret (x25519) ---")
|
|
print(" ⏭️ SKIP: 'cryptography' module not installed")
|
|
failed += 1
|
|
|
|
# 7. derive (secp256k1 at index 1)
|
|
r = test_verb(ser, "derive", ["derive-test-data", {"algorithm": "secp256k1", "index": 1}])
|
|
if r and "result" in r: passed += 1
|
|
else: failed += 1
|
|
|
|
# Role + role_path selector (replaces the deprecated nostr_index).
|
|
main_role = {"role": "main", "role_path": "m/44'/1237'/0'/0/0"}
|
|
|
|
# 8. nostr_get_public_key
|
|
r = test_verb(ser, "nostr_get_public_key", [main_role])
|
|
nostr_pub = None
|
|
if r and "result" in r:
|
|
passed += 1
|
|
nostr_pub = r["result"]
|
|
if isinstance(nostr_pub, dict):
|
|
nostr_pub = nostr_pub.get("public_key", "")
|
|
else:
|
|
failed += 1
|
|
|
|
# 9. nostr_sign_event
|
|
if nostr_pub:
|
|
event = {"kind": 1, "created_at": int(time.time()), "tags": [], "content": "hello from test_signer"}
|
|
r = test_verb(ser, "nostr_sign_event", [event, main_role])
|
|
if r and "result" in r: passed += 1
|
|
else: failed += 1
|
|
|
|
# 10. nostr_nip04_encrypt + decrypt (the bug we fixed)
|
|
if nostr_pub:
|
|
nip04_pt = "hello via nip04"
|
|
r = test_verb(ser, "nostr_nip04_encrypt", [nostr_pub, nip04_pt, main_role])
|
|
if r and "result" in r:
|
|
passed += 1
|
|
cipher = r["result"]
|
|
r = test_verb(ser, "nostr_nip04_decrypt", [nostr_pub, cipher, main_role])
|
|
if r and "result" in r and r["result"] == nip04_pt:
|
|
print(f" ✅ nip04 round-trip plaintext recovered")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ nip04 round-trip mismatch")
|
|
failed += 1
|
|
else:
|
|
failed += 1
|
|
|
|
# 11. nostr_nip44_encrypt + decrypt (the is_nip44 dispatch bug we fixed)
|
|
if nostr_pub:
|
|
nip44_pt = "hello via nip44"
|
|
r = test_verb(ser, "nostr_nip44_encrypt", [nostr_pub, nip44_pt, main_role])
|
|
if r and "result" in r:
|
|
passed += 1
|
|
cipher44 = r["result"]
|
|
r = test_verb(ser, "nostr_nip44_decrypt", [nostr_pub, cipher44, main_role])
|
|
if r and "result" in r and r["result"] == nip44_pt:
|
|
print(f" ✅ nip44 round-trip plaintext recovered")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ nip44 round-trip mismatch")
|
|
failed += 1
|
|
else:
|
|
failed += 1
|
|
|
|
# ---- Role + path authorization error cases ----
|
|
print("\n=== Role + path authorization error cases ===")
|
|
|
|
# 11a. nostr_index is deprecated → error 2006
|
|
r = test_verb(ser, "nostr_get_public_key", [{"nostr_index": 0}])
|
|
if r and "error" in r and r["error"].get("code") == 2006:
|
|
print(f" ✅ nostr_index rejected with error 2006 (deprecated)")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ nostr_index not rejected as expected: {r}")
|
|
failed += 1
|
|
|
|
# 11b. unknown role → error 1002
|
|
r = test_verb(ser, "nostr_get_public_key",
|
|
[{"role": "nonexistent", "role_path": "m/44'/1237'/0'/0/0"}])
|
|
if r and "error" in r and r["error"].get("code") == 1002:
|
|
print(f" ✅ unknown role rejected with error 1002")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ unknown role not rejected as expected: {r}")
|
|
failed += 1
|
|
|
|
# 11c. path out of range → error 2003
|
|
# (Requires a range role; the default "main" role is fixed-path, so this
|
|
# tests a path that doesn't match the fixed template.)
|
|
r = test_verb(ser, "nostr_get_public_key",
|
|
[{"role": "main", "role_path": "m/44'/1237'/999'/0/0"}])
|
|
if r and "error" in r and r["error"].get("code") == 2003:
|
|
print(f" ✅ path mismatch rejected with error 2003")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ path mismatch not rejected as expected: {r}")
|
|
failed += 1
|
|
|
|
# ---- PQ verbs (tested LAST: heap-heavy, may crash the device) ----
|
|
print("\n=== PQ verbs (heap-heavy; tested last) ===")
|
|
|
|
# 12. get_public_key for PQ algorithms
|
|
pq_algs = ["ml-dsa-65", "slh-dsa-128s", "ml-kem-768"]
|
|
for alg in pq_algs:
|
|
r = test_verb(ser, "get_public_key", [{"algorithm": alg, "index": 0}])
|
|
if r and "result" in r:
|
|
passed += 1
|
|
pubkeys[alg] = r["result"].get("public_key", "")
|
|
else:
|
|
failed += 1
|
|
|
|
# 13. sign (ml-dsa-65)
|
|
r = test_verb(ser, "sign", [msg_raw_hex, {"algorithm": "ml-dsa-65", "index": 0}])
|
|
if r and "result" in r: passed += 1
|
|
else: failed += 1
|
|
|
|
# 14. sign (slh-dsa-128s) — slow (~1-2s on Teensy)
|
|
print("\n--- sign (slh-dsa-128s) — may take 1-2 seconds ---")
|
|
r = test_verb(ser, "sign", [msg_raw_hex, {"algorithm": "slh-dsa-128s", "index": 0}])
|
|
if r and "result" in r: passed += 1
|
|
else: failed += 1
|
|
|
|
# 15. encapsulate + decapsulate (ML-KEM-768)
|
|
if "ml-kem-768" in pubkeys:
|
|
r = test_verb(ser, "encapsulate", [pubkeys["ml-kem-768"], {"algorithm": "ml-kem-768", "index": 0}])
|
|
ct_kem = None
|
|
ss_kem = None
|
|
if r and "result" in r:
|
|
passed += 1
|
|
ct_kem = r["result"].get("ciphertext", "")
|
|
ss_kem = r["result"].get("shared_secret", "")
|
|
else:
|
|
failed += 1
|
|
|
|
if ct_kem:
|
|
r = test_verb(ser, "decapsulate", [ct_kem, {"algorithm": "ml-kem-768", "index": 0}])
|
|
if r and "result" in r:
|
|
ss_result = r["result"].get("shared_secret", "")
|
|
if ss_result == ss_kem:
|
|
print(f" ✅ decapsulate shared secret matches encapsulate")
|
|
passed += 1
|
|
else:
|
|
print(f" ❌ shared secret mismatch")
|
|
failed += 1
|
|
else:
|
|
failed += 1
|
|
else:
|
|
print("\n--- decapsulate (ml-kem-768) ---")
|
|
print(" ⏭️ SKIP: no encapsulate ciphertext")
|
|
failed += 1
|
|
else:
|
|
print("\n--- encapsulate (ml-kem-768) ---")
|
|
print(" ⏭️ SKIP: no ml-kem-768 pubkey")
|
|
failed += 1
|
|
|
|
# Summary
|
|
print(f"\n{'='*60}")
|
|
print(f"RESULTS: {passed} passed, {failed} failed, {passed+failed} total")
|
|
print(f"{'='*60}")
|
|
|
|
ser.close()
|
|
return 0 if failed == 0 else 1
|
|
|
|
|
|
if __name__ == "__main__":
|
|
sys.exit(main())
|