fix: duplicate _refresh_channel_secret silently disabled exact echo matching

The class defined _refresh_channel_secret twice. The second definition
(legacy, returns None) shadowed the first one added in 77c3ffa, so
send_channel_message never got the channel secret, expected_payloads was
always empty, and every sent-message echo correlation fell through to the
loose 60s channel-hash-byte fallback. Any foreign GRP_TXT echo on the same
channel inside that window could then be mis-assigned to our sent message
(wrong hash + physically impossible path on the badge and in Path Analyzer).

Drop the legacy definition and let set_channel use the surviving one, which
also re-reads the secret and updates cache + DB.

Co-Authored-By: Claude Fable 5 <noreply@anthropic.com>
This commit is contained in:
MarekWo
2026-07-22 07:05:52 +02:00
parent c5a0fa8fd1
commit bb67c4d3ea
+1 -20
View File
@@ -2674,32 +2674,13 @@ class DeviceManager:
# Read back the actual secret from device (firmware may have
# generated it for # channels) and update in-memory cache + DB.
self._refresh_channel_secret(idx, name)
self._refresh_channel_secret(idx)
return {'success': True, 'message': f'Channel {idx} set'}
except Exception as e:
logger.error(f"Failed to set channel: {e}")
return {'success': False, 'error': str(e)}
def _refresh_channel_secret(self, idx: int, name: str = ''):
"""Read back a channel's secret from device and update cache + DB."""
try:
event = self.execute(self.mc.commands.get_channel(idx))
if event:
data = getattr(event, 'payload', None) or {}
secret = data.get('channel_secret', data.get('secret', b''))
if isinstance(secret, bytes):
secret = secret.hex()
if secret and len(secret) == 32:
self._channel_secrets[idx] = secret
ch_name = data.get('channel_name', data.get('name', ''))
if isinstance(ch_name, str):
ch_name = ch_name.strip('\x00').strip()
self.db.upsert_channel(idx, ch_name or name, secret)
logger.info(f"Refreshed channel {idx} secret into cache")
except Exception as e:
logger.warning(f"Failed to refresh channel {idx} secret: {e}")
def remove_channel(self, idx: int) -> Dict:
"""Remove a channel from the device."""
if not self.is_connected: