From 8ce61b7d3294d4d734b7bf1de48d91cf44ea10d0 Mon Sep 17 00:00:00 2001 From: Aruna Tennakoon Date: Mon, 31 Aug 2026 18:20:45 +0700 Subject: [PATCH 1/2] feat: local control --- CHANGELOG.md | 49 ++++ README.md | 27 ++ pyproject.toml | 5 +- sinricpro/__init__.py | 2 +- sinricpro/core/local_control/__init__.py | 6 + sinricpro/core/local_control/mdns.py | 159 ++++++++++++ sinricpro/core/local_control/udp_listener.py | 178 +++++++++++++ sinricpro/core/message_queue.py | 110 ++++++-- sinricpro/core/signature.py | 118 ++++++--- sinricpro/core/sinric_pro.py | 242 ++++++++++++----- sinricpro/core/types.py | 59 ++++- sinricpro/core/websocket_client.py | 19 +- sinricpro/utils/__init__.py | 11 +- sinricpro/utils/network.py | 48 ++++ tests/unit/test_local_control.py | 259 +++++++++++++++++++ tests/unit/test_signature.py | 139 ++++++++++ 16 files changed, 1288 insertions(+), 143 deletions(-) create mode 100644 sinricpro/core/local_control/__init__.py create mode 100644 sinricpro/core/local_control/mdns.py create mode 100644 sinricpro/core/local_control/udp_listener.py create mode 100644 sinricpro/utils/network.py create mode 100644 tests/unit/test_local_control.py create mode 100644 tests/unit/test_signature.py diff --git a/CHANGELOG.md b/CHANGELOG.md index cc29c1e..f8f722a 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -1,3 +1,52 @@ +## [6.0.0] +- feat: Local control. Devices answer signed commands over the LAN on UDP 3333 (multicast 224.9.9.9, unicast too), so they keep working with the cloud unreachable. Requests dispatch through the existing capability callbacks. +- feat: mDNS announcement of `_sinricpro._udp.local.` (TXT: `deviceIds`, `sdk`, `udp`) for app discovery. `zeroconf` is a required dependency, so a default install announces without extra steps. Local control is on by default; set `local_control=False` to opt out. +- feat: `SinricProConfig.local_control`, `.mdns` and `.local_control_interface` to configure the feature out or pin it to a LAN interface on multi-homed hosts. +- fix: Signatures are now verified against the bytes received, sliced between `"payload":` and `,"signature"`, instead of re-serializing the parsed message. A sender using different key order or spacing was rejected before. +- fix: Outgoing messages are signed over the exact bytes transmitted - the payload is serialized once and spliced into the envelope. +- fix: The send queue is gated per message rather than as a whole, so a cloud message waiting on the socket no longer blocks a LAN reply queued behind it. +- fix: A request that fails verification is answered with a signed "Signature is invalid" response, letting a client tell a wrong app secret from an unreachable device. +- change: `begin()` no longer raises when the cloud is unreachable. The SDK starts, retries in the background and keeps answering local control; only invalid configuration raises. Call `is_connected()` for cloud state. Callers that relied on `begin()` raising to detect an outage must check it instead. + + +How Local control works: + +- Listens on UDP port `3333`, joined to multicast group `224.9.9.9`, and answers + unicast to this host on the same port. +- Every request is HMAC-SHA256 verified with your `APP_SECRET`; responses are signed + the same way and go back only to the peer that asked. LAN responses are never + echoed to the cloud. +- A request that fails verification is answered with `"Signature is invalid"` rather + than dropped, so a client can tell a wrong secret from an unreachable device. +- The host announces `_sinricpro._udp.local.` over mDNS with TXT records + `deviceIds`, `sdk` and `udp=1`, refreshed whenever the device list changes. + +Check the announcement from another machine on the same network: + +```bash +avahi-browse -r _sinricpro._udp # Linux +dns-sd -B _sinricpro._udp # macOS / Windows (Bonjour) +``` + +### Configuration + +```python +config = SinricProConfig( + app_key="your-app-key", + app_secret="your-app-secret", + local_control=True, # False disables LAN control entirely + mdns=True, # False keeps UDP but skips the announcement + local_control_interface=None, # pin to a LAN interface, e.g. "192.168.1.50" +) +``` + +Set `local_control_interface` on hosts with several interfaces (Docker, VPN, WSL). +Left unset, the SDK joins the group on the interface the OS picks and announces the +address of the default route, which is not always the LAN you want. + +Without `zeroconf` installed the SDK logs a warning and skips the announcement only - +UDP control still works for a client that knows the device address. + ## [5.3.2] - feat: version number diff --git a/README.md b/README.md index f24d1ef..4587de1 100644 --- a/README.md +++ b/README.md @@ -11,6 +11,7 @@ Official Python SDK for [SinricPro](https://sinric.pro) - Control your IoT devic - ✅ **Type Safe** - Full type hints for better IDE support and error detection - ✅ **Voice Control** - Works with Alexa and Google Home - ✅ **Real-time** - WebSocket-based bidirectional communication +- ✅ **Local Control** - Keeps answering the app over the LAN when the cloud is unreachable - ✅ **Secure** - HMAC-SHA256 message signatures - ✅ **Reliable** - Auto-reconnection and heartbeat monitoring - ✅ **Flexible** - Support for multiple device types and capabilities @@ -49,10 +50,18 @@ Official Python SDK for [SinricPro](https://sinric.pro) - Control your IoT devic pip install sinricpro ``` +Local control announces the device over mDNS so the app can discover it on the LAN. +The announcement uses `zeroconf`, installed with the SDK: + +```bash +pip install sinricpro +``` + ## Requirements - Python 3.10 or higher - `websockets` library (automatically installed) +- `zeroconf` - the mDNS announcement used by local control ## Platform Support @@ -63,6 +72,13 @@ The SDK works on: - **macOS** 10.14+ - **Raspberry Pi** (All models with Python 3.10+) +## Local Control + +Devices answer signed commands over the LAN as well as through the cloud, so they +keep responding to the app when sinric.pro is unreachable. It is on by default and +needs no code change - a LAN request runs the same capability callbacks a cloud +request does. UDP listener on port `3333`, joined to multicast group `224.9.9.9` and answering unicast on the same port. Replies go back to the peer that sent the request, never to the cloud websocket. + ## Logging Enable debug logging to see detailed information: @@ -133,6 +149,17 @@ Full API documentation is available at [Read the Docs](https://sinricpro-python. 3. **Check network** - Ensure you have internet connectivity 4. **Enable debug logging** - Set `debug=True` in config to see detailed logs +### Local Control Issues + +1. **No device found on the LAN** - check the log for + `Local control listening on UDP 3333`. A failed multicast join leaves nothing + listening, and the log line says so. +2. **No mDNS record** - confirm `zeroconf` imported cleanly; the SDK logs a warning and falls back to UDP-only when it did not. +3. **Discovery answers on the wrong network** - set `local_control_interface` to the + LAN address of the host. +4. **Android clients need a `WifiManager.MulticastLock`**, and iOS clients need + `_sinricpro._udp` listed in `NSBonjourServices`, or discovery returns nothing. + ### Common Errors **"Invalid app_key format"** diff --git a/pyproject.toml b/pyproject.toml index 44dc306..84e7ac2 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -4,7 +4,7 @@ build-backend = "setuptools.build_meta" [project] name = "sinricpro" -version = "5.3.2" +version = "6.0.0" description = "Official SinricPro SDK for Python - Control IoT devices with Alexa and Google Home" authors = [{name = "SinricPro", email = "support@sinric.com"}] readme = "README.md" @@ -25,7 +25,8 @@ classifiers = [ ] dependencies = [ "websockets>=12.0", - "aiohttp>=3.9.0" + "aiohttp>=3.9.0", + "zeroconf>=0.131.0" ] [project.optional-dependencies] diff --git a/sinricpro/__init__.py b/sinricpro/__init__.py index c748b68..950ece4 100644 --- a/sinricpro/__init__.py +++ b/sinricpro/__init__.py @@ -9,7 +9,7 @@ This file is part of the SinricPro Python SDK (https://github.com/sinricpro/) """ -__version__ = "5.3.2" +__version__ = "6.0.0" from sinricpro.core.sinric_pro import SinricPro, SinricProConfig from sinricpro.core.sinric_pro_device import SinricProDevice diff --git a/sinricpro/core/local_control/__init__.py b/sinricpro/core/local_control/__init__.py new file mode 100644 index 0000000..106b901 --- /dev/null +++ b/sinricpro/core/local_control/__init__.py @@ -0,0 +1,6 @@ +"""Local control (LAN) components: UDP listener and mDNS announcement.""" + +from sinricpro.core.local_control.mdns import ZEROCONF_AVAILABLE, MdnsAnnouncer +from sinricpro.core.local_control.udp_listener import UdpListener + +__all__ = ["ZEROCONF_AVAILABLE", "MdnsAnnouncer", "UdpListener"] diff --git a/sinricpro/core/local_control/mdns.py b/sinricpro/core/local_control/mdns.py new file mode 100644 index 0000000..ca59816 --- /dev/null +++ b/sinricpro/core/local_control/mdns.py @@ -0,0 +1,159 @@ +""" +mDNS Announcer + +Publishes ``_sinricpro._udp.local.`` so the app can find this host on the LAN +without asking the cloud for its address. + +Requires the optional ``zeroconf`` dependency (``pip install sinricpro[mdns]``). +Without it local control still works - the app falls back to the address the +cloud reported for the device. +""" + +import socket +from typing import Any + +from sinricpro.core.types import MDNS_SERVICE_TYPE, UDP_MULTICAST_PORT +from sinricpro.utils.logger import SinricProLogger +from sinricpro.utils.network import get_local_ip, get_mdns_host_name + +try: + from zeroconf import IPVersion, ServiceInfo + from zeroconf.asyncio import AsyncZeroconf + + ZEROCONF_AVAILABLE = True +except ImportError: # pragma: no cover - depends on the install extras + ZEROCONF_AVAILABLE = False + + +class MdnsAnnouncer: + """ + Announces this host as a SinricPro local control endpoint. + + TXT records: + deviceIds comma-separated ids this host answers for + sdk SDK version + udp always "1" + + The record is refreshed when the device list changes - not on a timer and + not on every reconnect. + """ + + def __init__( + self, + sdk_version: str, + interface_ip: str | None = None, + port: int = UDP_MULTICAST_PORT, + host_name: str | None = None, + ) -> None: + """ + Initialize the announcer. + + Args: + sdk_version: Value of the ``sdk`` TXT record + interface_ip: IPv4 address to announce. None discovers the interface + carrying the default route. + port: UDP port to advertise + host_name: mDNS host label; defaults to ``sinricpro-`` + """ + self.sdk_version = sdk_version + self.interface_ip = interface_ip + self.port = port + self.host_name = host_name or get_mdns_host_name() + self._zeroconf: Any = None + self._info: Any = None + self._device_ids: str = "" + + def _build_info(self, device_ids: str) -> Any: + return ServiceInfo( + MDNS_SERVICE_TYPE, + f"{self.host_name}.{MDNS_SERVICE_TYPE}", + addresses=[socket.inet_aton(self.interface_ip)] if self.interface_ip else [], + port=self.port, + properties={ + "deviceIds": device_ids, + "sdk": self.sdk_version, + "udp": "1", + }, + server=f"{self.host_name}.local.", + ) + + async def start(self, device_ids: list[str]) -> bool: + """ + Register the service. + + Args: + device_ids: Device ids this host answers for + + Returns: + True if the service was registered + """ + if not ZEROCONF_AVAILABLE: + SinricProLogger.warn( + "zeroconf is not installed, skipping mDNS announcement " + "(local control still works via the cloud-reported address). " + "Install with: pip install sinricpro[mdns]" + ) + return False + + if self._zeroconf is not None: + return True + + if not self.interface_ip: + self.interface_ip = get_local_ip() + if not self.interface_ip: + SinricProLogger.warn( + "Could not determine a LAN address, skipping mDNS announcement" + ) + return False + + self._device_ids = ",".join(device_ids) + + try: + self._zeroconf = AsyncZeroconf( + interfaces=[self.interface_ip], ip_version=IPVersion.V4Only + ) + self._info = self._build_info(self._device_ids) + await self._zeroconf.async_register_service(self._info) + except Exception as e: + SinricProLogger.error(f"mDNS announcement failed: {e}") + await self.stop() + return False + + SinricProLogger.info( + f"Announced {MDNS_SERVICE_TYPE} as {self.host_name}.local. " + f"on {self.interface_ip}:{self.port} deviceIds={self._device_ids}" + ) + return True + + async def update(self, device_ids: list[str]) -> None: + """ + Re-announce, but only if the device list actually changed. + + Args: + device_ids: Device ids this host answers for + """ + joined = ",".join(device_ids) + if self._zeroconf is None or joined == self._device_ids: + return + + self._device_ids = joined + try: + self._info = self._build_info(joined) + await self._zeroconf.async_update_service(self._info) + SinricProLogger.info(f"mDNS deviceIds updated: {joined}") + except Exception as e: + SinricProLogger.error(f"mDNS update failed: {e}") + + async def stop(self) -> None: + """Unregister the service and close the responder.""" + if self._zeroconf is None: + return + try: + if self._info is not None: + await self._zeroconf.async_unregister_service(self._info) + await self._zeroconf.async_close() + except Exception as e: + SinricProLogger.error(f"mDNS shutdown failed: {e}") + finally: + self._zeroconf = None + self._info = None diff --git a/sinricpro/core/local_control/udp_listener.py b/sinricpro/core/local_control/udp_listener.py new file mode 100644 index 0000000..cf45bb1 --- /dev/null +++ b/sinricpro/core/local_control/udp_listener.py @@ -0,0 +1,178 @@ +""" +UDP Listener + +Receives signed SinricPro commands over the LAN, so a device keeps answering +the app while the cloud is unreachable. +""" + +import asyncio +import socket +from typing import Any + +from sinricpro.core.message_queue import MessageQueue +from sinricpro.core.types import ( + UDP_MULTICAST_IP, + UDP_MULTICAST_PORT, + MessageOrigin, + QueuedMessage, + Transport, +) +from sinricpro.utils.logger import SinricProLogger + + +class _SinricProDatagramProtocol(asyncio.DatagramProtocol): + """Hands every datagram, with its peer, to the listener.""" + + def __init__(self, listener: "UdpListener") -> None: + self._listener = listener + + def datagram_received(self, data: bytes, addr: tuple[str | Any, ...]) -> None: + self._listener._on_datagram(data, (str(addr[0]), int(addr[1]))) + + def error_received(self, exc: Exception) -> None: + SinricProLogger.error(f"UDP error: {exc}") + + def connection_lost(self, exc: Exception | None) -> None: + if exc: + SinricProLogger.error(f"UDP socket closed: {exc}") + + +class UdpListener: + """ + Listens for local control requests on UDP and answers them. + + Joins the SinricPro multicast group and also accepts unicast to this host on + the same port. Replies leave on the listening socket - a separate send-only + socket is a known dead end on lwIP stacks and buys nothing here. + """ + + def __init__( + self, + receive_queue: MessageQueue, + interface_ip: str | None = None, + multicast_ip: str = UDP_MULTICAST_IP, + port: int = UDP_MULTICAST_PORT, + ) -> None: + """ + Initialize the listener. + + Args: + receive_queue: Queue that received requests are pushed onto + interface_ip: IPv4 address of the interface to join the group on. + None lets the OS pick, which is wrong often enough on + multi-homed hosts to be worth configuring. + multicast_ip: Multicast group to join + port: UDP port to listen on + """ + self.receive_queue = receive_queue + self.interface_ip = interface_ip + self.multicast_ip = multicast_ip + self.port = port + self._transport: asyncio.DatagramTransport | None = None + + async def start(self) -> bool: + """ + Bind the socket and join the multicast group. + + Returns: + True if the listener is up, False if local control is unavailable + + Example: + >>> listener = UdpListener(receive_queue) + >>> await listener.start() + """ + if self._transport is not None: + return True + + sock: socket.socket | None = None + try: + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1) + if hasattr(socket, "SO_REUSEPORT"): + try: + sock.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEPORT, 1) + except OSError: + pass # not honoured on every platform; SO_REUSEADDR is enough + + # Bind to INADDR_ANY, not the group: unicast to this host on the + # same port must be received too. + sock.bind(("", self.port)) + + mreq = socket.inet_aton(self.multicast_ip) + socket.inet_aton( + self.interface_ip or "0.0.0.0" + ) + sock.setsockopt(socket.IPPROTO_IP, socket.IP_ADD_MEMBERSHIP, mreq) + sock.setblocking(False) + + loop = asyncio.get_running_loop() + transport, _ = await loop.create_datagram_endpoint( + lambda: _SinricProDatagramProtocol(self), sock=sock + ) + self._transport = transport + except OSError as e: + # A failed group join leaves nothing listening; say so rather than + # letting local control be silently dead. + if sock is not None: + sock.close() + SinricProLogger.error( + f"Could not listen on UDP {self.port} ({self.multicast_ip}), " + f"local control unavailable: {e}" + ) + return False + + SinricProLogger.info( + f"Local control listening on UDP {self.port}, joined {self.multicast_ip}" + + (f" on {self.interface_ip}" if self.interface_ip else "") + ) + return True + + def is_running(self) -> bool: + """ + Check whether the listener is bound. + + Returns: + True if the socket is open + """ + return self._transport is not None + + def _on_datagram(self, data: bytes, peer: tuple[str, int]) -> None: + """Queue a received request together with the peer that sent it.""" + try: + message = data.decode("utf-8") + except UnicodeDecodeError: + SinricProLogger.error(f"Discarding non-UTF-8 datagram from {peer[0]}:{peer[1]}") + return + + SinricProLogger.debug(f"UDP request from {peer[0]}:{peer[1]}: {message}") + self.receive_queue.push_sync( + QueuedMessage(message, MessageOrigin(Transport.UDP, peer)) + ) + + def send(self, message: str, peer: tuple[str, int] | None) -> None: + """ + Send a reply back to the peer that made the request. + + Args: + message: The serialized, signed response + peer: (host, port) the request came from + """ + if not peer or not peer[1]: + SinricProLogger.error("UDP message has no peer to answer, dropping") + return + if self._transport is None: + SinricProLogger.error("UDP listener is not running, dropping reply") + return + + try: + self._transport.sendto(message.encode("utf-8"), peer) + SinricProLogger.debug(f"UDP reply to {peer[0]}:{peer[1]}: {message}") + except OSError as e: + SinricProLogger.error(f"UDP reply to {peer[0]}:{peer[1]} failed: {e}") + + async def stop(self) -> None: + """Close the socket and leave the multicast group.""" + if self._transport is None: + return + self._transport.close() + self._transport = None + SinricProLogger.info("Local control listener stopped") diff --git a/sinricpro/core/message_queue.py b/sinricpro/core/message_queue.py index b95d769..d04cd6c 100644 --- a/sinricpro/core/message_queue.py +++ b/sinricpro/core/message_queue.py @@ -6,76 +6,148 @@ import asyncio from collections import deque +from typing import Callable + +from sinricpro.core.types import WEBSOCKET_ORIGIN, MessageOrigin, QueuedMessage +from sinricpro.utils.logger import SinricProLogger class MessageQueue: """ Thread-safe FIFO message queue. + Entries are :class:`QueuedMessage` - the serialized message plus the origin + it arrived on, so a response knows which transport and peer to go back to. + Plain strings are accepted and default to the websocket origin. + Uses a deque for efficient push/pop operations. """ - def __init__(self) -> None: - """Initialize an empty message queue.""" - self._queue: deque[str] = deque() - self._lock = asyncio.Lock() + def __init__(self, max_size: int = 0) -> None: + """ + Initialize an empty message queue. - async def push(self, message: str) -> None: + Args: + max_size: Drop the oldest entry once the queue grows past this. + 0 (default) means unbounded. + """ + self._queue: deque[QueuedMessage] = deque() + self._lock = asyncio.Lock() + self._max_size = max_size + + @staticmethod + def _coerce( + message: str | QueuedMessage, origin: MessageOrigin | None + ) -> QueuedMessage: + if isinstance(message, QueuedMessage): + return message + return QueuedMessage(message, origin or WEBSOCKET_ORIGIN) + + def _trim(self) -> None: + while self._max_size and len(self._queue) > self._max_size: + dropped = self._queue.popleft() + SinricProLogger.warn( + f"Send queue full ({self._max_size}), dropping oldest " + f"{dropped.origin.transport.value} message" + ) + + async def push( + self, message: str | QueuedMessage, origin: MessageOrigin | None = None + ) -> None: """ Add a message to the queue. Args: - message: The message string to add + message: The message string (or ready-made QueuedMessage) to add + origin: Transport and peer the message belongs to Example: >>> queue = MessageQueue() >>> await queue.push('{"type": "request"}') """ async with self._lock: - self._queue.append(message) + self._queue.append(self._coerce(message, origin)) + self._trim() - def push_sync(self, message: str) -> None: + def push_sync( + self, message: str | QueuedMessage, origin: MessageOrigin | None = None + ) -> None: """ Add a message to the queue synchronously. Args: - message: The message string to add + message: The message string (or ready-made QueuedMessage) to add + origin: Transport and peer the message belongs to Note: This is a synchronous version for use in callbacks. """ - self._queue.append(message) + self._queue.append(self._coerce(message, origin)) + self._trim() - async def pop(self) -> str | None: + def push_front_sync( + self, message: str | QueuedMessage, origin: MessageOrigin | None = None + ) -> None: + """ + Put a message back at the head of the queue, preserving order. + + Args: + message: The message to requeue + origin: Transport and peer the message belongs to + """ + self._queue.appendleft(self._coerce(message, origin)) + + async def pop( + self, predicate: Callable[[QueuedMessage], bool] | None = None + ) -> QueuedMessage | None: """ Remove and return the first message from the queue. + Args: + predicate: Optional filter; the first entry it accepts is removed. + Returns: - The first message in the queue, or None if empty + The first matching message in the queue, or None if there is none Example: >>> queue = MessageQueue() >>> await queue.push("message1") - >>> await queue.pop() + >>> (await queue.pop()).message 'message1' """ async with self._lock: - if self._queue: - return self._queue.popleft() - return None + return self._pop_locked(predicate) - def pop_sync(self) -> str | None: + def pop_sync( + self, predicate: Callable[[QueuedMessage], bool] | None = None + ) -> QueuedMessage | None: """ Remove and return the first message from the queue synchronously. + Args: + predicate: Optional filter; the first entry it accepts is removed. + Returns: - The first message in the queue, or None if empty + The first matching message in the queue, or None if there is none Note: This is a synchronous version for use in non-async contexts. """ - if self._queue: + return self._pop_locked(predicate) + + def _pop_locked( + self, predicate: Callable[[QueuedMessage], bool] | None + ) -> QueuedMessage | None: + if not self._queue: + return None + if predicate is None: return self._queue.popleft() + # Skipping over a message that cannot be delivered yet keeps the gate + # per message: a held cloud message must not block a LAN reply behind it. + for index, entry in enumerate(self._queue): + if predicate(entry): + del self._queue[index] + return entry return None def is_empty(self) -> bool: diff --git a/sinricpro/core/signature.py b/sinricpro/core/signature.py index c78042f..4fdd813 100644 --- a/sinricpro/core/signature.py +++ b/sinricpro/core/signature.py @@ -8,11 +8,15 @@ import hashlib import hmac import json -import re from typing import Any from sinricpro.utils.logger import SinricProLogger +# The envelope is sliced, never re-parsed, so these markers are part of the wire +# contract: the payload is delimited by them and the signature always follows it. +_PAYLOAD_MARKER = '"payload":' +_SIGNATURE_MARKER = ',"signature"' + class Signature: """ @@ -30,6 +34,11 @@ def __init__(self, app_secret: str) -> None: """ self.app_secret = app_secret.encode("utf-8") + def _hmac_b64(self, payload_str: str) -> str: + """Compute the base64 HMAC-SHA256 of an already serialized payload.""" + digest = hmac.new(self.app_secret, payload_str.encode("utf-8"), hashlib.sha256) + return base64.b64encode(digest.digest()).decode("utf-8") + def sign(self, message: dict[str, Any]) -> str: """ Generate HMAC-SHA256 signature for a message. @@ -40,69 +49,107 @@ def sign(self, message: dict[str, Any]) -> str: Returns: Base64-encoded signature string + Note: + Prefer :meth:`sign_message`, which returns the exact bytes to + transmit. Serializing the payload a second time to build the + envelope is what signature mismatches are made of. + Example: >>> sig = Signature("my-secret") >>> message = {"payload": {"action": "setPowerState"}} >>> signature = sig.sign(message) >>> message["signature"] = {"HMAC": signature} """ - # Convert payload to JSON string without spaces payload_str = json.dumps(message["payload"], separators=(",", ":"), sort_keys=False) + signature_b64 = self._hmac_b64(payload_str) + + if "signature" not in message: + message["signature"] = {} + message["signature"]["HMAC"] = signature_b64 + + return signature_b64 + + def sign_message(self, message: dict[str, Any]) -> str: + """ + Sign a message and return the exact string to transmit. + + The payload is serialized once and spliced into the envelope, so the + bytes on the wire are byte-for-byte the bytes that were signed. The + signature is always emitted last, which is what lets a receiver find the + payload by slicing between the markers. + + Args: + message: The message dict to sign (must have 'payload' key) - # Compute HMAC-SHA256 - signature = hmac.new(self.app_secret, payload_str.encode("utf-8"), hashlib.sha256) + Returns: + The serialized, signed message - # Base64 encode - signature_b64 = base64.b64encode(signature.digest()).decode("utf-8") + Example: + >>> sig = Signature("my-secret") + >>> sig.sign_message({"header": {}, "payload": {"action": "setPowerState"}}) + """ + payload_str = json.dumps(message["payload"], separators=(",", ":"), sort_keys=False) + signature_b64 = self._hmac_b64(payload_str) - # Add signature to message if "signature" not in message: message["signature"] = {} message["signature"]["HMAC"] = signature_b64 - return signature_b64 + parts = [ + json.dumps(key) + ":" + json.dumps(value, separators=(",", ":"), sort_keys=False) + for key, value in message.items() + if key not in ("payload", "signature") + ] + parts.append(_PAYLOAD_MARKER + payload_str) + parts.append( + '"signature":' + json.dumps(message["signature"], separators=(",", ":")) + ) + return "{" + ",".join(parts) + "}" def validate(self, message: dict[str, Any] | str) -> bool: """ Validate message signature. Args: - message: Message dict or JSON string containing signature + message: The raw received message, or a parsed message dict Returns: True if signature is valid, False otherwise + Note: + Pass the raw string wherever possible. A parsed dict has already + lost the sender's key order and spacing, so it can only be validated + against a re-serialization that assumes our own conventions. + Example: >>> sig = Signature("my-secret") - >>> is_valid = sig.validate(message_dict) + >>> is_valid = sig.validate(raw_message_string) """ try: - # Convert to dict if string if isinstance(message, str): - message = json.loads(message) + raw: str | None = message + parsed: dict[str, Any] = json.loads(message) + else: + raw = None + parsed = message - # Extract signature from message - if "signature" not in message or "HMAC" not in message["signature"]: + if "signature" not in parsed or "HMAC" not in parsed["signature"]: SinricProLogger.error("Message missing signature") return False - received_signature = message["signature"]["HMAC"] + received_signature = parsed["signature"]["HMAC"] - # Extract payload string from original message - payload_str = self._extract_payload(message) + payload_str = ( + self.extract_payload(raw) + if raw is not None + else json.dumps(parsed["payload"], separators=(",", ":"), sort_keys=False) + ) if not payload_str: SinricProLogger.error("Failed to extract payload for signature validation") return False - # Compute expected signature - expected_signature = hmac.new( - self.app_secret, payload_str.encode("utf-8"), hashlib.sha256 - ) - expected_signature_b64 = base64.b64encode(expected_signature.digest()).decode("utf-8") - - # Compare signatures - is_valid = hmac.compare_digest(received_signature, expected_signature_b64) + is_valid = hmac.compare_digest(received_signature, self._hmac_b64(payload_str)) if not is_valid: SinricProLogger.error("Signature validation failed") @@ -113,20 +160,21 @@ def validate(self, message: dict[str, Any] | str) -> bool: SinricProLogger.error(f"Error validating signature: {e}") return False - def _extract_payload(self, message: dict[str, Any]) -> str: + @staticmethod + def extract_payload(raw: str) -> str: """ - Extract payload as JSON string for signature validation. + Slice the payload out of a received message. Args: - message: The message dictionary + raw: The message exactly as received Returns: - JSON string of the payload + The payload substring, or "" if the markers are not both present """ - try: - # For validation, we need to reconstruct the payload exactly as it was signed - # This means using the same JSON serialization - return json.dumps(message["payload"], separators=(",", ":"), sort_keys=False) - except Exception as e: - SinricProLogger.error(f"Error extracting payload: {e}") + begin = raw.find(_PAYLOAD_MARKER) + if begin < 0: + return "" + end = raw.find(_SIGNATURE_MARKER, begin) + if end < 0: return "" + return raw[begin + len(_PAYLOAD_MARKER) : end] diff --git a/sinricpro/core/sinric_pro.py b/sinricpro/core/sinric_pro.py index 924da62..fc7ba71 100644 --- a/sinricpro/core/sinric_pro.py +++ b/sinricpro/core/sinric_pro.py @@ -10,10 +10,12 @@ import time from typing import Any +from sinricpro import __version__ from sinricpro.core.exceptions import ( SinricProConfigurationError, SinricProDeviceError, ) +from sinricpro.core.local_control import MdnsAnnouncer, UdpListener from sinricpro.core.message_queue import MessageQueue from sinricpro.core.signature import Signature from sinricpro.core.sinric_pro_device import SinricProDevice @@ -22,10 +24,15 @@ SinricProRequest, ConnectedCallback, DisconnectedCallback, + MessageOrigin, PongCallback, ModuleSettingCallback, + QueuedMessage, + Transport, EVENT_LIMIT_STATE, PHYSICAL_INTERACTION, + SEND_QUEUE_MAX, + WEBSOCKET_ORIGIN, ) from sinricpro.core.event_limiter import EventLimiter from sinricpro.core.websocket_client import WebSocketClient, WebSocketConfig @@ -55,8 +62,10 @@ def __init__(self) -> None: self.devices: dict[str, SinricProDevice] = {} self.websocket: WebSocketClient | None = None self.receive_queue = MessageQueue() - self.send_queue = MessageQueue() + self.send_queue = MessageQueue(max_size=SEND_QUEUE_MAX) self.signature: Signature | None = None + self.udp_listener: UdpListener | None = None + self.mdns: MdnsAnnouncer | None = None self.is_initialized = False self._processing_tasks: list[asyncio.Task[None]] = [] self._connected_callbacks: list[ConnectedCallback] = [] @@ -89,7 +98,10 @@ async def begin(self, config: SinricProConfig | dict[str, Any]) -> None: Raises: SinricProConfigurationError: If configuration is invalid - SinricProConnectionError: If connection fails + + A cloud connection failure is not fatal: the SDK stays up, retries in + the background, and keeps answering local control. Call + :meth:`is_connected` to check cloud state. Example: >>> config = SinricProConfig( @@ -119,7 +131,6 @@ async def begin(self, config: SinricProConfig | dict[str, Any]) -> None: # Initialize signature handler self.signature = Signature(self.config.app_secret) - # Initialize WebSocket try: ws_config = WebSocketConfig( server_url=self.config.server_url, @@ -128,23 +139,66 @@ async def begin(self, config: SinricProConfig | dict[str, Any]) -> None: ) self.websocket = WebSocketClient(ws_config) - - # Set up WebSocket event handlers self._setup_websocket_handlers() - # Connect to WebSocket - await self.websocket.connect() + # Local control comes up before the cloud so a host that never + # reaches sinric.pro still answers the app over the LAN. + await self._start_local_control() - # Start message processors + self.is_initialized = True self._start_message_processor() - self.is_initialized = True + try: + await self.websocket.connect() + except Exception as e: + # Transport failure is never fatal -- the reconnect timer is + # armed and local control already answers. Only invalid + # configuration raises, and it has done so above. + SinricProLogger.warn( + f"Cloud connection failed ({e}); will keep retrying" + ) + self.websocket.schedule_reconnect() + SinricProLogger.info("SinricPro SDK initialized successfully") except Exception as e: SinricProLogger.error(f"Failed to initialize SinricPro: {e}") + self.is_initialized = False + await self._stop_local_control() raise + async def _start_local_control(self) -> None: + """Bring up the LAN listener and the mDNS announcement.""" + if not self.config or not self.config.local_control: + SinricProLogger.info("Local control disabled by configuration") + return + + listener = UdpListener( + self.receive_queue, interface_ip=self.config.local_control_interface + ) + if not await listener.start(): + return + self.udp_listener = listener + + if not self.config.mdns: + SinricProLogger.info("mDNS announcement disabled by configuration") + return + + announcer = MdnsAnnouncer( + sdk_version=__version__, interface_ip=self.config.local_control_interface + ) + if await announcer.start(list(self.devices.keys())): + self.mdns = announcer + + async def _stop_local_control(self) -> None: + """Tear down the LAN listener and the mDNS announcement.""" + if self.mdns: + await self.mdns.stop() + self.mdns = None + if self.udp_listener: + await self.udp_listener.stop() + self.udp_listener = None + def add(self, device: SinricProDevice) -> SinricProDevice: """ Add a device to SinricPro. @@ -186,6 +240,15 @@ def add(self, device: SinricProDevice) -> SinricProDevice: if self.is_initialized and self.websocket: self.websocket.update_device_list(list(self.devices.keys())) + # The mDNS record is refreshed only when the device list changes. + if self.mdns: + try: + asyncio.get_running_loop().create_task( + self.mdns.update(list(self.devices.keys())) + ) + except RuntimeError: + SinricProLogger.warn("No running event loop, mDNS record not refreshed") + return device def get(self, device_id: str) -> SinricProDevice | None: @@ -342,6 +405,8 @@ async def stop(self) -> None: task.cancel() self._processing_tasks.clear() + await self._stop_local_control() + # Disconnect WebSocket if self.websocket: await self.websocket.disconnect() @@ -363,11 +428,7 @@ async def send_message(self, message: dict[str, Any]) -> None: SinricProLogger.error("Signature handler not initialized") return - # Sign the message - self.signature.sign(message) - - # Add to send queue - self.send_queue.push_sync(json.dumps(message, separators=(",", ":"), sort_keys=False)) + self.send_queue.push_sync(self.signature.sign_message(message), WEBSOCKET_ORIGIN) def get_timestamp(self) -> int: """ @@ -427,9 +488,9 @@ async def _process_receive_queue(self) -> None: """Process received messages.""" while self.is_initialized: try: - message_str = await self.receive_queue.pop() - if message_str: - await self._handle_message(message_str) + entry = await self.receive_queue.pop() + if entry: + await self._handle_message(entry) else: await asyncio.sleep(0.01) # Small delay if queue is empty except asyncio.CancelledError: @@ -437,29 +498,32 @@ async def _process_receive_queue(self) -> None: except Exception as e: SinricProLogger.error(f"Error processing received message: {e}") - async def _handle_message(self, message_str: str) -> None: - """Handle a received message.""" + async def _handle_message(self, entry: QueuedMessage) -> None: + """Handle a received message, from either the cloud or the LAN.""" try: - message = json.loads(message_str) + message = json.loads(entry.message) + origin = entry.origin # Handle timestamp message if "timestamp" in message: return - # Validate signature - if not self.signature or not self.signature.validate(message): + # Validated against the received bytes: the sender's key order and + # spacing are its own, so a re-serialized dict would not match. + if not self.signature or not self.signature.validate(entry.message): SinricProLogger.error("Invalid message signature") - self._send_invalid_signature_response(message) + self._send_invalid_signature_response(message, origin) return - # Route message + # Route message. LAN requests land in the same handlers as cloud + # requests - there is no second dispatch path. if message["payload"]["type"] == "request": # Check scope to determine if this is a module or device request scope = message["payload"].get("scope", "device") if scope == "module": - await self._handle_module_request(message) + await self._handle_module_request(message, origin) else: - await self._handle_request(message) + await self._handle_request(message, origin) elif message["payload"]["type"] == "response": # Response messages (not typically used in device SDK) pass @@ -467,14 +531,16 @@ async def _handle_message(self, message_str: str) -> None: except Exception as e: SinricProLogger.error(f"Error handling message: {e}") - async def _handle_request(self, message: dict[str, Any]) -> None: + async def _handle_request( + self, message: dict[str, Any], origin: MessageOrigin = WEBSOCKET_ORIGIN + ) -> None: """Handle an incoming request.""" device_id = message["payload"].get("deviceId") device = self.devices.get(device_id) if device_id else None if not device: SinricProLogger.error(f"Device not found: {device_id}") - self._send_error_response(message, f"Device {device_id} not found") + self._send_error_response(message, f"Device {device_id} not found", origin) return request = SinricProRequest( @@ -484,9 +550,13 @@ async def _handle_request(self, message: dict[str, Any]) -> None: ) success = await device.handle_request(request) - self._send_response(message, success, request.response_value, request.error_message) + self._send_response( + message, success, request.response_value, request.error_message, origin + ) - async def _handle_module_request(self, message: dict[str, Any]) -> None: + async def _handle_module_request( + self, message: dict[str, Any], origin: MessageOrigin = WEBSOCKET_ORIGIN + ) -> None: """Handle an incoming module-level request.""" action = message["payload"].get("action", "") request_value = message["payload"].get("value", {}) @@ -494,7 +564,9 @@ async def _handle_module_request(self, message: dict[str, Any]) -> None: if action == "setSetting": if not self._module_setting_callback: SinricProLogger.error("No module setting callback registered") - self._send_module_response(message, False, {}, "No module setting callback registered") + self._send_module_response( + message, False, {}, "No module setting callback registered", origin + ) return setting_id = request_value.get("id", "") @@ -503,13 +575,15 @@ async def _handle_module_request(self, message: dict[str, Any]) -> None: try: success = await self._module_setting_callback(setting_id, value) response_value = {"id": setting_id, "value": value} if success else {} - self._send_module_response(message, success, response_value) + self._send_module_response(message, success, response_value, origin=origin) except Exception as e: SinricProLogger.error(f"Error in module setting callback: {e}") - self._send_module_response(message, False, {}, str(e)) + self._send_module_response(message, False, {}, str(e), origin) else: SinricProLogger.error(f"Unknown module action: {action}") - self._send_module_response(message, False, {}, f"Unknown module action: {action}") + self._send_module_response( + message, False, {}, f"Unknown module action: {action}", origin + ) def _send_module_response( self, @@ -517,19 +591,21 @@ def _send_module_response( success: bool, value: dict[str, Any], error_message: str | None = None, + origin: MessageOrigin = WEBSOCKET_ORIGIN, ) -> None: """Send a module-level response message (without deviceId).""" + request_payload = request_message.get("payload", {}) response_message: dict[str, Any] = { "header": { "payloadVersion": 2, "signatureVersion": 1, }, "payload": { - "action": request_message["payload"]["action"], - "clientId": request_message["payload"]["clientId"], + "action": request_payload.get("action", ""), + "clientId": request_payload.get("clientId", ""), "createdAt": self.get_timestamp(), "message": error_message if error_message else ("OK" if success else "Request failed"), - "replyToken": request_message["payload"]["replyToken"], + "replyToken": request_payload.get("replyToken", ""), "scope": "module", "success": success, "type": "response", @@ -537,10 +613,7 @@ def _send_module_response( }, } - if self.signature: - self.signature.sign(response_message) - - self.send_queue.push_sync(json.dumps(response_message, separators=(",", ":"), sort_keys=False)) + self._enqueue_response(response_message, origin) def _send_response( self, @@ -548,20 +621,24 @@ def _send_response( success: bool, value: dict[str, Any], error_message: str | None = None, + origin: MessageOrigin = WEBSOCKET_ORIGIN, ) -> None: """Send a response message.""" + # A request that failed verification may be missing anything, so every + # echoed field is read defensively. + request_payload = request_message.get("payload", {}) response_message: dict[str, Any] = { "header": { "payloadVersion": 2, "signatureVersion": 1, }, "payload": { - "action": request_message["payload"]["action"], - "clientId": request_message["payload"]["clientId"], + "action": request_payload.get("action", ""), + "clientId": request_payload.get("clientId", ""), "createdAt": self.get_timestamp(), - "deviceId": request_message["payload"]["deviceId"], + "deviceId": request_payload.get("deviceId", ""), "message": error_message if error_message else ("OK" if success else "Request failed"), - "replyToken": request_message["payload"]["replyToken"], + "replyToken": request_payload.get("replyToken", ""), "scope": "device", "success": success, "type": "response", @@ -569,43 +646,72 @@ def _send_response( }, } - if "instanceId" in request_message["payload"]: - response_message["payload"]["instanceId"] = request_message["payload"]["instanceId"] + if "instanceId" in request_payload: + response_message["payload"]["instanceId"] = request_payload["instanceId"] - if self.signature: - self.signature.sign(response_message) + self._enqueue_response(response_message, origin) - self.send_queue.push_sync(json.dumps(response_message, separators=(",", ":"), sort_keys=False)) + def _enqueue_response( + self, response_message: dict[str, Any], origin: MessageOrigin + ) -> None: + """Sign a response and queue it for the transport it must go back on.""" + if not self.signature: + SinricProLogger.error("Signature handler not initialized") + return + self.send_queue.push_sync(self.signature.sign_message(response_message), origin) - def _send_error_response(self, message: dict[str, Any], error_message: str) -> None: + def _send_error_response( + self, + message: dict[str, Any], + error_message: str, + origin: MessageOrigin = WEBSOCKET_ORIGIN, + ) -> None: """Send an error response.""" - self._send_response(message, False, {"error": error_message}, error_message) + self._send_response(message, False, {"error": error_message}, error_message, origin) - def _send_invalid_signature_response(self, message: dict[str, Any]) -> None: - """Send invalid signature response.""" - self._send_error_response(message, "Invalid signature") + def _send_invalid_signature_response( + self, message: dict[str, Any], origin: MessageOrigin = WEBSOCKET_ORIGIN + ) -> None: + """Answer a request that failed verification. + + Answering rather than dropping is deliberate: it lets a client tell a + wrong app secret apart from an unreachable device. + """ + self._send_response(message, False, {}, "Signature is invalid", origin) async def _process_send_queue(self) -> None: """Process outgoing messages.""" while self.is_initialized: try: - if not self.is_connected(): - await asyncio.sleep(0.1) + # Gate per message, not per queue: a cloud message waiting for + # the socket must not hold up a LAN reply queued behind it. + connected = self.is_connected() + entry = self.send_queue.pop_sync( + lambda m: connected or m.origin.transport is not Transport.WEBSOCKET + ) + + if entry is None: + await asyncio.sleep(0.01) continue - message_str = self.send_queue.pop_sync() - if message_str and self.websocket: - try: - self.websocket.send(message_str) - except Exception as e: - # If send fails, put message back in queue - self.send_queue.push_sync(message_str) - SinricProLogger.error(f"Failed to send message, will retry later: {e}") - await asyncio.sleep(1) - else: - await asyncio.sleep(0.01) + if entry.origin.transport is Transport.UDP: + # LAN responses are never echoed to the cloud socket. + if self.udp_listener: + self.udp_listener.send(entry.message, entry.origin.peer) + continue + + if not self.websocket: + continue + + try: + self.websocket.send(entry.message) + except Exception as e: + self.send_queue.push_front_sync(entry) + SinricProLogger.error(f"Failed to send message, will retry later: {e}") + await asyncio.sleep(1) except asyncio.CancelledError: break except Exception as e: SinricProLogger.error(f"Error processing send queue: {e}") + await asyncio.sleep(0.01) diff --git a/sinricpro/core/types.py b/sinricpro/core/types.py index a395c4d..61090da 100644 --- a/sinricpro/core/types.py +++ b/sinricpro/core/types.py @@ -4,9 +4,10 @@ Common types, protocols, and constants used throughout the SDK. """ -from dataclasses import dataclass, field -from typing import Protocol, Any, Callable, Awaitable import re +from dataclasses import dataclass, field +from enum import Enum +from typing import Any, Awaitable, Callable, Protocol from sinricpro.core.exceptions import SinricProConfigurationError @@ -20,6 +21,14 @@ EVENT_LIMIT_STATE = 1000 # 1 second in milliseconds EVENT_LIMIT_SENSOR_VALUE = 60000 # 60 seconds in milliseconds +# Local control (LAN) - wire contract shared with the app and every other SinricPro SDK +UDP_MULTICAST_IP = "224.9.9.9" +UDP_MULTICAST_PORT = 3333 +MDNS_SERVICE_TYPE = "_sinricpro._udp.local." + +# Outgoing messages held while the cloud is unreachable. Oldest are dropped past this. +SEND_QUEUE_MAX = 128 + # Interaction types PHYSICAL_INTERACTION = "PHYSICAL_INTERACTION" APP_INTERACTION = "APP_INTERACTION" @@ -37,6 +46,42 @@ ModuleSettingCallback = Callable[[str, Any], Awaitable[bool]] +class Transport(str, Enum): + """Transport a message arrived on / must be answered over.""" + + WEBSOCKET = "websocket" + UDP = "udp" + + +@dataclass(frozen=True) +class MessageOrigin: + """ + Where a message came from, and where its response must go. + + The peer is carried per message rather than held on the listener: a response + can be sent several loop iterations after the request was queued, by which + time another peer may have sent a packet. + + Attributes: + transport: Transport the message arrived on + peer: (host, port) of the sender, for UDP only + """ + + transport: Transport = Transport.WEBSOCKET + peer: tuple[str, int] | None = None + + +WEBSOCKET_ORIGIN = MessageOrigin(Transport.WEBSOCKET) + + +@dataclass(frozen=True) +class QueuedMessage: + """A serialized message plus the origin that decides how it is delivered.""" + + message: str + origin: MessageOrigin = WEBSOCKET_ORIGIN + + @dataclass class SinricProConfig: """ @@ -47,12 +92,22 @@ class SinricProConfig: app_secret: SinricPro app secret (min 32 characters) server_url: WebSocket server URL (default: ws.sinric.pro) debug: Enable debug logging + local_control: Answer signed commands over the LAN (UDP 3333) + mdns: Announce the device over mDNS so the app can discover it. + Requires the optional ``zeroconf`` dependency; ignored without it. + local_control_interface: IPv4 address of the interface to bind the + multicast join to and to announce over mDNS. Leave unset to let the + OS choose - set it on multi-homed hosts (Docker, VPN, WSL) where the + default is not the LAN. """ app_key: str app_secret: str server_url: str = SINRICPRO_SERVER_URL debug: bool = False + local_control: bool = True + mdns: bool = True + local_control_interface: str | None = None def __post_init__(self) -> None: """Validate configuration after initialization.""" diff --git a/sinricpro/core/websocket_client.py b/sinricpro/core/websocket_client.py index c8b67d3..60997c0 100644 --- a/sinricpro/core/websocket_client.py +++ b/sinricpro/core/websocket_client.py @@ -7,23 +7,9 @@ import asyncio import time -import uuid from typing import Callable import websockets - - -def get_mac_address() -> str: - """Get the MAC address of this machine. - - Returns: - MAC address string in format XX:XX:XX:XX:XX:XX - """ - mac = uuid.getnode() - # Format as XX:XX:XX:XX:XX:XX - return ":".join(f"{(mac >> (8 * i)) & 0xFF:02X}" for i in range(5, -1, -1)) - - from websockets.client import WebSocketClientProtocol from sinricpro import __version__ @@ -35,6 +21,7 @@ def get_mac_address() -> str: WEBSOCKET_PONG_MISS_MAX, ) from sinricpro.utils.logger import SinricProLogger +from sinricpro.utils.network import get_mac_address class WebSocketConfig: @@ -263,6 +250,10 @@ def _stop_heartbeat(self) -> None: self._ping_task.cancel() self._ping_task = None + def schedule_reconnect(self) -> None: + """Schedule a reconnection attempt after the current backoff.""" + self._schedule_reconnect() + def _schedule_reconnect(self) -> None: """Schedule automatic reconnection.""" if self._reconnect_task: diff --git a/sinricpro/utils/__init__.py b/sinricpro/utils/__init__.py index a2a47bf..4719680 100644 --- a/sinricpro/utils/__init__.py +++ b/sinricpro/utils/__init__.py @@ -1,5 +1,12 @@ """Utility modules for SinricPro SDK.""" -from sinricpro.utils.logger import SinricProLogger, LogLevel +from sinricpro.utils.logger import LogLevel, SinricProLogger +from sinricpro.utils.network import get_local_ip, get_mac_address, get_mdns_host_name -__all__ = ["SinricProLogger", "LogLevel"] +__all__ = [ + "LogLevel", + "SinricProLogger", + "get_local_ip", + "get_mac_address", + "get_mdns_host_name", +] diff --git a/sinricpro/utils/network.py b/sinricpro/utils/network.py new file mode 100644 index 0000000..6022020 --- /dev/null +++ b/sinricpro/utils/network.py @@ -0,0 +1,48 @@ +""" +Network Helpers + +Host identity and interface discovery shared by the websocket client and the +local control listener. +""" + +import socket +import uuid + + +def get_mac_address() -> str: + """Get the MAC address of this machine. + + Returns: + MAC address string in format XX:XX:XX:XX:XX:XX + """ + mac = uuid.getnode() + return ":".join(f"{(mac >> (8 * i)) & 0xFF:02X}" for i in range(5, -1, -1)) + + +def get_mdns_host_name() -> str: + """Get the mDNS host label for this machine. + + Returns: + ``sinricpro-`` + """ + return "sinricpro-" + get_mac_address().replace(":", "").lower() + + +def get_local_ip() -> str | None: + """Get the IPv4 address of the interface that carries the default route. + + A multi-homed host (Docker, VPN, WSL) has several; this picks the one the + OS would use to reach off-box, which is the LAN interface in the common + case. No packet is sent - the socket is only connected to pick a route. + + Returns: + The interface address, or None if it could not be determined + """ + sock = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + try: + sock.connect(("8.8.8.8", 53)) + return str(sock.getsockname()[0]) + except OSError: + return None + finally: + sock.close() diff --git a/tests/unit/test_local_control.py b/tests/unit/test_local_control.py new file mode 100644 index 0000000..5e844b2 --- /dev/null +++ b/tests/unit/test_local_control.py @@ -0,0 +1,259 @@ +"""Local control: message origin, LAN dispatch and the invalid-signature reply.""" + +import asyncio +import json +import socket +from typing import Any, Iterator + +import pytest + +from sinricpro.core.local_control.udp_listener import UdpListener +from sinricpro.core.message_queue import MessageQueue +from sinricpro.core.signature import Signature +from sinricpro.core.sinric_pro import SinricPro +from sinricpro.core.types import ( + UDP_MULTICAST_PORT, + WEBSOCKET_ORIGIN, + MessageOrigin, + QueuedMessage, + SinricProConfig, + Transport, +) +from sinricpro.devices.sinric_pro_switch import SinricProSwitch + +APP_KEY = "8bc4a3fa-1b46-4ff1-9d1a-5ba6bd0a1234" +APP_SECRET = "8bc4a3fa-1b46-4ff1-9d1a-5ba6bd0a1234-1c7fd1c4-2b0e-4a5c-9d9e-9f7c0f9a4321" +DEVICE_ID = "5dc1564130a1b2c3d4e5f607" +PEER = ("192.168.1.42", 51234) + + +@pytest.fixture +def sinric_pro() -> Iterator[SinricPro]: + """A SinricPro instance wired up without touching the network.""" + SinricPro._instance = None + sp = SinricPro.get_instance() + sp.config = SinricProConfig(app_key=APP_KEY, app_secret=APP_SECRET) + sp.signature = Signature(APP_SECRET) + yield sp + SinricPro._instance = None + + +def build_request(action: str = "setPowerState", value: dict[str, Any] | None = None) -> dict[str, Any]: + return { + "header": {"payloadVersion": 2, "signatureVersion": 1}, + "payload": { + "action": action, + "clientId": "alexa-skill", + "createdAt": 1700000000, + "deviceId": DEVICE_ID, + "replyToken": "6f1c0d5e-1a2b", + "type": "request", + "value": value if value is not None else {"state": "On"}, + }, + } + + +def udp_entry(raw: str) -> QueuedMessage: + return QueuedMessage(raw, MessageOrigin(Transport.UDP, PEER)) + + +class TestQueueOrigin: + def test_plain_strings_default_to_the_websocket_origin(self) -> None: + queue = MessageQueue() + queue.push_sync("hello") + + entry = queue.pop_sync() + assert entry is not None + assert entry.message == "hello" + assert entry.origin == WEBSOCKET_ORIGIN + + def test_peer_is_carried_per_message(self) -> None: + """Two peers in flight at once must not share one slot on the listener.""" + queue = MessageQueue() + queue.push_sync(QueuedMessage("a", MessageOrigin(Transport.UDP, ("10.0.0.1", 1111)))) + queue.push_sync(QueuedMessage("b", MessageOrigin(Transport.UDP, ("10.0.0.2", 2222)))) + + first = queue.pop_sync() + second = queue.pop_sync() + assert first is not None and second is not None + assert first.origin.peer == ("10.0.0.1", 1111) + assert second.origin.peer == ("10.0.0.2", 2222) + + def test_pop_skips_entries_the_predicate_rejects(self) -> None: + queue = MessageQueue() + queue.push_sync("cloud") + queue.push_sync(QueuedMessage("lan", MessageOrigin(Transport.UDP, PEER))) + + entry = queue.pop_sync(lambda m: m.origin.transport is Transport.UDP) + assert entry is not None + assert entry.message == "lan" + assert len(queue) == 1 + + def test_oldest_entries_are_dropped_past_max_size(self) -> None: + queue = MessageQueue(max_size=2) + for i in range(4): + queue.push_sync(str(i)) + + assert [queue.pop_sync().message for _ in range(2)] == ["2", "3"] # type: ignore[union-attr] + + +class TestInvalidSignatureReply: + async def test_udp_request_with_a_bad_signature_gets_a_signed_reply( + self, sinric_pro: SinricPro + ) -> None: + raw = Signature("the-wrong-secret").sign_message(build_request()) + + await sinric_pro._handle_message(udp_entry(raw)) + + entry = sinric_pro.send_queue.pop_sync() + assert entry is not None + assert entry.origin.transport is Transport.UDP + assert entry.origin.peer == PEER + + response = json.loads(entry.message) + assert response["payload"]["success"] is False + assert response["payload"]["message"] == "Signature is invalid" + assert response["payload"]["replyToken"] == "6f1c0d5e-1a2b" + # Signed with our secret so the client can tell a wrong secret from silence. + assert sinric_pro.signature is not None + assert sinric_pro.signature.validate(entry.message) is True + + async def test_reply_survives_a_request_missing_every_echoed_field( + self, sinric_pro: SinricPro + ) -> None: + raw = '{"header":{},"payload":{"type":"request"},"signature":{"HMAC":"bogus"}}' + + await sinric_pro._handle_message(udp_entry(raw)) + + entry = sinric_pro.send_queue.pop_sync() + assert entry is not None + assert json.loads(entry.message)["payload"]["message"] == "Signature is invalid" + + +class TestLanDispatch: + async def test_udp_request_reaches_the_device_callback(self, sinric_pro: SinricPro) -> None: + seen: list[bool] = [] + + async def on_power_state(state: bool) -> bool: + seen.append(state) + return True + + switch = SinricProSwitch(DEVICE_ID) + switch.on_power_state(on_power_state) + sinric_pro.add(switch) + + assert sinric_pro.signature is not None + raw = sinric_pro.signature.sign_message(build_request()) + await sinric_pro._handle_message(udp_entry(raw)) + + assert seen == [True] + + entry = sinric_pro.send_queue.pop_sync() + assert entry is not None + assert entry.origin == MessageOrigin(Transport.UDP, PEER) + assert json.loads(entry.message)["payload"]["success"] is True + + async def test_cloud_request_still_answers_over_the_websocket( + self, sinric_pro: SinricPro + ) -> None: + async def on_power_state(state: bool) -> bool: + return True + + switch = SinricProSwitch(DEVICE_ID) + switch.on_power_state(on_power_state) + sinric_pro.add(switch) + + assert sinric_pro.signature is not None + raw = sinric_pro.signature.sign_message(build_request()) + await sinric_pro._handle_message(QueuedMessage(raw)) + + entry = sinric_pro.send_queue.pop_sync() + assert entry is not None + assert entry.origin.transport is Transport.WEBSOCKET + + +class TestSendQueueGating: + async def test_lan_reply_is_sent_while_the_cloud_is_unreachable( + self, sinric_pro: SinricPro + ) -> None: + """A held cloud message must not block a LAN reply queued behind it.""" + sent: list[tuple[str, tuple[str, int] | None]] = [] + + class FakeListener: + def send(self, message: str, peer: tuple[str, int] | None) -> None: + sent.append((message, peer)) + + sinric_pro.udp_listener = FakeListener() # type: ignore[assignment] + sinric_pro.send_queue.push_sync("cloud-event") + sinric_pro.send_queue.push_sync(QueuedMessage("lan-reply", MessageOrigin(Transport.UDP, PEER))) + sinric_pro.is_initialized = True + + task = asyncio.create_task(sinric_pro._process_send_queue()) + await asyncio.sleep(0.05) + sinric_pro.is_initialized = False + task.cancel() + + assert sent == [("lan-reply", PEER)] + # The cloud message is held, not dropped. + assert len(sinric_pro.send_queue) == 1 + + +class TestUdpListener: + async def test_datagram_is_queued_with_its_peer(self) -> None: + queue = MessageQueue() + listener = UdpListener(queue) + + listener._on_datagram(b'{"payload":{}}', PEER) + + entry = queue.pop_sync() + assert entry is not None + assert entry.message == '{"payload":{}}' + assert entry.origin == MessageOrigin(Transport.UDP, PEER) + + async def test_non_utf8_datagram_is_discarded(self) -> None: + queue = MessageQueue() + listener = UdpListener(queue) + + listener._on_datagram(b"\xff\xfe\x00", PEER) + + assert queue.is_empty() + + async def test_reply_goes_out_on_the_listening_socket(self) -> None: + sent: list[tuple[bytes, tuple[str, int]]] = [] + + class FakeTransport: + def sendto(self, data: bytes, addr: tuple[str, int]) -> None: + sent.append((data, addr)) + + listener = UdpListener(MessageQueue()) + listener._transport = FakeTransport() # type: ignore[assignment] + + listener.send("pong", PEER) + listener.send("dropped", None) + + assert sent == [(b"pong", PEER)] + + async def test_binds_and_receives_over_the_loopback(self) -> None: + queue = MessageQueue() + listener = UdpListener(queue, port=UDP_MULTICAST_PORT) + + if not await listener.start(): + pytest.skip("cannot bind UDP 3333 / join multicast in this environment") + + try: + sender = socket.socket(socket.AF_INET, socket.SOCK_DGRAM) + sender.sendto(b'{"hello":1}', ("127.0.0.1", UDP_MULTICAST_PORT)) + sender.close() + + for _ in range(50): + if not queue.is_empty(): + break + await asyncio.sleep(0.02) + finally: + await listener.stop() + + entry = queue.pop_sync() + assert entry is not None + assert entry.message == '{"hello":1}' + assert entry.origin.transport is Transport.UDP + assert entry.origin.peer is not None diff --git a/tests/unit/test_signature.py b/tests/unit/test_signature.py new file mode 100644 index 0000000..1f5c5bb --- /dev/null +++ b/tests/unit/test_signature.py @@ -0,0 +1,139 @@ +"""Signing and verification of the SinricPro message envelope.""" + +import base64 +import hashlib +import hmac +import json + +from sinricpro.core.signature import Signature + +APP_SECRET = "8bc4a3fa-1b46-4ff1-9d1a-5ba6bd0a1234-1c7fd1c4-2b0e-4a5c-9d9e-9f7c0f9a4321" + + +def hmac_b64(payload_str: str, secret: str = APP_SECRET) -> str: + digest = hmac.new(secret.encode(), payload_str.encode(), hashlib.sha256).digest() + return base64.b64encode(digest).decode() + + +def make_request() -> dict[str, object]: + return { + "header": {"payloadVersion": 2, "signatureVersion": 1}, + "payload": { + "action": "setPowerState", + "clientId": "alexa-skill", + "createdAt": 1700000000, + "deviceId": "5dc1564130xxxxxxxxxxxxxx", + "replyToken": "6f1c0d5e-1a2b", + "type": "request", + "value": {"state": "On"}, + }, + } + + +class TestSigning: + def test_sign_message_round_trips(self) -> None: + sig = Signature(APP_SECRET) + raw = sig.sign_message(make_request()) + + assert sig.validate(raw) is True + + def test_signed_bytes_are_the_transmitted_bytes(self) -> None: + """The HMAC must cover the payload substring that actually goes out.""" + sig = Signature(APP_SECRET) + message = make_request() + raw = sig.sign_message(message) + + payload_slice = Signature.extract_payload(raw) + assert payload_slice in raw + assert json.loads(raw)["signature"]["HMAC"] == hmac_b64(payload_slice) + + def test_signature_is_last_key(self) -> None: + """extract_payload() relies on ,"signature" following the payload.""" + raw = Signature(APP_SECRET).sign_message(make_request()) + + assert raw.index('"payload":') < raw.index(',"signature"') + assert raw.endswith("}}") + + def test_envelope_key_order_matches_the_wire_contract(self) -> None: + raw = Signature(APP_SECRET).sign_message(make_request()) + + assert raw.startswith('{"header":{"payloadVersion":2,"signatureVersion":1},"payload":') + + def test_sign_populates_signature_in_place(self) -> None: + sig = Signature(APP_SECRET) + message = make_request() + returned = sig.sign(message) + + assert message["signature"]["HMAC"] == returned # type: ignore[index] + + +class TestVerification: + def test_verifies_by_slicing_received_bytes(self) -> None: + """A foreign sender's key order and spacing are its own.""" + payload_str = '{"value":{"state":"On"},"action":"setPowerState","deviceId":"abc"}' + raw = ( + '{"header": {"payloadVersion": 2, "signatureVersion": 1},' + '"payload":' + payload_str + ',' + '"signature":{"HMAC":"' + hmac_b64(payload_str) + '"}}' + ) + + assert Signature(APP_SECRET).validate(raw) is True + + def test_key_order_independence(self) -> None: + """Two orderings of the same payload each verify against their own bytes.""" + sig = Signature(APP_SECRET) + a = '{"action":"setPowerState","deviceId":"abc"}' + b = '{"deviceId":"abc","action":"setPowerState"}' + + for payload_str in (a, b): + raw = ( + '{"header":{"payloadVersion":2,"signatureVersion":1},' + '"payload":' + payload_str + ',' + '"signature":{"HMAC":"' + hmac_b64(payload_str) + '"}}' + ) + assert sig.validate(raw) is True + + def test_reserializing_a_parsed_dict_would_reject_a_valid_message(self) -> None: + """Why validation slices: re-encoding imposes our own conventions.""" + payload_str = '{"action": "setPowerState", "deviceId": "abc"}' # sender used spaces + raw = ( + '{"header":{"payloadVersion":2,"signatureVersion":1},' + '"payload":' + payload_str + ',' + '"signature":{"HMAC":"' + hmac_b64(payload_str) + '"}}' + ) + sig = Signature(APP_SECRET) + + assert sig.validate(raw) is True + assert sig.validate(json.loads(raw)) is False + + def test_tampered_payload_is_rejected(self) -> None: + sig = Signature(APP_SECRET) + raw = sig.sign_message(make_request()) + tampered = raw.replace('"state":"On"', '"state":"Off"') + + assert tampered != raw + assert sig.validate(tampered) is False + + def test_wrong_secret_is_rejected(self) -> None: + raw = Signature(APP_SECRET).sign_message(make_request()) + + assert Signature("a-different-secret").validate(raw) is False + + def test_missing_signature_is_rejected(self) -> None: + raw = '{"header":{},"payload":{"action":"setPowerState"}}' + + assert Signature(APP_SECRET).validate(raw) is False + + def test_garbage_is_rejected(self) -> None: + assert Signature(APP_SECRET).validate("not json at all") is False + + +class TestExtractPayload: + def test_returns_the_exact_substring(self) -> None: + raw = '{"header":{},"payload":{"a":1},"signature":{"HMAC":"x"}}' + + assert Signature.extract_payload(raw) == '{"a":1}' + + def test_returns_empty_when_markers_are_absent(self) -> None: + assert Signature.extract_payload('{"payload":{"a":1}}') == "" + assert Signature.extract_payload('{"signature":{}}') == "" From 783512b2c8a3bfb6471071de60c6e4e370d292bd Mon Sep 17 00:00:00 2001 From: Aruna Tennakoon Date: Mon, 31 Aug 2026 22:46:26 +0700 Subject: [PATCH 2/2] fix: broadcast-based discovery --- sinricpro/core/sinric_pro.py | 6 ++++++ 1 file changed, 6 insertions(+) diff --git a/sinricpro/core/sinric_pro.py b/sinricpro/core/sinric_pro.py index fc7ba71..2e17591 100644 --- a/sinricpro/core/sinric_pro.py +++ b/sinricpro/core/sinric_pro.py @@ -539,6 +539,12 @@ async def _handle_request( device = self.devices.get(device_id) if device_id else None if not device: + if origin.transport is Transport.UDP: + SinricProLogger.debug( + f"Ignoring LAN request for unknown device: {device_id}" + ) + return + SinricProLogger.error(f"Device not found: {device_id}") self._send_error_response(message, f"Device {device_id} not found", origin) return