Sign requests for firmware that requires it

README.md

@@ -158,6 +158,8 @@ async def main():
     light = wizlight("192.168.1.27")
     # Set up the light with a custom port
     #light = wizlight("your bulb's IP address", port=12345)
+    # Newer firmware needs the signing key from the WiZ app (see below)
+    #light = wizlight("192.168.1.27", signing_key="<udp_signing_key>")
 
     # The following calls need to be done inside an asyncio coroutine
     # to run them from normal synchronous code, you can wrap them with
@@ -292,6 +294,18 @@ The discovery works with a UDP Broadcast request and collects all bulbs in the n
 - **state** - whether it's on or off
 - **schdPsetId** - rhythm id of the room
 
+## Signed requests (newer firmware)
+
+Some newer firmware answers every request with `"env":"pro"` and a `hmac` field. On those devices reads still work, but `setPilot` without a signature fails with `-32602 Invalid params`, and a bad signature fails with `-32605 Security error`.
+
+The key comes from the WiZ app, which can export your home for local integration as a JSON file. Pass its `udp_signing_key` value (32 hex characters, one per home) as `signing_key`:
+
+```python
+light = wizlight("192.168.1.27", signing_key="<udp_signing_key>")
+```
+
+Each request then gets `sigTs` (unix time) added to `params`, plus `"env":"pro"` and `hmac`, which is base64 HMAC-SHA256 of the compact `params` JSON keyed with the hex-decoded key. The bulb rejects a `sigTs` about a minute old, so the host clock needs to be right.
+
 ## Async I/O
 
 For async I/O this component uses Python's built-in asyncio DatagramTransport, which allows completely non-blocking UDP transport.

pywizlight/bulb.py

@@ -19,7 +19,12 @@
 from pywizlight.push_manager import PushManager
 from pywizlight.rgbcw import hs2rgbcw, rgb2rgbcw
 from pywizlight.scenes import SCENES, SCENES_BY_CLASS
-from pywizlight.utils import hex_to_percent, percent_to_hex, to_wiz_json
+from pywizlight.utils import (
+    hex_to_percent,
+    percent_to_hex,
+    sign_request,
+    to_wiz_json,
+)
 from pywizlight.vec import Vector
 
 _LOGGER = logging.getLogger(__name__)
@@ -556,10 +561,17 @@ def __init__(
         ip: str,
         port: int = PORT,
         mac: Optional[str] = None,
+        signing_key: Optional[str] = None,
     ) -> None:
-        """Create instance with the IP address of the bulb."""
+        """Create instance with the IP address of the bulb.
+
+        signing_key is the udp_signing_key from the WiZ app's local
+        integration export. Newer firmware rejects unsigned setPilot
+        requests without it.
+        """
         self.ip = ip
         self.port = port
+        self.signing_key = bytes.fromhex(signing_key) if signing_key else None
         self.state: List[Optional[PilotParser]] = []
         self.mac = mac
         self.bulbtype: Optional[BulbType] = None
@@ -1060,6 +1072,8 @@ async def fanSwitch(self) -> None:
     async def send(self, msg_dict: Dict) -> BulbResponse:
         """Serialize a dict to json and send it to device over UDP."""
         self.history.message(HISTORY_SEND, msg_dict)
+        if self.signing_key:
+            msg_dict = sign_request(msg_dict, self.signing_key)
         message = to_wiz_json(msg_dict)
         await self._ensure_connection()
         assert self.transport is not None

pywizlight/tests/test_signing.py

