Merge branch 'main' into feature/type_8
This commit is contained in:
@@ -105,6 +105,7 @@ class PowerViewBLE:
|
|||||||
self._info: PVDeviceInfo = PVDeviceInfo()
|
self._info: PVDeviceInfo = PVDeviceInfo()
|
||||||
self._cmd_lock: Final = asyncio.Lock()
|
self._cmd_lock: Final = asyncio.Lock()
|
||||||
self._cmd_next: tuple[ShadeCmd, bytes]
|
self._cmd_next: tuple[ShadeCmd, bytes]
|
||||||
|
self._is_encrypted: bool = False
|
||||||
self._cipher: Final[Cipher | None] = (
|
self._cipher: Final[Cipher | None] = (
|
||||||
Cipher(algorithms.AES(home_key), modes.CTR(bytearray(16)))
|
Cipher(algorithms.AES(home_key), modes.CTR(bytearray(16)))
|
||||||
if len(home_key) == 16
|
if len(home_key) == 16
|
||||||
@@ -115,6 +116,15 @@ class PowerViewBLE:
|
|||||||
await self._data_event.wait()
|
await self._data_event.wait()
|
||||||
self._data_event.clear()
|
self._data_event.clear()
|
||||||
|
|
||||||
|
@property
|
||||||
|
def encrypted(self) -> bool:
|
||||||
|
"""Return whether communication with this shade is encrypted."""
|
||||||
|
return self._is_encrypted
|
||||||
|
|
||||||
|
@encrypted.setter
|
||||||
|
def encrypted(self, value:bool) -> None:
|
||||||
|
self._is_encrypted = value
|
||||||
|
|
||||||
@property
|
@property
|
||||||
def info(self) -> PVDeviceInfo:
|
def info(self) -> PVDeviceInfo:
|
||||||
"""Return device information, e.g. SW version."""
|
"""Return device information, e.g. SW version."""
|
||||||
@@ -143,7 +153,7 @@ class PowerViewBLE:
|
|||||||
+ bytes([self._seqcnt, len(cmd_run[1])])
|
+ bytes([self._seqcnt, len(cmd_run[1])])
|
||||||
+ cmd_run[1]
|
+ cmd_run[1]
|
||||||
)
|
)
|
||||||
if self._cipher is not None:
|
if self._cipher is not None and self._is_encrypted:
|
||||||
enc: AEADEncryptionContext = self._cipher.encryptor()
|
enc: AEADEncryptionContext = self._cipher.encryptor()
|
||||||
tx_data = enc.update(tx_data) + enc.finalize()
|
tx_data = enc.update(tx_data) + enc.finalize()
|
||||||
self._data_event.clear()
|
self._data_event.clear()
|
||||||
@@ -232,7 +242,7 @@ class PowerViewBLE:
|
|||||||
def _verify_response(self, din: bytearray, seq_nr: int, cmd: ShadeCmd) -> bool:
|
def _verify_response(self, din: bytearray, seq_nr: int, cmd: ShadeCmd) -> bool:
|
||||||
"""Verify shade response data."""
|
"""Verify shade response data."""
|
||||||
data: bytearray = din
|
data: bytearray = din
|
||||||
if self._cipher is not None:
|
if self._cipher is not None and self._is_encrypted:
|
||||||
dec: AEADDecryptionContext = self._cipher.decryptor()
|
dec: AEADDecryptionContext = self._cipher.decryptor()
|
||||||
data = bytearray(dec.update(din) + dec.finalize())
|
data = bytearray(dec.update(din) + dec.finalize())
|
||||||
if len(data) < 4:
|
if len(data) < 4:
|
||||||
|
|||||||
@@ -103,6 +103,7 @@ class PVCoordinator(PassiveBluetoothDataUpdateCoordinator):
|
|||||||
bytearray(service_info.manufacturer_data.get(2073, b""))
|
bytearray(service_info.manufacturer_data.get(2073, b""))
|
||||||
)
|
)
|
||||||
)
|
)
|
||||||
|
self.api.encrypted = bool(self.data.get("home_id"))
|
||||||
|
|
||||||
LOGGER.debug("data sample %s", self.data)
|
LOGGER.debug("data sample %s", self.data)
|
||||||
super()._async_handle_bluetooth_event(service_info, change)
|
super()._async_handle_bluetooth_event(service_info, change)
|
||||||
|
|||||||
@@ -16,5 +16,5 @@
|
|||||||
"issue_tracker": "https://github.com/patman15/hdpv_ble/issues",
|
"issue_tracker": "https://github.com/patman15/hdpv_ble/issues",
|
||||||
"loggers": ["hunterdouglas_powerview_ble"],
|
"loggers": ["hunterdouglas_powerview_ble"],
|
||||||
"requirements": ["cryptography>=43.0.0"],
|
"requirements": ["cryptography>=43.0.0"],
|
||||||
"version": "0.21"
|
"version": "0.22"
|
||||||
}
|
}
|
||||||
|
|||||||
Reference in New Issue
Block a user