laantungir/n_signer
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 40 Wiki Activity

v0.1.2 - Teensy 4.1 secp256k1 stack reductions + shared context: reduced WINDOW_A 5->4 (Strauss ecmult tables 8->4 entries, fixes schnorr verify stack overflow) and ECMULT_CONST_GROUP_SIZE 5->4 (ECDH tables 16->8, fixes nip04 crash); added secp256k1_get_shared_context() and reused it in dispatch.cpp ECDSA/schnorr helpers to eliminate per-request secp256k1_context_create/destroy heap fragmentation; reordered test_signer.py to test classical+nostr verbs before PQ verbs and added nostr_sign_event/nip04/nip44 round-trip tests

Browse Source
This commit is contained in:
Laan Tungir
2026-07-27 12:18:38 -04:00
parent 2ba81c4bc8
commit 9282e22b00
7 changed files with 184 additions and 111 deletions
Download Patch File Download Diff File Expand all files Collapse all files
firmware/teensy41/signer/src/dispatch.cpp
+28 -28
View File
@@ -468,25 +468,29 @@ __attribute__((section(".flashmem"))) static int derive_alg_key(fw_alg_t alg, ui
* secp256k1 ECDSA sign/verify (scheme:"ecdsa")
* ==================================================================== */
/* The three secp256k1 helpers below reuse the persistent shared context
* (secp256k1_get_shared_context) instead of malloc/free-ing a fresh
* secp256k1_context per call. The Teensy 4.1 has only ~140 KB of free heap
* and a small DTCM stack; repeated context create/destroy fragments the
* heap and eventually crashes the device (observed during verify after a
* few sign/get_public_key calls). The shared context is created once and
* randomized at first use; it is never destroyed during normal operation. */
__attribute__((section(".flashmem"))) static int secp256k1_ecdsa_sign32(const uint8_t privkey[32],
const uint8_t msg32[32],
uint8_t sig_out[64]) {
const uint8_t msg32[32],
uint8_t sig_out[64]) {
secp256k1_context *ctx;
secp256k1_ecdsa_signature sig;
int rc = -1;
ctx = secp256k1_context_create(SECP256K1_CONTEXT_SIGN);
ctx = secp256k1_get_shared_context();
if (ctx == NULL) {
return -1;
}
if (!secp256k1_ecdsa_sign(ctx, &sig, msg32, privkey, NULL, NULL)) {
goto done;
return -1;
}
secp256k1_ecdsa_signature_serialize_compact(ctx, sig_out, &sig);
rc = 0;
done:
secp256k1_context_destroy(ctx);
return rc;
return 0;
}
__attribute__((section(".flashmem"))) static int secp256k1_ecdsa_verify32(const uint8_t msg32[32],
@@ -495,22 +499,18 @@ __attribute__((section(".flashmem"))) static int secp256k1_ecdsa_verify32(const
secp256k1_context *ctx;
secp256k1_ecdsa_signature sig;
secp256k1_pubkey pub;
int rc = -1;
ctx = secp256k1_context_create(SECP256K1_CONTEXT_VERIFY);
ctx = secp256k1_get_shared_context();
if (ctx == NULL) {
return -1;
}
if (!secp256k1_ecdsa_signature_parse_compact(ctx, &sig, sig64)) {
goto done;
return -1;
}
if (!secp256k1_ec_pubkey_parse(ctx, &pub, pub33, 33)) {
goto done;
return -1;
}
rc = secp256k1_ecdsa_verify(ctx, &sig, msg32, &pub) ? 0 : -1;
done:
secp256k1_context_destroy(ctx);
return rc;
return secp256k1_ecdsa_verify(ctx, &sig, msg32, &pub) ? 0 : -1;
}
__attribute__((section(".flashmem"))) static int secp256k1_priv_to_compressed_pub(const uint8_t priv[32],
@@ -518,23 +518,19 @@ __attribute__((section(".flashmem"))) static int secp256k1_priv_to_compressed_pu
secp256k1_context *ctx;
secp256k1_pubkey pub;
size_t olen = 33;
int rc = -1;
ctx = secp256k1_context_create(SECP256K1_CONTEXT_SIGN);
ctx = secp256k1_get_shared_context();
if (ctx == NULL) {
return -1;
}
if (!secp256k1_ec_pubkey_create(ctx, &pub, priv)) {
goto done;
return -1;
}
if (!secp256k1_ec_pubkey_serialize(ctx, pub33, &olen, &pub,
SECP256K1_EC_COMPRESSED)) {
goto done;
return -1;
}
rc = 0;
done:
secp256k1_context_destroy(ctx);
return rc;
return 0;
}
/* ====================================================================
@@ -1338,9 +1334,12 @@ __attribute__((section(".flashmem"))) static void handle_request(cJSON *req, con
secure_memzero(pub33, sizeof(pub33));
secure_memzero(msg32, sizeof(msg32));
} else {
/* schnorr verify against x-only pubkey */
/* schnorr verify against x-only pubkey.
* Reuse the shared context instead of
* malloc/free per call (heap-fragmentation
* crash fix). */
secp256k1_context *ctx =
secp256k1_context_create(SECP256K1_CONTEXT_VERIFY);
secp256k1_get_shared_context();
secp256k1_xonly_pubkey xonly;
uint8_t pub32[32];
if (ctx != NULL &&
@@ -1352,7 +1351,8 @@ __attribute__((section(".flashmem"))) static void handle_request(cJSON *req, con
&xonly)) {
valid = 1;
}
if (ctx) secp256k1_context_destroy(ctx);
/* Do NOT destroy ctx — it is the shared
* global context. */
secure_memzero(pub32, sizeof(pub32));
}
} else if (alg == FW_ALG_ED25519) {
firmware/teensy41/signer/src/key_derivation.cpp
+7
View File
@@ -126,6 +126,13 @@ void secp256k1_context_teardown(void) {
}
}
/* Public accessor for the persistent shared secp256k1 context. Used by
* dispatch.cpp's ECDSA/schnorr verify helpers so they don't malloc a fresh
* context per request (which fragments the small Teensy 4.1 heap). */
secp256k1_context *secp256k1_get_shared_context(void) {
return create_context();
}
__attribute__((section(".flashmem"))) static int compute_compressed_pubkey(const secp256k1_context *ctx,
const uint8_t priv[32],
uint8_t pub33[33]) {
firmware/teensy41/signer/src/key_derivation.h
+8
View File
@@ -37,6 +37,14 @@ int derive_secp256k1_keys_index(const uint8_t *seed, size_t seed_len,
int schnorr_sign32(const uint8_t privkey[32], const uint8_t msg32[32],
uint8_t sig64[64]);
/* Return the persistent, shared secp256k1 context (SIGN|VERIFY), creating
* it lazily on first call. Reused by all secp256k1 operations (key derivation,
* schnorr sign/verify, ECDSA sign/verify) to avoid per-request
* secp256k1_context_create/destroy, which fragments the small Teensy 4.1
* heap and eventually crashes. Returns NULL on allocation failure. */
struct secp256k1_context_struct;
struct secp256k1_context_struct *secp256k1_get_shared_context(void);
/* --- ed25519 (SSH signatures) --- */
/* Derive an ed25519 keypair from the mnemonic seed using SLIP-0010
firmware/teensy41/signer/src/secp256k1/src/ecmult_impl.h
+6 -2
View File
@@ -28,8 +28,12 @@
# define WINDOW_A 2
# endif
#else
/* optimal for 128-bit and 256-bit exponents. */
# define WINDOW_A 5
/* optimal for 128-bit and 256-bit exponents. Allow the build to override
* this (e.g. the Teensy 4.1 signer sets WINDOW_A=4 to halve the Strauss
* ecmult stack tables; see secp256k1_arduino_config.h). */
# ifndef WINDOW_A
# define WINDOW_A 5
# endif
/** Larger values for ECMULT_WINDOW_SIZE result in possibly better
* performance at the cost of an exponentially larger precomputed
* table. The exact table size is
firmware/teensy41/signer/src/secp256k1/src/secp256k1_arduino_config.h
+12
View File
@@ -60,6 +60,18 @@
#define ECMULT_CONST_GROUP_SIZE 4
#endif
/* WINDOW_A: the Strauss ecmult table size for the na*A term (used by
* secp256k1_ecmult, which backs schnorr verify and ECDSA verify). The
* default (5) allocates secp256k1_fe aux[8] + secp256k1_ge pre_a[8] on
* the stack inside secp256k1_ecmult (~1 KB). On the Teensy 4.1 the DTCM
* stack is only ~9.6 KB free (RAM1 remainder after ITCM), and the
* schnorr-verify call chain (handle_request -> secp256k1_schnorrsig_verify
* -> secp256k1_ecmult) overflows it. WINDOW_A=4 halves the tables (4
* entries each, ~512 B) at a small throughput cost. Must be in [2..24]. */
#ifndef WINDOW_A
#define WINDOW_A 4
#endif
/* Modules required by nostr_core_lib usage */
#ifndef ENABLE_MODULE_ECDH
#define ENABLE_MODULE_ECDH 1
firmware/teensy41/test_signer.py
+121 -79
View File
@@ -3,7 +3,16 @@
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.
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
@@ -22,10 +31,12 @@ 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")
@@ -33,17 +44,14 @@ def send_request(ser, req: dict) -> dict:
ser.write(header + payload)
ser.flush()
# Read response header (4 bytes)
resp_header = b""
while len(resp_header) < 4:
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) == 0 and time.time() % 5 < 0.1:
continue
if len(resp_header) < 4:
raise TimeoutError("Timeout reading response header")
@@ -51,7 +59,6 @@ def send_request(ser, req: dict) -> dict:
if resp_len == 0 or resp_len > 65536:
raise ValueError(f"Invalid response length: {resp_len}")
# Read response payload
resp_payload = b""
while len(resp_payload) < resp_len:
chunk = ser.read(resp_len - len(resp_payload))
@@ -59,14 +66,10 @@ def send_request(ser, req: dict) -> dict:
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]):
"""Send a request and return the result. Prints pass/fail.
params must be a JSON array (the canonical n_signer wire format) or None.
"""
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:
@@ -86,7 +89,6 @@ def test_verb(ser, name, params=None, id_counter=[0]):
return resp
elif "result" in resp:
result = resp["result"]
# Truncate long values for display
display = json.dumps(result, indent=2)
if len(display) > 500:
display = display[:500] + "... (truncated)"
@@ -96,6 +98,7 @@ def test_verb(ser, name, params=None, id_counter=[0]):
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)")
@@ -103,7 +106,7 @@ def main():
print(f"Connecting to {args.port}...")
ser = serial.Serial(args.port, BAUD, timeout=2.0)
time.sleep(2) # Wait for the device to be ready (auto-generate boot)
time.sleep(5) # wait for auto-generate boot + key derivation
# Drain any boot messages
while ser.in_waiting:
@@ -113,17 +116,16 @@ def main():
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 each algorithm
# Canonical format: params = [ { "algorithm": "<alg>", "index": 0 } ]
algorithms = ["secp256k1", "ed25519", "x25519", "ml-dsa-65", "slh-dsa-128s", "ml-kem-768"]
pubkeys = {}
for alg in algorithms:
# 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
@@ -132,10 +134,8 @@ def main():
failed += 1
# 3. sign + verify (secp256k1 schnorr)
# Canonical format: params = [ "<msg_hex>", { "algorithm": "secp256k1", "index": 0, "scheme": "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:
@@ -145,7 +145,6 @@ def main():
failed += 1
if sig_secp and "secp256k1" in pubkeys:
# verify: params = [ "<msg_hex>", "<sig_hex>", { "algorithm": "secp256k1", "index": 0, "scheme": "schnorr" } ]
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
@@ -157,10 +156,8 @@ def main():
failed += 1
# 4. sign (ed25519)
# Canonical format: params = [ "<msg_hex>", { "algorithm": "ed25519", "index": 0 } ]
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:
@@ -180,28 +177,9 @@ def main():
print(" ⏭️ SKIP: no signature or pubkey")
failed += 1
# 5. 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
# 6. sign (slh-dsa-128s) — this is 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
# 7. encrypt + decrypt (OTP pad)
# Canonical format: params = [ "<payload_b64>", { "algorithm": "otp" } ]
# The OTP encrypt/decrypt verbs take a base64 payload and XOR it against
# the bound OTP pad, returning the base64 result + pad offset.
# 5. encrypt + decrypt (OTP pad)
plaintext = b"Hello, OTP pad encryption test!"
pt_b64 = __import__("base64").b64encode(plaintext).decode()
pt_b64 = base64.b64encode(plaintext).decode()
r = test_verb(ser, "encrypt", [pt_b64, {"algorithm": "otp"}])
ct_b64 = None
if r and "result" in r:
@@ -213,7 +191,7 @@ def main():
if ct_b64:
r = test_verb(ser, "decrypt", [ct_b64, {"algorithm": "otp"}])
if r and "result" in r:
pt_result = __import__("base64").b64decode(r["result"].get("result", ""))
pt_result = base64.b64decode(r["result"].get("result", ""))
if pt_result == plaintext:
print(f" ✅ decrypt matches original plaintext")
passed += 1
@@ -227,10 +205,102 @@ def main():
print(" ⏭️ SKIP: no ciphertext")
failed += 1
# 8. encapsulate + decapsulate (ML-KEM-768)
# Canonical format:
# encapsulate: params = [ "<peer_pub_hex>", { "algorithm": "ml-kem-768", "index": 0 } ]
# decapsulate: params = [ "<ct_hex>", { "algorithm": "ml-kem-768", "index": 0 } ]
# 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
# 8. nostr_get_public_key
r = test_verb(ser, "nostr_get_public_key", [{"nostr_index": 0}])
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, {"nostr_index": 0}])
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, {"nostr_index": 0}])
if r and "result" in r:
passed += 1
cipher = r["result"]
r = test_verb(ser, "nostr_nip04_decrypt", [nostr_pub, cipher, {"nostr_index": 0}])
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, {"nostr_index": 0}])
if r and "result" in r:
passed += 1
cipher44 = r["result"]
r = test_verb(ser, "nostr_nip44_decrypt", [nostr_pub, cipher44, {"nostr_index": 0}])
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
# ---- 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
@@ -263,35 +333,6 @@ def main():
print(" ⏭️ SKIP: no ml-kem-768 pubkey")
failed += 1
# 9. derive_shared_secret (X25519) — need a peer pubkey
# Canonical format: params = [ "<peer_pub_hex>", { "algorithm": "x25519", "index": 0 } ]
# The peer pubkey is 32 raw bytes -> 64 hex chars.
try:
from cryptography.hazmat.primitives.asymmetric.x25519 import X25519PrivateKey
peer_priv = X25519PrivateKey.generate()
peer_pub = peer_priv.public_key()
peer_pub_bytes = peer_pub.public_bytes_raw()
peer_pub_hex = peer_pub_bytes.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
# 10. derive (secp256k1 at index 1)
# Canonical format: params = [ "<data>", { "algorithm": "secp256k1", "index": 1 } ]
# index is REQUIRED for derive (no default).
r = test_verb(ser, "derive", ["derive-test-data", {"algorithm": "secp256k1", "index": 1}])
if r and "result" in r:
passed += 1
else:
failed += 1
# Summary
print(f"\n{'='*60}")
print(f"RESULTS: {passed} passed, {failed} failed, {passed+failed} total")
@@ -300,5 +341,6 @@ def main():
ser.close()
return 0 if failed == 0 else 1
if __name__ == "__main__":
sys.exit(main())
src/main.c
+2 -2
View File
@@ -762,8 +762,8 @@ int socket_name_random(char *out, size_t out_len);
/* Version information (auto-updated by build/version tooling) */
#define NSIGNER_VERSION_MAJOR 0
#define NSIGNER_VERSION_MINOR 1
#define NSIGNER_VERSION_PATCH 1
#define NSIGNER_VERSION "v0.1.1"
#define NSIGNER_VERSION_PATCH 2
#define NSIGNER_VERSION "v0.1.2"
/* NSIGNER_HEADERLESS_DECLS_END */