Handle timed writes as the HAP spec describes

pyhap/accessory_driver.py

@@ -29,7 +29,7 @@
 import tempfile
 import threading
 import time
-from typing import Any
+from typing import Any, TypeGuard
 
 from zeroconf import ServiceInfo
 from zeroconf.asyncio import AsyncZeroconf
@@ -45,6 +45,7 @@
 from pyhap.const import (
     HAP_PERMISSION_NOTIFY,
     HAP_PERMISSION_READ,
+    HAP_PERMISSION_TIMED_WRITE,
     HAP_PERMISSION_WRITE,
     HAP_PERMISSION_WRITE_RESPONSE,
     HAP_PROTOCOL_SHORT_VERSION,
@@ -93,6 +94,11 @@
 MAX_PREPARED_WRITE_TTL_SECONDS = 30.0
 
 
+def _is_hap_int(value: Any) -> TypeGuard[int]:
+    """Whether ``value`` is a JSON integer, not a bool."""
+    return isinstance(value, int) and not isinstance(value, bool)
+
+
 def _stored_value(char: Characteristic, value: HAPValue) -> HAPValue:
     """Return the value ``client_update_value`` stores for ``value``."""
     try:
@@ -247,6 +253,7 @@ def __init__(
         async_zeroconf_instance: Any = None,
         zeroconf_server: Any = None,
         advertise: bool = True,
+        enforce_timed_writes: bool = False,
     ) -> None:
         """Initialize a new AccessoryDriver object.
 
@@ -301,6 +308,13 @@ def __init__(
             the driver starts. Pass False to run the HAP server without
             mDNS advertising, e.g. for tests or benchmarks.
         :type advertise: bool
+
+        :param enforce_timed_writes: Whether to reject a plain write to a
+            characteristic with the timed write ("tw") permission with -70410,
+            as the HAP specification requires. Off by default, because Home
+            hubs write some tw characteristics, e.g. HomeKit Camera Active,
+            without a timed write.
+        :type enforce_timed_writes: bool
         """
         if loop is None:
             if sys.platform == "win32":
@@ -333,6 +347,7 @@ def __init__(
         self.stop_event = threading.Event()
 
         self.safe_mode = False
+        self.enforce_timed_writes = enforce_timed_writes
 
         self.mdns_service_info: ServiceInfo | None = None
         self._srp_verifier: SrpServer | None = None
@@ -1060,6 +1075,23 @@ def set_characteristics(
         """
         queries: list[dict[str, Any]] = chars_query[HAP_REPR_CHARS]
 
+        timed_write = HAP_REPR_PID in chars_query
+        if timed_write and not self._take_prepared_write(
+            chars_query[HAP_REPR_PID], client_addr
+        ):
+            # The spec requires a timed write that was never prepared, was
+            # already used or has expired to be ignored as a whole.
+            return {
+                HAP_REPR_CHARS: [
+                    {
+                        HAP_REPR_AID: query[HAP_REPR_AID],
+                        HAP_REPR_IID: query[HAP_REPR_IID],
+                        HAP_REPR_STATUS: HAP_SERVER_STATUS.INVALID_VALUE_IN_REQUEST,
+                    }
+                    for query in queries
+                ]
+            }
+
         not_notifiable = self._notify(queries, client_addr)
 
         updates_by_accessories_services: dict[
@@ -1070,12 +1102,6 @@ def set_characteristics(
         )
         char_to_iid: dict[Characteristic, int] = {}
 
-        expired = False
-        if HAP_REPR_PID in chars_query:
-            pid = chars_query[HAP_REPR_PID]
-            expire_time = self.prepared_writes.get(client_addr, {}).pop(pid, None)
-            expired = expire_time is None or time.time() > expire_time
-
         primary_accessory = self.accessory
         assert primary_accessory is not None  # noqa: S101
         primary_aid = primary_accessory.aid
@@ -1090,7 +1116,7 @@ def set_characteristics(
             iid = query[HAP_REPR_IID]
             if (aid, iid) in not_notifiable:
                 continue
-            if HAP_REPR_VALUE not in query and not expired:
+            if HAP_REPR_VALUE not in query:
                 # Listed with status 0 if the response has other results.
                 results[aid][iid] = {HAP_REPR_STATUS: HAP_SERVER_STATUS.SUCCESS}
                 continue
@@ -1120,6 +1146,22 @@ def set_characteristics(
                 }
                 continue
 
+            if (
+                value is not None
+                and not timed_write
+                and HAP_PERMISSION_TIMED_WRITE in char.properties[PROP_PERMISSIONS]
+            ):
+                if self.enforce_timed_writes:
+                    results[aid][iid] = {
+                        HAP_REPR_STATUS: HAP_SERVER_STATUS.INVALID_VALUE_IN_REQUEST
+                    }
+                    continue
+                logger.debug(
+                    "%s: %s requires a timed write but was written without one",
+                    client_addr,
+                    char.display_name,
+                )
+
             # Only characteristics with the write response permission answer
             # a write with a value, and only when the controller asks for it.
             write_response = (
@@ -1208,6 +1250,14 @@ def set_characteristics(
 
         return {HAP_REPR_CHARS: characteristics} if nonempty_results_exist else None
 
+    def _take_prepared_write(self, pid: Any, client_addr: tuple[str, int]) -> bool:
+        """Consume the prepared write ``pid`` and return whether it is still valid."""
+        if not _is_hap_int(pid):
+            # Controller input. A list or dict is not even hashable.
+            return False
+        expires_at = self.prepared_writes.get(client_addr, {}).pop(pid, None)
+        return expires_at is not None and time.monotonic() <= expires_at
+
     def _store_write_response(
         self,
         char: Characteristic,
@@ -1258,24 +1308,28 @@ def prepare(
 
         :type prepare_query: dict
         """
+        if not isinstance(prepare_query, dict):
+            return {HAP_REPR_STATUS: HAP_SERVER_STATUS.INVALID_VALUE_IN_REQUEST}
+        ttl = prepare_query.get(HAP_REPR_TTL)
+        pid = prepare_query.get(HAP_REPR_PID)
+        if not (_is_hap_int(ttl) and _is_hap_int(pid)):
+            return {HAP_REPR_STATUS: HAP_SERVER_STATUS.INVALID_VALUE_IN_REQUEST}
         try:
-            ttl = prepare_query[HAP_REPR_TTL]
-            pid = prepare_query[HAP_REPR_PID]
-            now = time.time()
+            now = time.monotonic()
             pending = self.prepared_writes.setdefault(client_addr, {})
             # Prune expired entries and cap the number of pending writes so a
             # paired client cannot grow prepared_writes without bound by
             # registering endless unique pids with arbitrary TTLs.
             for stale_pid in [
                 prepared_pid
                 for prepared_pid, expires_at in pending.items()
-                if expires_at <= now
+                if expires_at < now
             ]:
                 del pending[stale_pid]
             if len(pending) >= MAX_PREPARED_WRITES_PER_CLIENT:
                 del pending[min(pending, key=pending.get)]
             pending[pid] = now + min(ttl / 1000, MAX_PREPARED_WRITE_TTL_SECONDS)
-        except (KeyError, TypeError, ValueError):
+        except OverflowError:
             return {HAP_REPR_STATUS: HAP_SERVER_STATUS.INVALID_VALUE_IN_REQUEST}
 
         return {HAP_REPR_STATUS: HAP_SERVER_STATUS.SUCCESS}

pyhap/const.py

@@ -56,6 +56,7 @@
 HAP_PERMISSION_HIDDEN = "hd"
 HAP_PERMISSION_NOTIFY = "ev"
 HAP_PERMISSION_READ = "pr"
+HAP_PERMISSION_TIMED_WRITE = "tw"
 HAP_PERMISSION_WRITE = "pw"
 HAP_PERMISSION_WRITE_RESPONSE = "wr"
 

tests/test_accessory_driver.py

@@ -1,7 +1,6 @@
 """Tests for pyhap.accessory_driver."""
 
 import asyncio
-from concurrent.futures import ThreadPoolExecutor
 from functools import partial
 import os
 import tempfile
@@ -1280,24 +1279,33 @@ def fail_callback(*_: Any) -> NoReturn:
     }
 
 
-def test_start_from_sync(driver: AccessoryDriver) -> None:
-    """Start from sync."""
+def test_start_from_sync(async_zeroconf: Any) -> None:
+    """Start from sync, on the loop and executor the driver makes itself.
 
-    class Acc(Accessory):
-        @Accessory.run_at_interval(0)
-        async def run(self) -> None:
-            driver.executor = ThreadPoolExecutor()
-            driver.loop.set_default_executor(driver.executor)
-            await driver.async_stop()
+    Not the shared test loop: stopping shuts the executor down, which would
+    leave that loop unable to run later tests' executor jobs.
+    """
+    with (
+        patch("pyhap.accessory_driver.HAPServer.async_stop", new_callable=MagicMock),
+        patch("pyhap.accessory_driver.HAPServer.async_start", new_callable=AsyncMock),
+        patch("pyhap.accessory_driver.AccessoryDriver.persist"),
+    ):
+        driver = AccessoryDriver()
+
+        class Acc(Accessory):
+            @Accessory.run_at_interval(0)
+            async def run(self) -> None:
+                await driver.async_stop()
 
-        def setup_message(self) -> None:
-            pass
+            def setup_message(self) -> None:
+                pass
 
-    acc = Acc(driver, "TestAcc")
-    driver.add_accessory(acc)
-    with patch.object(driver.loop, "close"):
+        acc = Acc(driver, "TestAcc")
+        driver.add_accessory(acc)
         driver.start()
 
+    assert driver.loop.is_closed()
+
 
 def test_accessory_level_callbacks(driver: AccessoryDriver) -> None:
     bridge = Bridge(driver, "mybridge")
@@ -2084,7 +2092,7 @@ def test_prepare_bounds_prepared_writes(driver: AccessoryDriver) -> None:
     # An absurd TTL is clamped to the maximum.
     driver.prepare({HAP_REPR_TTL: 10_000_000_000, HAP_REPR_PID: 1}, client_addr)
     assert (
-        driver.prepared_writes[client_addr][1] - time.time()
+        driver.prepared_writes[client_addr][1] - time.monotonic()
         <= MAX_PREPARED_WRITE_TTL_SECONDS
     )