feat: change protocol implementation

Author: robgee86

src/arduino/app_internal/core/peripherals/__init__.py

@@ -0,0 +1,4 @@
+from .bpp_codec import BPPCodec
+from .bpp_stream_codec import BPPStreamCodec
+
+__all__ = ["BPPCodec", "BPPStreamCodec"]

src/arduino/app_internal/core/peripherals/bpp_codec.py

@@ -0,0 +1,249 @@
+# SPDX-FileCopyrightText: Copyright (C) 2025 ARDUINO SA <http://www.arduino.cc>
+#
+# SPDX-License-Identifier: MPL-2.0
+
+import base64
+import time
+import hashlib
+import hmac
+import secrets
+import struct
+from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305
+
+from arduino.app_utils.logger import Logger
+
+logger = Logger("BPPCodec")
+
+BPP_VERSION = 0x00
+MODE_NONE = 0x00
+MODE_SIGN = 0x01
+MODE_ENC = 0x02
+
+# Big-endian header: [Version:1] [Mode:1] [Timestamp:8] [Random:4]
+HEADER_FORMAT = ">BBQL"
+HEADER_SIZE = struct.calcsize(HEADER_FORMAT)  # 14 bytes
+WINDOW_US = 10_000_000  # 10s in µs
+
+
+class ReplayProtection:
+    """
+    Manages the sliding window replay protection and the temporary cache storing
+    the IVs already seen within the validity window.
+    """
+
+    def __init__(self, window_us: int = WINDOW_US):
+        self.window_us = window_us
+        self.cache: dict[bytes, int] = {}  # IV -> Expiration timestamp
+
+    def check_and_update(self, iv: bytes, timestamp_us: int) -> bool:
+        """
+        Determines if the message is valid by assessing replay attack conditions:
+        timestamp out of validity window and IV reuse.
+        """
+        now = time.time_ns() // 1_000
+
+        # Check time window
+        if abs(now - timestamp_us) > self.window_us:
+            logger.warning(f"Message outside validity window. Drift: {(now - timestamp_us) / 1000}ms")
+            return False
+
+        # Check IV reuse
+        if iv in self.cache:
+            logger.warning("IV reuse detected")
+            return False
+
+        # Prune old entries if cache grows too large
+        if len(self.cache) > 1000:
+            self._prune(now)
+
+        self.cache[iv] = now + self.window_us
+
+        return True
+
+    def _prune(self, now: int):
+        # Remove expired entries
+        expired_keys = [k for k, v in self.cache.items() if now > v]
+        for k in expired_keys:
+            del self.cache[k]
+
+
+class BPPCodec:
+    """
+    Binary Peripheral Protocol (BPP) Codec.
+    Implements a secure container format for peripherals and allows to encode and
+    decode payloads.
+    This codec is intended to be used with message-based protocols, i.e. with builtin
+    message boundaries (e.g., WebSocket). If used with stream-based protocols (e.g.,
+    TCP, BLE, UART), it must be wrapped in BPPStreamCodec.
+
+    The protocol supports three security modes:
+    - Mode 0: No Security;
+    - Mode 1: HMAC-SHA256 Signing, useful for authentication and data integrity;
+    - Mode 2: ChaCha20-Poly1305 Encryption and Signing, providing confidentiality,
+        authentication and data integrity.
+
+    The binary format is as follows:
+
+    [Version (1)] [Mode (1)] [Timestamp (8)] [Random (4)] [Payload (Var)] [AuthTag/Sig (16/32)]
+
+    - Version: Protocol version (currently 0x01).
+    - Mode: Security mode (0x00: None, 0x01: HMAC-SHA256, 0x02: ChaCha20-Poly1305).
+    - Timestamp: Microsecond-precision timestamp (Unix epoch).
+    - Random: 32-bit random value for uniqueness.
+    - Payload: Actual data being transmitted.
+    - AuthTag/Sig: HMAC signature (32 bytes for Mode 1) or AuthTag (16 bytes for Mode 2).
+
+    Text-safe encoding/decoding via Base64URL are also provided.
+    """
+
+    def __init__(self, secret: str = "", enable_encryption: bool = False):
+        """
+        Initialize codec.
+
+        Args:
+            secret: Pre-shared secret. Default: empty (no security).
+            enable_encryption: If True, uses ChaCha20-Poly1305. If False, uses HMAC-SHA256 if
+                secret is provided. Default: False.
+        """
+        self.secret = secret.encode() if secret else b""
+        self.enable_encryption = enable_encryption and bool(secret)
+        self.cc_cipher = None
+
+        if self.enable_encryption:
+            # Derive 32-byte key for ChaCha20
+            key = hashlib.sha256(self.secret).digest()
+            self.cc_cipher = ChaCha20Poly1305(key)
+
+        self.replay_protection = ReplayProtection()
+
+    def encode(self, data: bytes) -> bytes:
+        """
+        Packs data into a BPP message and returns its bytes.
+
+        Args:
+            data: The payload to encode.
+        Returns:
+            The complete BPP message (bytes).
+        """
+        # Assemble the header
+        mode = MODE_ENC if self.enable_encryption else (MODE_SIGN if self.secret else MODE_NONE)
+        timestamp_us = time.time_ns() // 1_000
+        random_val = secrets.randbits(32)
+        header = struct.pack(HEADER_FORMAT, BPP_VERSION, mode, timestamp_us, random_val)
+
+        if mode == MODE_ENC and self.cc_cipher:
+            # Encrypt with ChaCha20-Poly1305, use header as AAD
+            # Note: cryptography lib appends the 16-byte Poly1305 AuthTag automatically
+            iv = header[2:]  # Last 12 bytes of header (Timestamp + Random)
+            encrypted_payload = self.cc_cipher.encrypt(iv, data, header)
+            return header + encrypted_payload
+
+        elif mode == MODE_SIGN and self.secret:
+            # HMAC Signature
+            msg_to_sign = header + data
+            signature = hmac.new(self.secret, msg_to_sign, hashlib.sha256).digest()
+            return header + data + signature
+
+        else:
+            # No Security
+            return header + data
+
+    def decode(self, message: bytes) -> bytes | None:
+        """
+        Unpacks a BPP message and returns its payload.
+
+        Args:
+            message: The complete BPP message to decode.
+        Returns:
+            The decoded payload (bytes) if valid, else None.
+        """
+        if len(message) < HEADER_SIZE:
+            logger.warning("Message too short for header")
+            return None
+
+        try:
+            ver, mode, timestamp_us, random_val = struct.unpack(HEADER_FORMAT, message[:HEADER_SIZE])
+        except struct.error:
+            logger.warning("Header parsing failed")
+            return None
+
+        if ver != BPP_VERSION:
+            logger.warning(f"Unsupported version {ver}")
+            return None
+
+        # Check expected minimum size
+        footer_size = 16 if mode == MODE_ENC else (32 if mode == MODE_SIGN else 0)
+        min_size = HEADER_SIZE + footer_size
+        if len(message) < min_size:
+            logger.warning("Message too short (truncated)")
+            return None
+
+        # Check for downgrade attacks
+        if self.enable_encryption:
+            if mode != MODE_ENC:
+                logger.warning(f"Security mode mismatch: expected Mode 2 (encrypt), but received Mode {mode}.")
+                return None
+        elif self.secret:
+            if mode != MODE_SIGN:
+                logger.warning(f"Security mode mismatch: expected Mode 1 (sign), but received Mode {mode}.")
+                return None
+        else:
+            if mode != MODE_NONE:
+                logger.warning(f"Security mode mismatch: expected Mode 0 (none), but received Mode {mode}.")
+                return None
+
+        # Check for replay attacks
+        replay_id = message[2:HEADER_SIZE]  # Timestamp (8) + Random (4)
+        if not self.replay_protection.check_and_update(replay_id, timestamp_us):
+            return None
+
+        header_bytes = message[:HEADER_SIZE]
+
+        # Decrypt/verify
+        try:
+            if mode == MODE_ENC:
+                iv = replay_id
+                ciphertext_with_tag = message[HEADER_SIZE:]
+                return self.cc_cipher.decrypt(iv, ciphertext_with_tag, header_bytes)
+
+            elif mode == MODE_SIGN:
+                if len(message) < HEADER_SIZE + 32:
+                    return None
+
+                payload = message[HEADER_SIZE:-32]
+                received_sig = message[-32:]
+
+                msg_to_verify = header_bytes + payload
+                expected_sig = hmac.new(self.secret, msg_to_verify, hashlib.sha256).digest()
+
+                if not hmac.compare_digest(received_sig, expected_sig):
+                    logger.warning("HMAC verification failed")
+                    return None
+
+                return payload
+
+            elif mode == MODE_NONE:
+                return message[HEADER_SIZE:]
+
+        except Exception as e:
+            logger.warning(f"Crypto Error: {e}")
+            return None
+
+    def encode_text(self, data: bytes) -> str:
+        """
+        Encodes a text-safe BPP packet to a Base64URL string.
+        """
+        binary_packet = self.encode(data)
+        return base64.b64encode(binary_packet).decode("ascii")
+
+    def decode_text(self, b64_str: str) -> bytes | None:
+        """
+        Decodes a text-safe BPP packet from a Base64URL string.
+        """
+        try:
+            binary_packet = base64.b64decode(b64_str)
+            return self.decode(binary_packet)
+
+        except Exception as e:
+            logger.warning(f"Text decode failed: {e}")
+            return None

src/arduino/app_internal/core/peripherals/bpp_stream_codec.py

@@ -0,0 +1,93 @@
+# SPDX-FileCopyrightText: Copyright (C) 2025 ARDUINO SA <http://www.arduino.cc>
+#
+# SPDX-License-Identifier: MPL-2.0
+
+import struct
+from typing import Iterator
+
+from .bpp_codec import BPPCodec
+
+MAGIC = 0xAA
+HEADER_FORMAT = ">BIB"
+HEADER_SIZE = struct.calcsize(HEADER_FORMAT)  # 6 bytes
+
+
+class BPPStreamCodec:
+    """
+    Wraps a BPPCodec to provide support for stream-based protocols (e.g. TCP,
+    BLE, UART).
+
+    The binary format is as follows:
+
+    [Magic(1)] [Length(4)] [HeaderCRC(1)] [BPP Packet]
+
+    - Magic Byte: 0xAA, marks the start of a BPP frame.
+    - Length: 4-byte big-endian unsigned int indicating the length of the BPP packet.
+    - HeaderCRC: Simple checksum over the Length and Magic byte for header integrity.
+    - BPP Packet: The actual BPP-encoded packet as per BPPCodec.
+    """
+
+    def __init__(self, codec: BPPCodec):
+        self.codec = codec
+        self._buffer = bytearray()
+
+    def encode(self, data: bytes) -> bytes:
+        bpp_packet = self.codec.encode(data)
+        length = len(bpp_packet)
+        checksum = self._calc_header_checksum(length)
+        header = struct.pack(HEADER_FORMAT, MAGIC, length, checksum)
+        return header + bpp_packet
+
+    def decode(self, chunk: bytes = b"") -> Iterator[bytes]:
+        """
+        Ingests a stream chunk and yields all fully decoded BPP payloads found.
+
+        Yields:
+            Decoded payloads (bytes)
+        """
+        if chunk:
+            self._buffer.extend(chunk)
+
+        while True:
+            if not self._buffer:
+                break
+
+            # Look for the Magic byte
+            if self._buffer[0] != MAGIC:
+                try:
+                    idx = self._buffer.index(MAGIC)
+                    del self._buffer[:idx]
+                except ValueError:
+                    self._buffer.clear()
+                    break
+
+            # Do we have a full header?
+            if len(self._buffer) < HEADER_SIZE:
+                break  # Need more data, wait for next chunk
+
+            magic, length, checksum = struct.unpack(HEADER_FORMAT, self._buffer[:HEADER_SIZE])
+            if self._calc_header_checksum(length) != checksum:
+                del self._buffer[0]
+                continue
+
+            total_frame_size = HEADER_SIZE + length
+
+            # Do we have the full frame?
+            if len(self._buffer) < total_frame_size:
+                break  # Need more data, wait for next chunk
+
+            bpp_raw = self._buffer[HEADER_SIZE:total_frame_size]
+            del self._buffer[:total_frame_size]
+
+            # Decode BPP payload and yield if valid
+            payload = self.codec.decode(bytes(bpp_raw))
+            if payload is not None:
+                yield payload
+
+    def _calc_header_checksum(self, length: int) -> int:
+        """ "Calculates simple checksum for header integrity verification."""
+        b0 = (length >> 24) & 0xFF
+        b1 = (length >> 16) & 0xFF
+        b2 = (length >> 8) & 0xFF
+        b3 = length & 0xFF
+        return (MAGIC ^ b0 ^ b1 ^ b2 ^ b3) & 0xFF