Enforce admin-only characteristic access

pyhap/accessory_driver.py

@@ -32,13 +32,16 @@
 import threading
 import time
 from typing import Any, NamedTuple, TypeGuard
+from uuid import UUID
 
 from zeroconf import ServiceInfo
 from zeroconf.asyncio import AsyncZeroconf
 
 from pyhap import util
 from pyhap.accessory import Accessory, Bridge, get_topic
 from pyhap.characteristic import (
+    ADMIN_ONLY_WRITE,
+    PROP_ADMIN_ONLY_ACCESS,
     PROP_PERMISSIONS,
     Characteristic,
     CharacteristicError,
@@ -1111,7 +1114,11 @@ def derive_session_key(
         return connection.derive_key(salt, info)
 
     def set_characteristics(
-        self, chars_query: dict[str, Any], client_addr: tuple[str, int]
+        self,
+        chars_query: dict[str, Any],
+        client_addr: tuple[str, int],
+        *,
+        client_uuid: UUID | None = None,
     ) -> dict[str, Any] | None:
         """Called from ``HAPServerHandler`` when iOS configures the characteristics.
 
@@ -1142,8 +1149,15 @@ def set_characteristics(
            }
 
         :type chars_query: dict
+
+        :param client_uuid: The paired controller that sent the request. Writes
+            to characteristics restricted to admins are refused with -70401
+            unless it is an admin.
+        :type client_uuid: uuid.UUID or None
         """
-        results, pending = self._apply_characteristic_writes(chars_query, client_addr)
+        results, pending = self._apply_characteristic_writes(
+            chars_query, client_addr, client_uuid
+        )
         for write in pending:
             self.add_job(self._finish_pending_write(write, client_addr))
         return _set_characteristics_response(results)
@@ -1170,6 +1184,8 @@ def _set_characteristics_for_controller(
         chars_query: dict[str, Any],
         client_addr: tuple[str, int],
         timeout: float,
+        *,
+        client_uuid: UUID | None = None,
     ) -> dict[str, Any] | None | Awaitable[dict[str, Any] | None]:
         """Apply a controller's write, waiting for coroutine setters.
 
@@ -1179,20 +1195,27 @@ def _set_characteristics_for_controller(
         cancelled and its characteristic answers -70408. Cancelling the
         awaitable, e.g. because the connection closed, cancels the setters.
         """
-        results, pending = self._apply_characteristic_writes(chars_query, client_addr)
+        results, pending = self._apply_characteristic_writes(
+            chars_query, client_addr, client_uuid
+        )
         if pending:
             return self._async_finish_set_characteristics(
                 results, pending, client_addr, timeout
             )
         return _set_characteristics_response(results)
 
     def _apply_characteristic_writes(
-        self, chars_query: dict[str, Any], client_addr: tuple[str, int]
+        self,
+        chars_query: dict[str, Any],
+        client_addr: tuple[str, int],
+        client_uuid: UUID | None,
     ) -> tuple[dict[int, dict[int, dict[str, Any]]], list[_PendingWrite]]:
         """Run the setters of a write and collect the per-characteristic results."""
         pending: list[_PendingWrite] = []
         try:
-            return self._run_characteristic_writes(chars_query, client_addr, pending)
+            return self._run_characteristic_writes(
+                chars_query, client_addr, client_uuid, pending
+            )
         except BaseException:
             # A malformed later entry must not leave a setter coroutine
             # un-awaited.
@@ -1204,6 +1227,7 @@ def _run_characteristic_writes(
         self,
         chars_query: dict[str, Any],
         client_addr: tuple[str, int],
+        client_uuid: UUID | None,
         pending: list[_PendingWrite],
     ) -> tuple[dict[int, dict[int, dict[str, Any]]], list[_PendingWrite]]:
         queries: list[dict[str, Any]] = chars_query[HAP_REPR_CHARS]
@@ -1240,6 +1264,7 @@ def _run_characteristic_writes(
                 HAP_REPR_STATUS: HAP_SERVER_STATUS.NOTIFICATION_NOT_SUPPORTED
             }
 
+        is_admin = self.state.is_admin(client_uuid)
         for query in queries:
             aid = query[HAP_REPR_AID]
             iid = query[HAP_REPR_IID]
@@ -1266,6 +1291,17 @@ def _run_characteristic_writes(
             if char is None:
                 continue
 
+            # Privileges are checked before the permissions, so a controller
+            # that may not write the characteristic learns nothing else.
+            # Also for an explicit null, so no setter sees the characteristic.
+            if not is_admin and ADMIN_ONLY_WRITE in char.properties.get(
+                PROP_ADMIN_ONLY_ACCESS, ()
+            ):
+                results[aid][iid] = {
+                    HAP_REPR_STATUS: HAP_SERVER_STATUS.INSUFFICIENT_PRIVILEGES
+                }
+                continue
+
             if (
                 value is not None
                 and HAP_PERMISSION_WRITE not in char.properties[PROP_PERMISSIONS]

pyhap/characteristic.py

@@ -98,6 +98,11 @@ def _setter_wants_client_addr(setter: Callable[..., Any]) -> bool:
 HAP_UNIT_SECONDS = "seconds"
 
 # ### Properties ###
+# Accesses only admin controllers may use. Not part of the HAP representation.
+# The accessory driver enforces "write" only. Reads, the accessory database
+# and events are not restricted.
+PROP_ADMIN_ONLY_ACCESS = "AdminOnlyAccess"
+ADMIN_ONLY_WRITE = "write"
 PROP_FORMAT = "Format"
 PROP_MAX_VALUE = "maxValue"
 PROP_MIN_STEP = "minStep"

pyhap/hap_handler.py

@@ -716,6 +716,7 @@ def handle_set_characteristics(self) -> None:
             requested_chars,
             cast("tuple[str, int]", self.client_address),
             RESPONSE_TIMEOUT,
+            client_uuid=self.client_uuid,
         )
         assert self.response is not None  # noqa: S101
         if inspect.isawaitable(result):

pyhap/state.py

@@ -69,7 +69,7 @@ def paired(self) -> bool:
         """Return if main accessory is currently paired."""
         return len(self.paired_clients) > 0
 
-    def is_admin(self, client_uuid: UUID) -> bool:
+    def is_admin(self, client_uuid: UUID | None) -> bool:
         """Check if a paired client is an admin."""
         if client_uuid not in self.client_properties:
             return False

scripts/gen_hap_types.py

@@ -206,8 +206,8 @@ def perms_from_bitmask(
     `properties` is the integer bitmask; `hap_properties` is the HAP.Properties
     legend (name -> {"Position": n}) from the same file.
     Admin-only characteristics (the "adminOnly" bit) are recorded separately in
-    "AdminOnlyAccess": ["write"] in the output, mirroring HAP-NodeJS; HAP-python
-    does not enforce this, it is informational.
+    "AdminOnlyAccess": ["write"] in the output, mirroring HAP-NodeJS. The
+    AccessoryDriver refuses such writes from controllers that are no admins.
     """
     perms: list[str] = []
     admin_only = False