diff --git a/TheengsGateway/ble_gateway.py b/TheengsGateway/ble_gateway.py index 619afc99..e71d4ae0 100644 --- a/TheengsGateway/ble_gateway.py +++ b/TheengsGateway/ble_gateway.py @@ -715,6 +715,12 @@ def handle_encrypted_advertisement( decoded_json = decodeBLE(json.dumps(data_json)) if decoded_json: decoded_json = json.loads(decoded_json) # type: ignore[arg-type] + # Check if the mic matches the first byte of the bindkey for bindkey verification + elif mic != bindkey[:1].hex(): + logger.exception( + "Bindkey does not seem to be correct for `%s`", + get_address(decoded_json), + ) else: logger.exception( "Decrypted payload not supported: `%s`", diff --git a/TheengsGateway/decryption.py b/TheengsGateway/decryption.py index d3e85d84..89469cfb 100644 --- a/TheengsGateway/decryption.py +++ b/TheengsGateway/decryption.py @@ -135,10 +135,53 @@ def replace_encrypted_data( bthome_service_data.extend(decrypted_data) data_json["servicedata"] = bthome_service_data.hex() +class VictronDecryptor(AdvertisementDecryptor): + """Class for decryption of Victron Energy encrypted advertisements.""" + + def compute_nonce(self, address: str, decoded_json: dict) -> bytes: + """Get the nonce from a specific address and JSON input.""" + # The nonce is provided in the message and needs to be padded to 8 bytes + nonce = bytes.fromhex(decoded_json["ctr"]) + nonce = nonce.ljust(8, b"\x00") # Pad to 8 bytes with zeros + return nonce + + def decrypt( + self, + bindkey: bytes, + address: str, + decoded_json: dict, + ) -> bytes: + """Decrypt ciphertext from JSON input with AES CTR.""" + nonce = self.compute_nonce(address, decoded_json) + cipher = AES.new(bindkey, AES.MODE_CTR, nonce=nonce) + payload = bytes.fromhex(decoded_json["cipher"]) + decrypted_data = cipher.decrypt(payload) + return decrypted_data + + def replace_encrypted_data( + self, + decrypted_data: bytes, + data_json: dict, + decoded_json: dict, + ) -> None: + """Replace the encrypted data with decrypted payload.""" + # Extract the first 10 octets of the manufacturer data + victron_manufacturer_data = bytearray(bytes.fromhex(decoded_json["manufacturerdata"][:20])) + + # Replace indexes 4-5 and 14-17 with "11" and "ffff" to indicate decrypted data + victron_manufacturer_data[2:3] = binascii.unhexlify("11") + victron_manufacturer_data[7:9] = binascii.unhexlify("ffff") + + # Append the decrypted payload to the manufacturer data + victron_manufacturer_data.extend(decrypted_data) + + # Update the manufacturerdata field in the JSON + data_json["manufacturerdata"] = victron_manufacturer_data.hex() _DECRYPTORS = { 1: LYWSD03MMC_PVVXDecryptor, 2: BTHomeV2Decryptor, + 3: VictronDecryptor, }