Security: random nonces/IVs and GCM auth enforcement for v3.4/v3.5 sessions

RELEASE.md

@@ -1,5 +1,11 @@
 # RELEASE NOTES
 
+## 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.
+* **Security: random session-key client nonce (v3.4/v3.5)** — the client nonce used in session-key negotiation was hardcoded to `0123456789abcdef`, making the session key a deterministic function of the device nonce alone (and reusing the same GCM IV every 3.5 session). It now uses `os.urandom(16)` per negotiation. Devices accept any client nonce, so this is wire-compatible.
+* **Security: enforce GCM authentication on receive** — 6699/GCM frames that fail their authentication tag were previously passed to the decoder as raw ciphertext. They are now rejected (`DecodeError` in `XenonDevice._receive`, dropped in `Monitor`), so a forged or corrupt frame can no longer feed unverified bytes into payload processing.
+
 ## v1.19.0 - Monitor Class, IPv6, and Community Fixes
 
 * **New Feature: `Monitor` class** — Single-thread, multi-device status monitoring using `selectors` (`select`/`poll`/`epoll`). Watch any number of Tuya devices on one OS thread with callback-driven updates (`on_status`, `on_connect`, `on_disconnect`), automatic heartbeats, gateway/cid routing, thread-safe command queue, and optional `auto_reconnect`. No `asyncio`, no per-device threads, no new dependencies. See `examples/monitor_example.py` and `examples/monitor_poll_example.py`. Implements the [proposal by @3735943886](https://github.com/jasonacox/tinytuya/pull/649#issuecomment-4628381086) via [#712](https://github.com/jasonacox/tinytuya/pull/712) by @jasonacox-sam. **Note:** `Monitor` is an experimental class. See [#713](https://github.com/jasonacox/tinytuya/issues/713) for feedback and future refactoring plans.

tests.py

@@ -423,5 +423,50 @@ def test_oversized_payload_rejected(self):
             mh.parse_header(oversized + b'\x00' * 40)
 
 
+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."""
+
+    def test_client_nonce_is_random_and_16_bytes(self):
+        d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY, version=3.5)
+        d._negotiate_session_key_generate_step_1()
+        first = d.local_nonce
+        d._negotiate_session_key_generate_step_1()
+        second = d.local_nonce
+        self.assertEqual(len(first), 16)
+        self.assertEqual(len(second), 16)
+        self.assertNotEqual(first, second, "client nonce must differ between negotiations")
+
+    def test_gcm_iv_is_random_and_12_bytes(self):
+        cipher = mh.AESCipher(LOCAL_KEY.encode('latin1'))
+        if not cipher.CRYPTOLIB_HAS_GCM:
+            self.skipTest("crypto backend has no GCM support")
+        if log.isEnabledFor(logging.DEBUG):
+            self.skipTest("debug mode uses a fixed IV for packet troubleshooting")
+        iv1 = cipher.get_encryption_iv(True)
+        iv2 = cipher.get_encryption_iv(True)
+        self.assertEqual(len(iv1), 12)
+        self.assertEqual(len(iv2), 12)
+        self.assertNotEqual(iv1, iv2, "GCM IV must be random per message")
+
+    def test_receive_rejects_gcm_auth_failure(self):
+        key = LOCAL_KEY.encode('latin1')
+        cipher = mh.AESCipher(key)
+        if not cipher.CRYPTOLIB_HAS_GCM:
+            self.skipTest("crypto backend has no GCM support")
+
+        # Build a valid 6699/GCM frame, then flip 1 bit in the tag.
+        payload = b'{"x":1}'
+        good = mh.pack_message(
+            mh.TuyaMessage(1, 16, None, payload, 0, True, mh.H.PREFIX_6699_VALUE, b'\x00' * 12),
+            hmac_key=key,
+        )
+        bad = good[:-5] + bytes([good[-5] ^ 1]) + good[-4:]
+
+        d = tinytuya.OutletDevice('DEVICE_ID_HERE', 'IP_ADDRESS_HERE', LOCAL_KEY, version=3.5)
+        d._recv_all = lambda n: bad
+        d.local_key = key
+        with self.assertRaises(DecodeError):
+            d._receive()
 if __name__ == '__main__':
     unittest.main()

tinytuya/core/Monitor.py

@@ -506,6 +506,12 @@ def _decode_and_dispatch(self, state, frame_data, header):
         if msg is None:
             return
 
+        # For 6699/GCM frames a failed authentication tag means the payload is
+        # undecryptable ciphertext; drop the frame instead of decoding garbage.
+        if msg.prefix == H.PREFIX_6699_VALUE and not msg.crc_good:
+            log.debug('Monitor: GCM authentication failed for %s - frame dropped', device.id)
+            return
+
         # Null payload — heartbeat ack, etc.
         if not msg.payload or len(msg.payload) == 0:
             log.debug('Monitor: null payload from %s (cmd=%s)', device.id, msg.cmd)

tinytuya/core/XenonDevice.py

@@ -6,6 +6,7 @@
 import json
 from hashlib import md5, sha256
 import logging
+import os
 import socket
 import struct
 import time
@@ -493,7 +494,13 @@ def _receive(self):
         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
-        return unpack_message(data, header=header, hmac_key=hmac_key, no_retcode=no_retcode)
+        msg = unpack_message(data, header=header, hmac_key=hmac_key, no_retcode=no_retcode)
+        # For 6699/GCM frames a failed authentication tag means the payload is
+        # undecryptable ciphertext; reject the frame instead of feeding garbage
+        # to the decoder.  DecodeError is handled by the caller's retry logic.
+        if msg.prefix == H.PREFIX_6699_VALUE and not msg.crc_good:
+            raise DecodeError('GCM authentication failed - frame rejected')
+        return msg
 
     # similar to _send_receive() but never retries sending and does not decode the response
     def _send_receive_quick(self, payload, recv_retries, from_child=None): # pylint: disable=W0613
@@ -882,7 +889,11 @@ def _negotiate_session_key(self):
         return True
 
     def _negotiate_session_key_generate_step_1( self ):
-        self.local_nonce = b'0123456789abcdef' # not-so-random random key
+        # Fresh random client nonce per negotiation.  A hardcoded nonce means the
+        # session key is a deterministic function of the device nonce alone (and,
+        # for 3.5, reuses the same GCM IV every session); the device accepts any
+        # nonce, so this is wire-compatible.
+        self.local_nonce = os.urandom(16)
         self.remote_nonce = b''
         self.local_key = self.real_local_key
 

tinytuya/core/crypto_helper.py

@@ -4,7 +4,7 @@
 from __future__ import print_function  # python 2.7 support
 import base64
 import logging
-import time
+import os
 
 for clib in ('pyca/cryptography', 'PyCryptodomex', 'PyCrypto', 'pyaes'):
     Crypto = Crypto_modes = AES = CRYPTOLIB = None
@@ -52,9 +52,15 @@ def get_encryption_iv( cls, iv ):
             raise NotImplementedError( 'Crypto library does not support GCM' )
         if iv is True:
             if log.isEnabledFor( logging.DEBUG ):
+                # Debug mode: fixed IV for troubleshooting and packet analysis
                 iv = b'0123456789ab'
             else:
-                iv = str(time.time() * 10)[:12].encode('utf8')
+                # GCM nonce: must be unique per (key, message).  A time-derived
+                # or constant nonce reuses the value across messages under the
+                # same session key, which breaks GCM confidentiality and allows
+                # tag forgery.  The IV is transmitted in the frame, so a random
+                # value is fully wire-compatible.
+                iv = os.urandom(12)
         return iv
 
     @classmethod