Fix 16 verified bugs across Cloud, core, BulbDevice, scanner, Monitor

RELEASE.md

@@ -1,5 +1,20 @@
 # RELEASE NOTES
 
+## Unreleased - Bug Fixes
+
+* **Cloud**: Token-refresh retry in `_tuyaplatform()` now forwards `query` and `content_type`, so paginated/queried calls (device list, `getdevicelog`, `getdps`) no longer silently lose their query string when the access token expires mid-flight.
+* **Cloud**: `_gettoken()`, `_getuid()`, `_getdevice()`, `getdps()`, `sendcommand()`, and `getconnectstatus()` now return the standard error dict on a failed or empty Cloud response instead of raising `TypeError`/`KeyError`.
+* **Device**: `set_timer()` now selects the timer DP numerically instead of lexicographically (previously `"9"` sorted above `"102"`, targeting the wrong DP on devices with DP indices spanning a power of ten).
+* **XenonDevice**: Exhausted-retry socket timeouts now return `ERR_TIMEOUT` (902, "Timeout Waiting for Device") instead of the misleading `ERR_KEY_OR_VER` ("Check device key or version").
+* **XenonDevice**: Frame resync in `_receive()` now picks the earliest prefix when both 55AA and 6699 markers appear in buffered garbage; bare `except:` clauses in `_send_receive_quick()` no longer swallow `KeyboardInterrupt`; `received_wrong_cid_queue` is capped at 100 entries; hot-path `binascii.hexlify()` debug logging is now gated on the log level.
+* **BulbDevice**: `hexvalue_to_hsv()` reads Hue from the correct hex offset `[6:10]` in the rgb8 format (latent off-by-one); corrected the `set_music_colour()` docstring to match the actual argument order (`transition` first).
+* **Monitor**: A corrupt or desynced stream no longer stalls a device permanently — oversized/garbage headers now trigger a buffer resync to the next frame prefix instead of buffering forever.
+* **message_helper**: Truncated 55AA frames raise `DecodeError` (retryable) instead of leaking `struct.error`.
+* **udp_helper**: `decrypt_udp()` no longer raises `IndexError` on an empty or all-NUL 6699 broadcast payload.
+* **error_helper**: `error_json()` builds its dict directly (no JSON string round-trip) and returns "Unknown Error" for unrecognized codes instead of raising `KeyError`.
+* **scanner**: `devices()` no longer uses a shared mutable default for `tuyadevices` and no longer mutates the caller's device dicts when adding cloud-only entries to scan results.
+* Added offline regression tests for `set_timer` DP selection, `error_json` shape/unknown codes, and rgb8 hue round-trip (tests 23 → 29).
+
 ## Unreleased - Session Crypto Hardening
 
 * **Security: random AES-GCM nonces (v3.5)** — GCM message nonces were derived from the wall clock (`time.time()`, ~0.1 s granularity) and became a fixed constant whenever debug logging was enabled, causing nonce reuse under a single session key (which breaks GCM confidentiality and allows tag forgery). Nonces now come from `os.urandom(12)`. The IV is transmitted in the frame, so this is fully wire-compatible.

tests.py

@@ -76,6 +76,25 @@ def test_set_timer(self):
         self.assertEqual(result_cmd, expected_cmd)
         self.assertDictEqual(result_payload, expected_payload)
 
+    def test_set_timer_picks_numeric_max_dp(self):
+        # regression: DP keys must be selected numerically, not lexicographically
+        # ("9" > "102" as strings, but 102 > 9 as ints)
+        d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY)
+        d.set_version(3.1)
+        # status() supplies the DPS the timer selection scans
+        d.status = lambda nowait=False: {"devId":"DEVICE_ID","dps":{"1":False,"9":0,"102":0}}
+        d._send_receive = MagicMock(return_value={"devId":"DEVICE_ID","dps":{"1":False,"9":0,"102":0}})
+
+        # act
+        d.set_timer(6666)
+
+        # gather results
+        result_cmd, result_payload = get_results_from_mock(d)
+
+        # assert the timer targeted DP "102", not "9"
+        self.assertEqual(result_cmd, tinytuya.CONTROL)
+        self.assertDictEqual(result_payload, {"uid":"DEVICE_ID_HERE","devId":"DEVICE_ID_HERE","t":"","dps":{"102":6666}})
+
     def test_set_status(self):
         # arrange
         d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY)
@@ -423,6 +442,55 @@ def test_oversized_payload_rejected(self):
             mh.parse_header(oversized + b'\x00' * 40)
 
 
+class TestErrorJson(unittest.TestCase):
+    """error_json must return the expected dict shape and never raise, even
+    for an unknown error code."""
+
+    def test_known_code_shape(self):
+        from tinytuya.core.error_helper import error_json, ERR_TIMEOUT
+        result = error_json(ERR_TIMEOUT)
+        self.assertEqual(result["Error"], "Timeout Waiting for Device")
+        self.assertEqual(result["Err"], "902")
+        self.assertIsNone(result["Payload"])
+        self.assertEqual(set(result.keys()), {"Error", "Err", "Payload"})
+
+    def test_payload_preserved(self):
+        from tinytuya.core.error_helper import error_json, ERR_CLOUD
+        result = error_json(ERR_CLOUD, "some detail")
+        self.assertEqual(result["Payload"], "some detail")
+        self.assertEqual(result["Err"], "913")
+
+    def test_unknown_code_does_not_raise(self):
+        from tinytuya.core.error_helper import error_json
+        result = error_json(99999)
+        self.assertEqual(result["Error"], "Unknown Error")
+        self.assertEqual(result["Err"], "99999")
+
+    def test_default_none_code(self):
+        from tinytuya.core.error_helper import error_json
+        result = error_json()
+        self.assertEqual(result["Error"], "Unknown Error")
+        self.assertEqual(result["Err"], "None")
+        self.assertIsNone(result["Payload"])
+
+
+class TestHexvalueRoundTrip(unittest.TestCase):
+    """hexvalue_to_hsv must decode the hue written by rgb_to_hexvalue at the
+    correct offset (regression for the [7:10] vs [6:10] off-by-one)."""
+
+    def test_rgb8_round_trip(self):
+        import colorsys
+        for rgb in ((255, 128, 0), (0, 255, 64), (30, 60, 200)):
+            hexvalue = tinytuya.BulbDevice.rgb_to_hexvalue(*rgb, 'rgb8')
+            self.assertEqual(len(hexvalue), 14)
+            h, s, v = tinytuya.BulbDevice.hexvalue_to_hsv(hexvalue, 'rgb8')
+            eh, es, ev = colorsys.rgb_to_hsv(rgb[0] / 255.0, rgb[1] / 255.0, rgb[2] / 255.0)
+            # allow small quantization error from the 8-bit hex encoding
+            self.assertAlmostEqual(h, eh, delta=0.01)
+            self.assertAlmostEqual(s, es, delta=0.01)
+            self.assertAlmostEqual(v, ev, delta=0.01)
+
+
 class TestSessionCrypto(unittest.TestCase):
     """Session-key negotiation and GCM nonces must use fresh randomness so the
     same nonce/IV is never reused across messages or sessions."""
@@ -468,5 +536,6 @@ def test_receive_rejects_gcm_auth_failure(self):
         d.local_key = key
         with self.assertRaises(DecodeError):
             d._receive()
+
 if __name__ == '__main__':
     unittest.main()

tinytuya/BulbDevice.py

@@ -23,7 +23,7 @@
         set_timer(self, num_secs, nowait=False):
         set_musicmode(self, transition, modify_settings=True, nowait=False):
         unset_musicmode( self ):
-        set_music_colour( self, red, green, blue, brightness=None, colourtemp=None, transition=None, nowait=False ):
+        set_music_colour( self, transition, red, green, blue, brightness=None, colourtemp=None, nowait=False ):
         set_colour(r, g, b, nowait):
         set_hsv(h, s, v, nowait):
         set_white_percentage(brightness=100, colourtemp=0, nowait):
@@ -363,7 +363,7 @@ def hexvalue_to_hsv(hexvalue, hexformat=None):
                 h, s, v = colorsys.rgb_to_hsv(rgb[0] / 255.0, rgb[1] / 255.0, rgb[2] / 255.0)
             else:
                 # hexvalue is in rgb+hsv
