Allow one encryption counter reset per 100 messages to allow changing…

pyproject.toml

@@ -1,6 +1,6 @@
 [tool.poetry]
 name = "bthome-ble"
-version = "3.8.1"
+version = "3.8.0"
 description = "BThome BLE support"
 authors = ["Ernst Klamer <e.klamer@gmail.com>"]
 license = "MIT"

src/bthome_ble/__init__.py

@@ -15,7 +15,7 @@
 
 from .parser import BTHomeBluetoothDeviceData
 
-__version__ = "3.8.1"
+__version__ = "3.8.0"
 
 __all__ = [
     "BinarySensorDeviceClass",

src/bthome_ble/parser.py

@@ -140,6 +140,8 @@ def __init__(self, bindkey: bytes | None = None) -> None:
         # advertisements is increasing, to have some replay protection. We always
         # start at zero allow the first message after a restart.
         self.encryption_counter = 0.0
+        self.reset_counter = 0  # Tracks the number of resets
+        self.message_since_last_reset = 0  # Tracks messages since the last reset
 
         # The packet_id is used to filter duplicate messages in BTHome V2.
         self.packet_id: float | None = None
@@ -644,14 +646,58 @@ def _decrypt_bthome(
             )
             raise ValueError
 
+        # decrypt the data. 
+        # if decryption fails then message is likely corrupted and no need to check the encryption counter.
+        try:
+            decrypted_payload = self.cipher.decrypt(
+                nonce, encrypted_payload + mic, associated_data
+            )
+        except InvalidTag as error:
+            if self.decryption_failed is True:
+                # we only ask for reautentification after the decryption has failed twice.
+                self.bindkey_verified = False
+            else:
+                self.decryption_failed = True
+            _LOGGER.warning("%s: Decryption failed: %s", self.title, error)
+            _LOGGER.debug("%s: mic: %s", self.title, mic.hex())
+            _LOGGER.debug("%s: nonce: %s", self.title, nonce.hex())
+            _LOGGER.debug(
+                "%s: encrypted_payload: %s", self.title, encrypted_payload.hex()
+            )
+            raise ValueError
+
         # filter advertisements with decreasing encryption counter.
+        self.message_since_last_reset += 1  # Increment messages since last reset
         if (
             new_encryption_counter < last_encryption_counter
             and self.bindkey_verified is True
         ):
-            if new_encryption_counter < 100 and last_encryption_counter >= 4294967195:
-                # the counter has (most likely) restarted from 0 after reaching the highest number.
+            _LOGGER.debug(
+                "%s: Encryption counter decrement detected. got counter: %i, had counter: %i, reset counter: %i, message counter: %i",                
+                self.title,
+                new_encryption_counter,
+                last_encryption_counter,
+                self.reset_counter,
+                self.message_since_last_reset
+            )
+            # Replay attack protection is two-edged sword: if you don't implement it, you allow attacker to replay a whole bunch of measurements.
+            # If you don't give some wiggle room for allowing resets changing batteries becomes a challenge
+            # Beware: If attacker manages to record a message with high encryption counter number they can 
+            # DoS all of your actual measurements until encryption counter reaches even higher value
+            if ((new_encryption_counter < 1000 and last_encryption_counter >= 4294967195) or 
+                (self.reset_counter == 0) or
+                (self.reset_counter <= 1 and self.message_since_last_reset >= 100)):
+                # Counter reset logic: either it's a legitimate overflow reset, or a reset is allowed per policy
                 self.encryption_counter = new_encryption_counter
+                self.message_since_last_reset = 0  # Reset the message counter since the last reset
+                self.reset_counter += 1  # Increment the reset counter
+                _LOGGER.warning(
+                    "%s: The new encryption counter (%i) is lower than the previous value (%i). "
+                    "Treating as a legitimate reset.",
+                    self.title,
+                    new_encryption_counter,
+                    last_encryption_counter,
+                )
             else:
                 # in all other cases, we assume the data has been comprimised and skip the
                 # advertisement
@@ -662,28 +708,19 @@ def _decrypt_bthome(
                     new_encryption_counter,
                     last_encryption_counter,
                 )
+                self.message_since_last_reset = 0  # Reset the message counter since the last reset
+                self.reset_counter += 1  # Increment the reset counter
                 raise ValueError
-        else:
-            self.encryption_counter = new_encryption_counter
+        # Reset the reset_counter if a hundred messages have been received since the last reset
+        if self.message_since_last_reset >= 100 and self.reset_counter >= 1:
+            if self.reset_counter <= 1:
+                # There has been only one reset. No big deal. Clear it out.
+                self.reset_counter = 0
+            if self.reset_counter > 1:
+                 # We're under seige, there has been multiple resets - carry this knowledge over into next hundred block.
+                self.reset_counter = 1
+            self.message_since_last_reset = 0
 
-        # decrypt the data
-        try:
-            decrypted_payload = self.cipher.decrypt(
-                nonce, encrypted_payload + mic, associated_data
-            )
-        except InvalidTag as error:
-            if self.decryption_failed is True:
-                # we only ask for reautentification after the decryption has failed twice.
-                self.bindkey_verified = False
-            else:
-                self.decryption_failed = True
-            _LOGGER.warning("%s: Decryption failed: %s", self.title, error)
-            _LOGGER.debug("%s: mic: %s", self.title, mic.hex())
-            _LOGGER.debug("%s: nonce: %s", self.title, nonce.hex())
-            _LOGGER.debug(
-                "%s: encrypted_payload: %s", self.title, encrypted_payload.hex()
-            )
-            raise ValueError
         if decrypted_payload is None:
             self.bindkey_verified = False
             _LOGGER.error(
@@ -694,5 +731,6 @@ def _decrypt_bthome(
             raise ValueError
         self.decryption_failed = False
         self.bindkey_verified = True
+        self.encryption_counter = new_encryption_counter
 
         return decrypted_payload