@@ -0,0 +1,133 @@
+"""Tests for signed requests on "env":"pro" firmware."""
+
+import asyncio
+import base64
+import hashlib
+import hmac
+import json
+import time
+from typing import Any, AsyncGenerator, Dict, Tuple
+
+import pytest
+
+from pywizlight import PilotBuilder, wizlight
+from pywizlight.exceptions import WizLightConnectionError
+from pywizlight.utils import sign_request
+
+KEY_HEX = "00112233445566778899aabbccddeeff"
+KEY = bytes.fromhex(KEY_HEX)
+
+
+def _mac(obj: Dict[str, Any]) -> str:
+    payload = json.dumps(obj, separators=(",", ":")).encode()
+    return base64.b64encode(hmac.new(KEY, payload, hashlib.sha256).digest()).decode()
+
+
+def test_sign_request_known_answer() -> None:
+    """The signature covers compact params including sigTs."""
+    signed = sign_request(
+        {"method": "setPilot", "params": {"state": True}}, KEY, timestamp=1790441723
+    )
+    assert signed == {
+        "method": "setPilot",
+        "env": "pro",
+        "params": {"state": True, "sigTs": 1790441723},
+        "hmac": "vARMv4JpVsxYabcBovABd25ISBzYT8+71v0CUtpU1sg=",
+    }
+
+
+def test_sign_request_leaves_input_alone() -> None:
+    msg = {"method": "setPilot", "params": {"dimming": 50}}
+    sign_request(msg, KEY)
+    assert msg == {"method": "setPilot", "params": {"dimming": 50}}
+
+
+class SignedBulb(asyncio.DatagramProtocol):
+    """Answers like fw 1.39 on ESP25_SHORGB_01: reads open, writes signed."""
+
+    def __init__(self) -> None:
+        self.pilot: Dict[str, Any] = {"mac": "a8bb5006033d", "state": False}
+        self.transport: asyncio.DatagramTransport
+
+    def connection_made(self, transport: asyncio.BaseTransport) -> None:
+        self.transport = transport  # type: ignore[assignment]
+
+    def _reply(self, method: str, key: str, body: Dict[str, Any], addr: Any) -> None:
+        body = {**body, "sigTs": int(time.time())}
+        resp = {"method": method, "env": "pro", key: body, "hmac": _mac(body)}
+        self.transport.sendto(json.dumps(resp, separators=(",", ":")).encode(), addr)
+
+    def datagram_received(self, data: bytes, addr: Tuple[str, int]) -> None:
+        req = json.loads(data)
+        method, params = req["method"], req.get("params", {})
+        if method == "getPilot":
+            self._reply(method, "result", self.pilot, addr)
+            return
+        if method == "getSystemConfig":
+            config = {"moduleName": "ESP25_SHORGB_01", "fwVersion": "1.39.0"}
+            self._reply(method, "result", config, addr)
+            return
+        if method == "getModelConfig":
+            model = {
+                "devTotal": 1,
+                "headTotal": 1,
+                "cctRange": [2700, 2700, 5000, 5000],
+            }
+            self._reply(method, "result", model, addr)
+            return
+        if method == "getUserConfig":
+            self._reply(method, "result", {"fadeIn": 500, "fadeOut": 2000}, addr)
+            return
+        if method.startswith("get"):
+            not_found = {"code": -32601, "message": "Method not found"}
+            self._reply(method, "error", not_found, addr)
+            return
+        if "sigTs" not in params:
+            self._reply(
+                method, "error", {"code": -32602, "message": "Invalid params"}, addr
+            )
+            return
+        if req.get("hmac") != _mac(params) or time.time() - params["sigTs"] > 30:
+            self._reply(
+                method, "error", {"code": -32605, "message": "Security error"}, addr
+            )
+            return
+        self.pilot.update({k: v for k, v in params.items() if k != "sigTs"})
+        self._reply(method, "result", {"success": True}, addr)
+
+
+@pytest.fixture()
+async def signed_bulb_port() -> AsyncGenerator[int, None]:
+    loop = asyncio.get_running_loop()
+    transport, _ = await loop.create_datagram_endpoint(
+        SignedBulb, local_addr=("127.0.0.1", 0)
+    )
+    yield transport.get_extra_info("sockname")[1]
+    transport.close()
+
+
+@pytest.mark.asyncio
+async def test_unsigned_write_is_rejected(signed_bulb_port: int) -> None:
+    bulb = wizlight(ip="127.0.0.1", port=signed_bulb_port)
+    with pytest.raises(WizLightConnectionError, match="-32602"):
+        await bulb.turn_on(PilotBuilder(brightness=128))
+    await bulb.async_close()
+
+
+@pytest.mark.asyncio
+async def test_wrong_key_is_rejected(signed_bulb_port: int) -> None:
+    bulb = wizlight(ip="127.0.0.1", port=signed_bulb_port, signing_key="ff" * 16)
+    with pytest.raises(WizLightConnectionError, match="-32605"):
+        await bulb.turn_on(PilotBuilder(brightness=128))
+    await bulb.async_close()
+
+
+@pytest.mark.asyncio
+async def test_signed_write_and_read(signed_bulb_port: int) -> None:
+    bulb = wizlight(ip="127.0.0.1", port=signed_bulb_port, signing_key=KEY_HEX)
+    await bulb.turn_on(PilotBuilder(brightness=128, colortemp=2700))
+    states = await bulb.updateState()
+    assert states and states[0]
+    assert states[0].get_state() is True
+    assert states[0].get_colortemp() == 2700
+    await bulb.async_close()

pywizlight/utils.py

@@ -1,8 +1,12 @@
+import base64
+import hashlib
+import hmac
 import json
 import logging
 import random
 import socket
-from typing import Optional, cast
+import time
+from typing import Any, Dict, Optional, cast
 
 _LOGGER = logging.getLogger(__name__)
 
@@ -22,6 +26,28 @@ def to_wiz_json(dump_obj):
     return json.dumps(dump_obj, separators=(",", ":"))
 
 
+def sign_request(
+    msg: Dict[str, Any], signing_key: bytes, timestamp: Optional[int] = None
+) -> Dict[str, Any]:
+    """Sign a request for firmware that answers with "env":"pro".
+
+    The HMAC-SHA256 covers the compact JSON of params, with sigTs (unix
+    seconds) added to params. The key is the hex-decoded udp_signing_key
+    from the WiZ app's local integration export.
+    """
+    params = dict(msg.get("params", {}))
+    params["sigTs"] = int(time.time()) if timestamp is None else timestamp
+    digest = hmac.new(
+        signing_key, to_wiz_json(params).encode("utf-8"), hashlib.sha256
+    ).digest()
+    return {
+        "method": msg["method"],
+        "env": "pro",
+        "params": params,
+        "hmac": base64.b64encode(digest).decode("ascii"),
+    }
+
+
 def create_udp_broadcast_socket(listen_port: int) -> socket.socket:
     """Create a udp broadcast socket used for communicating with the device."""
     return _create_udp_socket(listen_port, reuseaddr=True, broadcast=True)