-                h = int(hexvalue[7:10], 16) / 360.0
+                h = int(hexvalue[6:10], 16) / 360.0
                 s = int(hexvalue[10:12], 16) / 255.0
                 v = int(hexvalue[12:14], 16) / 255.0
         elif hexformat == 'hsv16':

tinytuya/Cloud.py

@@ -251,7 +251,7 @@ def _tuyaplatform(self, uri, action='GET', post=None, ver='v1.0', recursive=Fals
                 log.debug("Failed to renew token")
                 return None
             else:
-                return self._tuyaplatform(uri, action, post, ver, True)
+                return self._tuyaplatform(uri, action, post, ver, True, query, content_type)
 
         try:
             response_dict = json.loads(response.content.decode())
@@ -274,9 +274,10 @@ def _gettoken(self):
         response_dict = self._tuyaplatform('token?grant_type=1')
 
         if not response_dict or 'success' not in response_dict or not response_dict['success']:
+            msg = response_dict.get('msg') if response_dict else 'no response'
             self.error = error_json(
                 ERR_CLOUDTOKEN,
-                "Cloud _gettoken() failed: %r" % response_dict['msg'],
+                "Cloud _gettoken() failed: %r" % msg,
             )
             return self.error
 
@@ -301,17 +302,15 @@ def _getuid(self, deviceid=None):
         uri = 'devices/%s' % deviceid
         response_dict = self._tuyaplatform(uri)
 
-        if not response_dict['success']:
-            if 'code' not in response_dict:
-                response_dict['code'] = -1
-            if 'msg' not in response_dict:
-                response_dict['msg'] = 'Unknown Error'
+        if not response_dict or not response_dict.get('success'):
+            code = response_dict.get('code', -1) if response_dict else -1
+            msg = response_dict.get('msg', 'Unknown Error') if response_dict else 'no response'
             log.debug(
-                "Error from Tuya Cloud: %r", response_dict['msg'],
+                "Error from Tuya Cloud: %r", msg,
             )
             return error_json(
                 ERR_CLOUD,
-                "Error from Tuya Cloud: Code %r: %r" % (response_dict['code'], response_dict['msg'])
+                "Error from Tuya Cloud: Code %r: %r" % (code, msg)
             )
 
         uid = response_dict['result']['uid']
@@ -613,9 +612,14 @@ def _getdevice(self, param='status', deviceid=None):
         uri = 'iot-03/devices/%s/%s' % (deviceid, param)
         response_dict = self._tuyaplatform(uri)
 
-        if not response_dict['success']:
+        if not response_dict or not response_dict.get('success'):
+            msg = response_dict.get('msg', 'Unknown Error') if response_dict else 'no response'
             log.debug(
-                "Error from Tuya Cloud: %r", response_dict['msg'],
+                "Error from Tuya Cloud: %r", msg,
+            )
+            return error_json(
+                ERR_CLOUD,
+                "Error from Tuya Cloud: %r" % msg
             )
         return response_dict
 
@@ -651,9 +655,14 @@ def getdps(self, deviceid=None):
         uri = 'devices/%s/specifications' % (deviceid)
         response_dict = self._tuyaplatform(uri, ver='v1.1')
 
-        if not response_dict['success']:
+        if not response_dict or not response_dict.get('success'):
+            msg = response_dict.get('msg', 'Unknown Error') if response_dict else 'no response'
             log.debug(
-                "Error from Tuya Cloud: %r", response_dict['msg'],
+                "Error from Tuya Cloud: %r", msg,
+            )
+            return error_json(
+                ERR_CLOUD,
+                "Error from Tuya Cloud: %r" % msg
             )
         return response_dict
 
@@ -671,9 +680,14 @@ def sendcommand(self, deviceid=None, commands=None, uri='iot-03/devices/'):
         uri += '%s/commands' % (deviceid)
         response_dict = self._tuyaplatform(uri,action='POST',post=commands)
 
-        if not response_dict['success']:
+        if not response_dict or not response_dict.get('success'):
+            msg = response_dict.get('msg', 'Unknown Error') if response_dict else 'no response'
             log.debug(
-                "Error from Tuya Cloud: %r", response_dict['msg'],
+                "Error from Tuya Cloud: %r", msg,
+            )
+            return error_json(
+                ERR_CLOUD,
+                "Error from Tuya Cloud: %r" % msg
             )
         return response_dict
 
@@ -691,8 +705,13 @@ def getconnectstatus(self, deviceid=None):
         uri = 'devices/%s' % (deviceid)
         response_dict = self._tuyaplatform(uri, ver='v1.0')
 
-        if not response_dict['success']:
-            log.debug("Error from Tuya Cloud: %r", response_dict['msg'])
+        if not response_dict or not response_dict.get('success'):
+            msg = response_dict.get('msg', 'Unknown Error') if response_dict else 'no response'
+            log.debug("Error from Tuya Cloud: %r", msg)
+            return error_json(
+                ERR_CLOUD,
+                "Error from Tuya Cloud: %r" % msg
+            )
         return(response_dict["result"]["online"])
 
     def getdevicelog(self, deviceid=None, start=None, end=None, evtype=None, size=0, max_fetches=50, start_row_key=None, params=None):

tinytuya/core/Device.py

@@ -176,7 +176,10 @@ def set_timer(self, num_secs, dps_id=0, nowait=False):
             if "dps" in status:
                 devices = status["dps"]
                 devices_numbers = list(devices.keys())
-                devices_numbers.sort()
+                try:
+                    devices_numbers.sort(key=int)
+                except (TypeError, ValueError):
+                    devices_numbers.sort()
                 dps_id = devices_numbers[-1]
             else:
                 log.debug("set_timer received error=%r", status)

tinytuya/core/Monitor.py

@@ -66,6 +66,7 @@ def on_status(device, result):
 import time
 
 from . import header as H
+from .const import MAX_PAYLOAD_LENGTH
 from .message_helper import (
     parse_header,
     unpack_message,
@@ -475,6 +476,22 @@ def _process_buffer(self, state):
             try:
                 header = parse_header(buf)
             except Exception:
+                # Normally this means the header is incomplete, so wait for more
+                # data.  But if the buffer is already larger than any legal frame,
+                # "not enough data" cannot be the real cause — the stream is
+                # desynced/corrupt.  Resync by discarding up to the next prefix.
+                if len(buf) > MAX_PAYLOAD_LENGTH:
+                    next_55aa = buf.find(H.PREFIX_55AA_BIN, 1)
+                    next_6699 = buf.find(H.PREFIX_6699_BIN, 1)
+                    candidates = [o for o in (next_55aa, next_6699) if o >= 1]
+                    if candidates:
+                        next_offset = min(candidates)
+                        log.debug("Corrupt/oversized header, resyncing buffer to next prefix at offset %d", next_offset)
+                        state.recv_buffer = buf[next_offset:]
+                        continue
+                    log.debug("Corrupt/oversized header and no further prefix found, dropping buffer")
+                    state.recv_buffer = buf[-3:]
+                    return
                 # Incomplete header — wait for more data
                 return
 

tinytuya/core/XenonDevice.py

@@ -14,7 +14,7 @@
 
 from .const import DEVICEFILE, TCPPORT
 from .crypto_helper import AESCipher
-from .error_helper import ERR_CONNECT, ERR_DEVTYPE, ERR_JSON, ERR_KEY_OR_VER, ERR_OFFLINE, ERR_PAYLOAD, error_json
+from .error_helper import ERR_CONNECT, ERR_DEVTYPE, ERR_JSON, ERR_KEY_OR_VER, ERR_OFFLINE, ERR_PAYLOAD, ERR_TIMEOUT, error_json
 from .exceptions import DecodeError
 from .message_helper import MessagePayload, TuyaMessage, pack_message, unpack_message, parse_header
 from . import command_types as CT, header as H
@@ -479,7 +479,13 @@ def _receive(self):
             if prefix_offset_55AA < 0 and prefix_offset_6699 < 0:
                 data = data[1-prefix_len:]
             else:
-                prefix_offset = prefix_offset_6699 if prefix_offset_55AA < 0 else prefix_offset_55AA
+                # pick the earliest (minimum non-negative) prefix offset
+                if prefix_offset_55AA < 0:
+                    prefix_offset = prefix_offset_6699
+                elif prefix_offset_6699 < 0:
+                    prefix_offset = prefix_offset_55AA
+                else:
+                    prefix_offset = min(prefix_offset_55AA, prefix_offset_6699)
                 data = data[prefix_offset:]
 
             data += self._recv_all( min_len - len(data) )
@@ -491,7 +497,8 @@ def _receive(self):
         if remaining > 0:
             data += self._recv_all( remaining )
 
-        log.debug("received data=%r", binascii.hexlify(data))
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("received data=%r", binascii.hexlify(data))
         hmac_key = self.local_key if self.version >= 3.4 else None
         no_retcode = False #None if self.version >= 3.5 else False
         msg = unpack_message(data, header=header, hmac_key=hmac_key, no_retcode=no_retcode)
@@ -516,20 +523,20 @@ def _send_receive_quick(self, payload, recv_retries, from_child=None): # pylint:
         enc_payload = self._encode_message(payload) if type(payload) == MessagePayload else payload
         try:
             self.socket.sendall(enc_payload)
-        except:
+        except Exception:
             self._check_socket_close(True)
             return None
         try:
             self.raw_sent = parse_header(enc_payload)
-        except:
+        except Exception:
             self.raw_sent = None
         if not recv_retries:
             return True
         while recv_retries:
             try:
                 msg = self._receive()
                 self.raw_recv.append(msg)
-            except:
+            except Exception:
                 msg = None
             if msg:
                 self._get_retcode(self.raw_sent, msg) # set self.cmd_retcode
@@ -645,7 +652,7 @@ def _send_receive(self, payload, minresponse=28, getresponse=True, decode_respon
                         self.socketRetryLimit
                     )
                     # timeout reached - return error
-                    return error_json(ERR_KEY_OR_VER)
+                    return error_json(ERR_TIMEOUT)
                 # wait a bit before retrying
                 time.sleep(0.1)
             except DecodeError as err:
@@ -751,6 +758,10 @@ def _process_message( self, msg, dev_type=None, from_child=None, minresponse=28,
                         self._cache_response(result)
                         result = self._process_response(result)
                     self.received_wrong_cid_queue.append( (found_child, result) )
+                    # cap the queue so a stream of wrong-CID updates cannot grow it unbounded
+                    if len(self.received_wrong_cid_queue) > 100:
+                        log.debug( 'received_wrong_cid_queue exceeded 100 entries, dropping oldest' )
+                        self.received_wrong_cid_queue.pop(0)
                 # events should not be coming in so fast that we will never timeout a read, so don't worry about loops
                 return self._send_receive( None, minresponse, True, decode_response, from_child=from_child)
 
@@ -980,7 +991,8 @@ def _encode_message( self, msg ):
                 msg = TuyaMessage(self.seqno, msg.cmd, None, payload, 0, True, H.PREFIX_6699_VALUE, True)
                 self.seqno += 1  # increase message sequence number
                 data = pack_message(msg,hmac_key=self.local_key)
-                log.debug("payload [%d] encrypted=%r",self.seqno, binascii.hexlify(data) )
+                if log.isEnabledFor(logging.DEBUG):
+                    log.debug("payload [%d] encrypted=%r",self.seqno, binascii.hexlify(data) )
                 return data
 
             payload = self.cipher.encrypt(payload, False)
@@ -1015,7 +1027,8 @@ def _encode_message( self, msg ):
         msg = TuyaMessage(self.seqno, msg.cmd, 0, payload, 0, True, H.PREFIX_55AA_VALUE, False)
         self.seqno += 1  # increase message sequence number
         buffer = pack_message(msg,hmac_key=hmac_key)
-        log.debug("payload encrypted=%r",binascii.hexlify(buffer))
+        if log.isEnabledFor(logging.DEBUG):
+            log.debug("payload encrypted=%r",binascii.hexlify(buffer))
         return buffer
 
     def _get_retcode(self, sent, msg):