diff --git a/RELEASE.md b/RELEASE.md index 7fa12cd5..b4ad8efd 100644 --- a/RELEASE.md +++ b/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. diff --git a/tests.py b/tests.py index 28693eab..974c8af0 100755 --- a/tests.py +++ b/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() diff --git a/tinytuya/core/Monitor.py b/tinytuya/core/Monitor.py index fc46f317..c0f6ac25 100644 --- a/tinytuya/core/Monitor.py +++ b/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) diff --git a/tinytuya/core/XenonDevice.py b/tinytuya/core/XenonDevice.py index ad2748ed..4f06e934 100644 --- a/tinytuya/core/XenonDevice.py +++ b/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 diff --git a/tinytuya/core/crypto_helper.py b/tinytuya/core/crypto_helper.py index 43697343..c9457ff7 100644 --- a/tinytuya/core/crypto_helper.py +++ b/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