Centralize duplicated crypto code

app/fanout/community_mqtt.py

@@ -20,9 +20,8 @@ from datetime import datetime
 from typing import Any, Protocol
 
 import aiomqtt
-import nacl.bindings
-
 from app.fanout.mqtt_base import BaseMqttPublisher
+from app.keystore import ed25519_sign_expanded
 from app.path_utils import parse_packet_envelope, split_path_hex
 from app.version_info import get_app_build_info
 
@@ -40,9 +39,6 @@ _TOKEN_RENEWAL_THRESHOLD = _TOKEN_LIFETIME - 3600  # 23 hours
 _STATS_REFRESH_INTERVAL = 300  # 5 minutes
 _STATS_MIN_CACHE_SECS = 60  # Don't re-fetch stats within 60s
 
-# Ed25519 group order
-_L = 2**252 + 27742317777372353535851937790883648493
-
 # Route type mapping: bottom 2 bits of first byte
 _ROUTE_MAP = {0: "F", 1: "F", 2: "D", 3: "T"}
 
@@ -69,28 +65,6 @@ def _base64url_encode(data: bytes) -> str:
     return base64.urlsafe_b64encode(data).rstrip(b"=").decode("ascii")
 
 
-def _ed25519_sign_expanded(
-    message: bytes, scalar: bytes, prefix: bytes, public_key: bytes
-) -> bytes:
-    """Sign a message using MeshCore's expanded Ed25519 key format.
-
-    MeshCore stores 64-byte "orlp" format keys: scalar(32) || prefix(32).
-    Standard Ed25519 libraries expect seed format and would re-SHA-512 the key.
-    This performs the signing manually using the already-expanded key material.
-
-    Port of meshcore-packet-capture's ed25519_sign_with_expanded_key().
-    """
-    # r = SHA-512(prefix || message) mod L
-    r = int.from_bytes(hashlib.sha512(prefix + message).digest(), "little") % _L
-    # R = r * B (base point multiplication)
-    R = nacl.bindings.crypto_scalarmult_ed25519_base_noclamp(r.to_bytes(32, "little"))
-    # k = SHA-512(R || public_key || message) mod L
-    k = int.from_bytes(hashlib.sha512(R + public_key + message).digest(), "little") % _L
-    # s = (r + k * scalar) mod L
-    s = (r + k * int.from_bytes(scalar, "little")) % _L
-    return R + s.to_bytes(32, "little")
-
-
 def _generate_jwt_token(
     private_key: bytes,
     public_key: bytes,
@@ -127,7 +101,7 @@ def _generate_jwt_token(
 
     scalar = private_key[:32]
     prefix = private_key[32:]
-    signature = _ed25519_sign_expanded(signing_input, scalar, prefix, public_key)
+    signature = ed25519_sign_expanded(signing_input, scalar, prefix, public_key)
 
     return f"{header_b64}.{payload_b64}.{signature.hex()}"
 

app/fanout/map_upload.py

@@ -41,7 +41,7 @@ import httpx
 
 from app.decoder import parse_advertisement, parse_packet
 from app.fanout.base import FanoutModule
-from app.keystore import get_private_key, get_public_key
+from app.keystore import ed25519_sign_expanded, get_private_key, get_public_key
 from app.services.radio_runtime import radio_runtime
 
 logger = logging.getLogger(__name__)
@@ -56,30 +56,6 @@ _REUPLOAD_SECONDS = 3600
 # blocklist so that new roles cannot accidentally start populating the map.
 _ALLOWED_DEVICE_ROLES = {2, 3}
 
-# Ed25519 group order (L)
-_L = 2**252 + 27742317777372353535851937790883648493
-
-
-def _ed25519_sign_expanded(
-    message: bytes, scalar: bytes, prefix: bytes, public_key: bytes
-) -> bytes:
-    """Sign using MeshCore's expanded Ed25519 key format.
-
-    MeshCore stores 64-byte keys as scalar(32) || prefix(32). Standard
-    Ed25519 libraries expect seed format and would re-SHA-512 the key, so
-    we perform the signing manually using the already-expanded key material.
-
-    Mirrors the implementation in app/fanout/community_mqtt.py.
-    """
-    import nacl.bindings
-
-    r = int.from_bytes(hashlib.sha512(prefix + message).digest(), "little") % _L
-    R = nacl.bindings.crypto_scalarmult_ed25519_base_noclamp(r.to_bytes(32, "little"))
-    k = int.from_bytes(hashlib.sha512(R + public_key + message).digest(), "little") % _L
-    s = (r + k * int.from_bytes(scalar, "little")) % _L
-    return R + s.to_bytes(32, "little")
-
-
 def _get_radio_params() -> dict:
     """Read radio frequency parameters from the connected radio's self_info.
 
@@ -278,7 +254,7 @@ class MapUploadModule(FanoutModule):
         data_hash = hashlib.sha256(json_str.encode()).digest()
         scalar = private_key[:32]
         prefix_bytes = private_key[32:]
-        signature = _ed25519_sign_expanded(data_hash, scalar, prefix_bytes, public_key)
+        signature = ed25519_sign_expanded(data_hash, scalar, prefix_bytes, public_key)
 
         request_payload = {
             "data": json_str,

app/keystore.py

@@ -1,14 +1,19 @@
 """
-Ephemeral keystore for storing sensitive keys in memory.
+Ephemeral keystore for storing sensitive keys in memory, plus the Ed25519
+signing primitive used by fanout modules that need to sign requests with the
+radio's own key.
 
 The private key is stored in memory only and is never persisted to disk.
 It's exported from the radio on startup and reconnect, then used for
 server-side decryption of direct messages.
 """
 
+import hashlib
 import logging
 from typing import TYPE_CHECKING
 
+import nacl.bindings
+
 from meshcore import EventType
 
 from app.decoder import derive_public_key
@@ -25,11 +30,32 @@ NO_EVENT_RECEIVED_GUIDANCE = (
     "issue commands to the radio."
 )
 
+# Ed25519 group order (L) — used in the expanded signing primitive below
+_L = 2**252 + 27742317777372353535851937790883648493
+
 # In-memory storage for the private key and derived public key
 _private_key: bytes | None = None
 _public_key: bytes | None = None
 
 
+def ed25519_sign_expanded(
+    message: bytes, scalar: bytes, prefix: bytes, public_key: bytes
+) -> bytes:
+    """Sign a message using MeshCore's expanded Ed25519 key format.
+
+    MeshCore stores 64-byte keys as scalar(32) || prefix(32).  Standard
+    Ed25519 libraries expect seed format and would re-SHA-512 the key, so we
+    perform the signing manually using the already-expanded key material.
+
+    Port of meshcore-packet-capture's ed25519_sign_with_expanded_key().
+    """
+    r = int.from_bytes(hashlib.sha512(prefix + message).digest(), "little") % _L
+    R = nacl.bindings.crypto_scalarmult_ed25519_base_noclamp(r.to_bytes(32, "little"))
+    k = int.from_bytes(hashlib.sha512(R + public_key + message).digest(), "little") % _L
+    s = (r + k * int.from_bytes(scalar, "little")) % _L
+    return R + s.to_bytes(32, "little")
+
+
 def clear_keys() -> None:
     """Clear any stored private/public key material from memory."""
     global _private_key, _public_key

tests/test_community_mqtt.py

@@ -19,11 +19,11 @@ from app.fanout.community_mqtt import (
     _build_status_topic,
     _calculate_packet_hash,
     _decode_packet_fields,
-    _ed25519_sign_expanded,
     _format_raw_packet,
     _generate_jwt_token,
     _get_client_version,
 )
+from app.keystore import ed25519_sign_expanded
 from app.fanout.mqtt_community import (
     _config_to_settings,
     _publish_community_packet,
@@ -173,13 +173,13 @@ class TestEddsaSignExpanded:
     def test_produces_64_byte_signature(self):
         private_key, public_key = _make_test_keys()
         message = b"test message"
-        sig = _ed25519_sign_expanded(message, private_key[:32], private_key[32:], public_key)
+        sig = ed25519_sign_expanded(message, private_key[:32], private_key[32:], public_key)
         assert len(sig) == 64
 
     def test_signature_verifies_with_nacl(self):
         private_key, public_key = _make_test_keys()
         message = b"hello world"
-        sig = _ed25519_sign_expanded(message, private_key[:32], private_key[32:], public_key)
+        sig = ed25519_sign_expanded(message, private_key[:32], private_key[32:], public_key)
 
         signed_message = sig + message
         verified = nacl.bindings.crypto_sign_open(signed_message, public_key)
@@ -187,8 +187,8 @@ class TestEddsaSignExpanded:
 
     def test_different_messages_produce_different_signatures(self):
         private_key, public_key = _make_test_keys()
-        sig1 = _ed25519_sign_expanded(b"msg1", private_key[:32], private_key[32:], public_key)
-        sig2 = _ed25519_sign_expanded(b"msg2", private_key[:32], private_key[32:], public_key)
+        sig1 = ed25519_sign_expanded(b"msg1", private_key[:32], private_key[32:], public_key)
+        sig2 = ed25519_sign_expanded(b"msg2", private_key[:32], private_key[32:], public_key)
         assert sig1 != sig2