Dev

pyproject.toml

@@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta"
 
 [project]
 name = "pymc_core"
-version = "1.0.6"
+version = "1.0.7"
 authors = [
     {name = "Lloyd Newton", email = "lloyd@rightup.co.uk"},
 ]

src/pymc_core/__init__.py

@@ -3,7 +3,7 @@
 Clean, simple API for building mesh network applications.
 """
 
-__version__ = "1.0.6"
+__version__ = "1.0.7"
 
 # Core mesh functionality
 from .node.node import MeshNode

src/pymc_core/hardware/sx1262_wrapper.py

@@ -48,6 +48,7 @@ def __init__(
         is_waveshare: bool = False,
         use_dio3_tcxo: bool = False,
         dio3_tcxo_voltage: float = 1.8,
+        use_dio2_rf: bool = False,
     ):
         """
         Initialize SX1262 radio
@@ -73,6 +74,7 @@ def __init__(
             is_waveshare: Use alternate initialization needed for Waveshare HAT
             use_dio3_tcxo: Enable DIO3 TCXO control (default: False)
             dio3_tcxo_voltage: TCXO reference voltage in volts (default: 1.8)
+            use_dio2_rf: Enable DIO2 as RF switch control (default: False)
         """
         # Check if there's already an active instance and clean it up
         if SX1262Radio._active_instance is not None:
@@ -105,6 +107,7 @@ def __init__(
         self.is_waveshare = is_waveshare
         self.use_dio3_tcxo = use_dio3_tcxo
         self.dio3_tcxo_voltage = dio3_tcxo_voltage
+        self.use_dio2_rf = use_dio2_rf
 
         # State variables
         self.lora: Optional[SX126x] = None
@@ -365,7 +368,11 @@ async def _rx_irq_background_task(self):
                         try:
                             # Use the IRQ status stored by the interrupt handler
                             irqStat = self._last_irq_status
-                            if irqStat & self.lora.IRQ_RX_DONE:
+
+                            # Check CRC error FIRST - if CRC failed, don't read FIFO
+                            if irqStat & self.lora.IRQ_CRC_ERR:
+                                logger.warning("[RX] CRC error detected - discarding packet")
+                            elif irqStat & self.lora.IRQ_RX_DONE:
                                 (
                                     payloadLengthRx,
                                     rxStartBufferPointer,
@@ -407,8 +414,6 @@ async def _rx_irq_background_task(self):
                                         logger.warning("[RX] No RX callback registered!")
                                 else:
                                     logger.warning("[RX] Empty packet received")
-                            elif irqStat & self.lora.IRQ_CRC_ERR:
-                                logger.warning("[RX] CRC error detected")
                             elif irqStat & self.lora.IRQ_TIMEOUT:
                                 logger.warning("[RX] RX timeout detected")
                             elif irqStat & self.lora.IRQ_HEADER_ERR:
@@ -518,8 +523,10 @@ def begin(self) -> bool:
             self.lora._reset = self.reset_pin
             self.lora._busy = self.busy_pin
             self.lora._irq = self.irq_pin_number
-            self.lora._txen = self.txen_pin
-            self.lora._rxen = self.rxen_pin
+            # Pass -1 for TXEN/RXEN to prevent SX126x driver from controlling them
+            # The wrapper handles these pins correctly via _control_tx_rx_pins()
+            self.lora._txen = -1  # Managed by wrapper, not low-level driver
+            self.lora._rxen = -1  # Managed by wrapper, not low-level driver
             self.lora._wake = -1  # Not used
 
             # Setup TXEN pin if needed
@@ -537,6 +544,11 @@ def begin(self) -> bool:
                 else:
                     logger.warning(f"Could not setup RXEN pin {self.rxen_pin}")
 
+            # Ensure TX/RX pins are in default state (RX mode)
+            if self.txen_pin != -1 or self.rxen_pin != -1:
+                self._control_tx_rx_pins(tx_mode=False)
+                logger.debug("TX/RX control pins set to RX mode")
+
             # Setup LED pins if specified
             if self.txled_pin != -1 and not self._txled_pin_setup:
                 if self._gpio_manager.setup_output_pin(self.txled_pin, initial_value=False):
@@ -634,7 +646,9 @@ def begin(self) -> bool:
 
                 self.lora.setRegulatorMode(self.lora.REGULATOR_DC_DC)
                 self.lora.calibrate(0x7F)
-                self.lora.setDio2RfSwitch(False)
+                self.lora.setDio2RfSwitch(self.use_dio2_rf)
+                if self.use_dio2_rf:
+                    logger.info("DIO2 RF switch control enabled")
 
                 # Set packet type and frequency
                 rfFreq = int(self.frequency * 33554432 / 32000000)
@@ -1002,15 +1016,8 @@ async def _restore_rx_mode(self) -> None:
         logger.debug("[TX->RX] Starting RX mode restoration after transmission")
         try:
             if self.lora:
-                self.lora.clearIrqStatus(0xFFFF)
                 self.lora.setStandby(self.lora.STANDBY_RC)
-                await asyncio.sleep(0.05)
-
                 self.lora.request(self.lora.RX_CONTINUOUS)
-
-                await asyncio.sleep(0.05)
-                self.lora.clearIrqStatus(0xFFFF)
-
                 logger.debug("[TX->RX] RX mode restoration completed")
 
         except Exception as e:

src/pymc_core/node/handlers/advert.py

@@ -1,5 +1,5 @@
 import time
-from typing import Optional, Dict, Any
+from typing import Any, Dict, Optional
 
 from ...protocol import Identity, Packet, decode_appdata
 from ...protocol.constants import (
@@ -40,7 +40,8 @@ def _extract_advert_components(self, packet: Packet):
 
         if len(appdata) > MAX_ADVERT_DATA_SIZE:
             self.log(
-                f"Advert appdata too large ({len(appdata)} bytes); truncating to {MAX_ADVERT_DATA_SIZE}"
+                f"Advert appdata too large ({len(appdata)} bytes). "
+                f"Truncating to {MAX_ADVERT_DATA_SIZE}"
             )
             appdata = appdata[:MAX_ADVERT_DATA_SIZE]
 
@@ -51,25 +52,33 @@ def _verify_advert_signature(
     ) -> bool:
         """Verify the cryptographic signature of the advert packet."""
         try:
-
             if len(pubkey) != PUB_KEY_SIZE:
-                self.log(f"Invalid public key length: {len(pubkey)} bytes (expected {PUB_KEY_SIZE})")
+                self.log(
+                    f"Invalid public key length: {len(pubkey)} bytes (expected {PUB_KEY_SIZE})"
+                )
                 return False
-            
+
             if len(signature) != SIGNATURE_SIZE:
-                self.log(f"Invalid signature length: {len(signature)} bytes (expected {SIGNATURE_SIZE})")
+                self.log(
+                    f"Invalid signature length: {len(signature)} bytes (expected {SIGNATURE_SIZE})"
+                )
                 return False
-            
+
             peer_identity = Identity(pubkey)
         except ValueError as exc:
-            self.log(f"Unable to construct peer identity - invalid key format: {exc}")
+            self.log(f"Malformed public key in advert - invalid key format: {exc}")
             return False
         except Exception as exc:
-            self.log(f"Unable to construct peer identity: {type(exc).__name__}: {exc}")
+            exc_type = type(exc).__name__
+            self.log(
+                f"Cryptographic error constructing identity from public key: " f"{exc_type}: {exc}"
+            )
             return False
 
         signed_region = pubkey + timestamp + appdata
         if not peer_identity.verify(signed_region, signature):
+            pubkey_prefix = pubkey[:8].hex()
+            self.log(f"Signature verification failed for advert " f"(pubkey={pubkey_prefix}...)")
             return False
         return True
 
@@ -85,7 +94,9 @@ async def __call__(self, packet: Packet) -> Optional[Dict[str, Any]]:
             pubkey_hex = pubkey_bytes.hex()
 
             # Verify cryptographic signature
-            if not self._verify_advert_signature(pubkey_bytes, timestamp_bytes, appdata, signature_bytes):
+            if not self._verify_advert_signature(
+                pubkey_bytes, timestamp_bytes, appdata, signature_bytes
+            ):
                 self.log(f"Rejecting advert with invalid signature (pubkey={pubkey_hex[:8]}...)")
                 return None
 
@@ -119,8 +130,8 @@ async def __call__(self, packet: Packet) -> Optional[Dict[str, Any]]:
                 "contact_type_id": contact_type_id,
                 "contact_type": contact_type,
                 "timestamp": int(time.time()),
-                "snr": packet._snr if hasattr(packet, '_snr') else 0.0,
-                "rssi": packet._rssi if hasattr(packet, '_rssi') else 0,
+                "snr": packet._snr if hasattr(packet, "_snr") else 0.0,
+                "rssi": packet._rssi if hasattr(packet, "_rssi") else 0,
                 "valid": True,
             }
 

src/pymc_core/protocol/constants.py

@@ -38,7 +38,9 @@
 # ---------------------------------------------------------------------------
 # Payload version values (2 bits)
 # ---------------------------------------------------------------------------
-PAYLOAD_VER_1 = 0x00
+PAYLOAD_VER_1 = 0x00  # Currently supported
+PAYLOAD_VER_2 = 0x01  # Reserved for future use
+MAX_SUPPORTED_PAYLOAD_VERSION = PAYLOAD_VER_2  # Accept versions 0-1
 
 # ---------------------------------------------------------------------------
 # Misc sizes

src/pymc_core/protocol/identity.py

@@ -93,14 +93,33 @@ def __init__(self, seed: Optional[bytes] = None):
         generates a new random key pair.
 
         Args:
-            seed: Optional 32-byte seed for deterministic key generation.
-        """
-        self.signing_key = SigningKey(seed) if seed else SigningKey.generate()
-        self.verify_key = self.signing_key.verify_key
-
-        # Build 64-byte Ed25519 secret key: seed + pub
-        ed25519_pub = self.verify_key.encode()
-        ed25519_sk = self.signing_key.encode() + ed25519_pub
+            seed: Optional 32 or 64-byte seed. 32-byte for standard PyNaCl key generation,
+                  64-byte for MeshCore firmware expanded key format [scalar||nonce].
+        """
+        # Detect MeshCore 64-byte expanded key format
+        if seed and len(seed) == 64:
+            from nacl.bindings import crypto_scalarmult_ed25519_base_noclamp
+
+            # MeshCore format: [32-byte clamped scalar][32-byte nonce]
+            self._firmware_key = seed
+            self.signing_key = None
+
+            # Derive public key from scalar
+            scalar = seed[:32]
+            ed25519_pub = crypto_scalarmult_ed25519_base_noclamp(scalar)
+            self.verify_key = VerifyKey(ed25519_pub)
+
+            # Build ed25519_sk for X25519 conversion (use reconstructed format)
+            ed25519_sk = scalar + ed25519_pub
+        else:
+            # Standard 32-byte seed or None
+            self._firmware_key = None
+            self.signing_key = SigningKey(seed) if seed else SigningKey.generate()
+            self.verify_key = self.signing_key.verify_key
+
+            # Build 64-byte Ed25519 secret key: seed + pub
+            ed25519_pub = self.verify_key.encode()
+            ed25519_sk = self.signing_key.encode() + ed25519_pub
 
         # X25519 keypair for ECDH
         self._x25519_private = CryptoUtils.ed25519_sk_to_x25519(ed25519_sk)
@@ -137,6 +156,20 @@ def get_shared_public_key(self) -> bytes:
         """
         return self._x25519_public
 
+    def get_signing_key_bytes(self) -> bytes:
+        """
+        Get the signing key bytes for this identity.
+
+        For standard keys, returns the 32-byte Ed25519 seed.
+        For firmware keys, returns the 64-byte expanded key format [scalar||nonce].
+
+        Returns:
+            The signing key bytes (32 or 64 bytes depending on key type).
+        """
+        if self._firmware_key:
+            return self._firmware_key
+        return self.signing_key.encode()
+
     def sign(self, message: bytes) -> bytes:
         """
         Sign a message with the Ed25519 private key.
@@ -147,4 +180,37 @@ def sign(self, message: bytes) -> bytes:
         Returns:
             The 64-byte Ed25519 signature.
         """
+        if self._firmware_key:
+            # Use MeshCore/orlp ed25519 signing algorithm
+            import hashlib
+
+            from nacl.bindings import (
+                crypto_core_ed25519_scalar_add,
+                crypto_core_ed25519_scalar_mul,
+                crypto_core_ed25519_scalar_reduce,
+                crypto_scalarmult_ed25519_base_noclamp,
+            )
+
+            scalar = self._firmware_key[:32]
+            nonce_prefix = self._firmware_key[32:64]
+            public_key = self.get_public_key()
+
+            # r = H(nonce_prefix || message)
+            r_hash = hashlib.sha512(nonce_prefix + message).digest()
+            r = crypto_core_ed25519_scalar_reduce(r_hash)
+
+            # R = r * G
+            R_point = crypto_scalarmult_ed25519_base_noclamp(r)
+
+            # h = H(R || pubkey || message)
+            h_hash = hashlib.sha512(R_point + public_key + message).digest()
+            h = crypto_core_ed25519_scalar_reduce(h_hash)
+
+            # s = (h * scalar + r) mod L
+            h_times_scalar = crypto_core_ed25519_scalar_mul(h, scalar)
+            s = crypto_core_ed25519_scalar_add(h_times_scalar, r)
+
+            # Signature is R || s
+            return R_point + s
+
         return self.signing_key.sign(message).signature

src/pymc_core/protocol/packet.py

@@ -2,6 +2,7 @@
 
 from .constants import (
     MAX_PATH_SIZE,
+    MAX_SUPPORTED_PAYLOAD_VERSION,
     PH_ROUTE_MASK,
     PH_TYPE_MASK,
     PH_TYPE_SHIFT,
@@ -319,6 +320,11 @@ def read_from(self, data: ByteString) -> bool:
         self.header = data[idx]
         idx += 1
 
+        # Validate packet version (must match C++ supported versions)
+        version = self.get_payload_ver()
+        if version > MAX_SUPPORTED_PAYLOAD_VERSION:
+            raise ValueError(f"Unsupported packet version: {version}")
+
         # Read transport codes if present
         if self.has_transport_codes():
             self._check_bounds(idx, 4, data_len, "missing transport codes")

src/pymc_core/protocol/packet_builder.py

@@ -170,10 +170,10 @@ def _encode_advert_data(
 
         # Add name if present
         if final_flags & ADVERT_FLAG_HAS_NAME:
-            name_bytes = name.encode("utf-8")[:31] + b"\x00"
-            buf += name_bytes
-        else:
-            buf += bytes(32)
+            name_bytes = name.encode("utf-8")
+            # Copy name bytes up to remaining space in MAX_ADVERT_DATA_SIZE
+            remaining = MAX_ADVERT_DATA_SIZE - len(buf)
+            buf += name_bytes[:remaining]
 
         return bytes(buf)
 

tests/test_basic.py

@@ -2,7 +2,7 @@
 
 
 def test_version():
-    assert __version__ == "1.0.6"
+    assert __version__ == "1.0.7"
 
 
 def test_import():