From 2e12ebcf9a98e5d2c5ae74b10ae0b45d786d396d Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:48:29 +0100 Subject: [PATCH 1/7] Add HKSV 17.99 camera services and characteristics to the catalog Add the eight services and 27 characteristics of Apple's HomeKit Secure Video Open Source Compatibility Guide (Developer Preview 1.0) to the loader resources. Two of them, Contributing Sensors and Motion Enabled, also become optional characteristics of the Motion Sensor service. Timed write is advertised where the guide requires it. Writes to Motion Enabled and Streaming Enabled are marked admin only with an "AdminOnlyAccess": ["write"] catalog property. It is not part of the HAP representation and not enforced yet. Camera Recording Management keeps the characteristics classic recording relies on. --- pyhap/resources/characteristics.json | 235 +++++++++++++++++++++++++++ pyhap/resources/services.json | 94 ++++++++++- tests/test_loader.py | 166 ++++++++++++++++++- 3 files changed, 493 insertions(+), 2 deletions(-) diff --git a/pyhap/resources/characteristics.json b/pyhap/resources/characteristics.json index 596e5898..f185818d 100644 --- a/pyhap/resources/characteristics.json +++ b/pyhap/resources/characteristics.json @@ -133,6 +133,88 @@ "minValue": 0, "unit": "percentage" }, + "BufferActivityCommand": { + "Format": "tlv8", + "Permissions": [ + "pw" + ], + "UUID": "00008017-0000-1000-8000-0026BB765291" + }, + "BufferEventCommand": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008014-0000-1000-8000-0026BB765291" + }, + "BufferEventSequenceNumber": { + "Format": "uint32", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008015-0000-1000-8000-0026BB765291" + }, + "BufferUploadCommand": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008013-0000-1000-8000-0026BB765291" + }, + "CameraCapabilities": { + "Format": "tlv8", + "Permissions": [ + "pr" + ], + "UUID": "00008011-0000-1000-8000-0026BB765291" + }, + "CameraClientCSR": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008081-0000-1000-8000-0026BB765291" + }, + "CameraClientCertificate": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "tw" + ], + "UUID": "00008082-0000-1000-8000-0026BB765291" + }, + "CameraClientCertificateStatus": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008083-0000-1000-8000-0026BB765291" + }, + "CameraKey": { + "Format": "tlv8", + "Permissions": [ + "pw", + "tw" + ], + "UUID": "00008051-0000-1000-8000-0026BB765291" + }, + "CameraKeyID": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008052-0000-1000-8000-0026BB765291" + }, "CameraOperatingModeIndicator": { "Format": "bool", "Permissions": [ @@ -142,6 +224,22 @@ ], "UUID": "0000021D-0000-1000-8000-0026BB765291" }, + "CameraRecordingPublishingPoint": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw" + ], + "UUID": "00008016-0000-1000-8000-0026BB765291" + }, + "CameraZones": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw" + ], + "UUID": "00008022-0000-1000-8000-0026BB765291" + }, "CarbonDioxideDetected": { "Format": "uint8", "Permissions": [ @@ -265,6 +363,14 @@ "ContactNotDetected": 1 } }, + "ContributingSensors": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008086-0000-1000-8000-0026BB765291" + }, "CoolingThresholdTemperature": { "Format": "float", "Permissions": [ @@ -825,6 +931,19 @@ ], "UUID": "00000022-0000-1000-8000-0026BB765291" }, + "MotionEnabled": { + "AdminOnlyAccess": [ + "write" + ], + "Format": "bool", + "Permissions": [ + "pr", + "pw", + "ev", + "tw" + ], + "UUID": "00008087-0000-1000-8000-0026BB765291" + }, "Mute": { "Format": "bool", "Permissions": [ @@ -1055,6 +1174,15 @@ "SinglePress": 0 } }, + "RTPStreamingControl": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008045-0000-1000-8000-0026BB765291" + }, "RecordingAudioActive": { "Format": "uint8", "Permissions": [ @@ -1237,6 +1365,13 @@ ], "UUID": "00000117-0000-1000-8000-0026BB765291" }, + "SensorUUID": { + "Format": "data", + "Permissions": [ + "pr" + ], + "UUID": "0000805B-0000-1000-8000-0026BB765291" + }, "SerialNumber": { "Format": "string", "MaximumLength": 64, @@ -1386,6 +1521,19 @@ "Tampered": 1 } }, + "StreamingEnabled": { + "AdminOnlyAccess": [ + "write" + ], + "Format": "bool", + "Permissions": [ + "pr", + "pw", + "ev", + "tw" + ], + "UUID": "00008041-0000-1000-8000-0026BB765291" + }, "StreamingStatus": { "Format": "tlv8", "Permissions": [ @@ -1420,6 +1568,14 @@ ], "UUID": "00000115-0000-1000-8000-0026BB765291" }, + "SupportedAudioStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008044-0000-1000-8000-0026BB765291" + }, "SupportedCameraRecordingConfiguration": { "Format": "tlv8", "Permissions": [ @@ -1464,6 +1620,14 @@ ], "UUID": "00000144-0000-1000-8000-0026BB765291" }, + "SupportedVideoStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008043-0000-1000-8000-0026BB765291" + }, "SwingMode": { "Format": "uint8", "Permissions": [ @@ -1804,6 +1968,77 @@ "minValue": 0, "unit": "percentage" }, + "WebRTCNumberOfActiveSessions": { + "Format": "uint8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008057-0000-1000-8000-0026BB765291", + "maxValue": 255, + "minValue": 0 + }, + "WebRTCProvideAnswer": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008054-0000-1000-8000-0026BB765291" + }, + "WebRTCReoffer": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008058-0000-1000-8000-0026BB765291" + }, + "WebRTCSolicitOffer": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008053-0000-1000-8000-0026BB765291" + }, + "WebRTCStreamingControl": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00008056-0000-1000-8000-0026BB765291" + }, + "WebRTCSupportedAudioStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "0000805A-0000-1000-8000-0026BB765291" + }, + "WebRTCSupportedVideoStreamTiers": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00008059-0000-1000-8000-0026BB765291" + }, + "WebRTCUpdateSession": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "0000805C-0000-1000-8000-0026BB765291" + }, "HardwareFinish": { "Format": "tlv8", "Permissions": [ diff --git a/pyhap/resources/services.json b/pyhap/resources/services.json index 98c95aa4..8b3b4e97 100644 --- a/pyhap/resources/services.json +++ b/pyhap/resources/services.json @@ -61,6 +61,80 @@ ], "UUID": "00000096-0000-1000-8000-0026BB765291" }, + "CameraBufferManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "BufferUploadCommand", + "BufferActivityCommand", + "BufferEventCommand", + "BufferEventSequenceNumber", + "CameraRecordingPublishingPoint" + ], + "UUID": "00008000-0000-1000-8000-0026BB765291" + }, + "CameraCapabilities": { + "OptionalCharacteristics": [ + "ManuallyDisabled" + ], + "RequiredCharacteristics": [ + "Version", + "CameraCapabilities" + ], + "UUID": "00008010-0000-1000-8000-0026BB765291" + }, + "CameraClientCertificateManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "CameraClientCSR", + "CameraClientCertificate", + "CameraClientCertificateStatus" + ], + "UUID": "00008080-0000-1000-8000-0026BB765291" + }, + "CameraGlobalOperatingMode": { + "OptionalCharacteristics": [ + "ManuallyDisabled", + "NightVision", + "ThirdPartyCameraActive" + ], + "RequiredCharacteristics": [ + "HomeKitCameraActive", + "StreamingEnabled", + "CameraOperatingModeIndicator" + ], + "UUID": "00008032-0000-1000-8000-0026BB765291" + }, + "CameraKeyManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "CameraKey", + "CameraKeyID" + ], + "UUID": "00008050-0000-1000-8000-0026BB765291" + }, + "CameraMotionZones": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "Version", + "Active", + "CameraZones" + ], + "UUID": "00008021-0000-1000-8000-0026BB765291" + }, + "CameraMultiTierRTPStreamManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "StreamingEnabled", + "StatusActive", + "SupportedVideoStreamTiers", + "SupportedAudioStreamTiers", + "SupportedRTPConfiguration", + "SetupEndpoints", + "RTPStreamingControl", + "SensorUUID" + ], + "UUID": "00008031-0000-1000-8000-0026BB765291" + }, "CameraOperatingMode": { "OptionalCharacteristics": [ "CameraOperatingModeIndicator", @@ -103,6 +177,22 @@ ], "UUID": "00000204-0000-1000-8000-0026BB765291" }, + "CameraWebRTCStreamManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "WebRTCSolicitOffer", + "WebRTCProvideAnswer", + "WebRTCStreamingControl", + "WebRTCNumberOfActiveSessions", + "WebRTCReoffer", + "WebRTCUpdateSession", + "WebRTCSupportedVideoStreamTiers", + "WebRTCSupportedAudioStreamTiers", + "StreamingEnabled", + "SensorUUID" + ], + "UUID": "00008033-0000-1000-8000-0026BB765291" + }, "CarbonDioxideSensor": { "OptionalCharacteristics": [ "StatusActive", @@ -400,7 +490,9 @@ "StatusFault", "StatusTampered", "StatusLowBattery", - "Name" + "Name", + "ContributingSensors", + "MotionEnabled" ], "RequiredCharacteristics": [ "MotionDetected" diff --git a/tests/test_loader.py b/tests/test_loader.py index afe640a1..8714f65c 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -1,9 +1,18 @@ """Tests for pyhap.loader.""" +from unittest.mock import patch +from uuid import UUID + import pytest from pyhap import CHARACTERISTICS_FILE, SERVICES_FILE -from pyhap.characteristic import Characteristic +from pyhap.characteristic import ( + PROP_FORMAT, + PROP_MAX_VALUE, + PROP_MIN_VALUE, + PROP_PERMISSIONS, + Characteristic, +) from pyhap.loader import Loader, get_loader from pyhap.service import Service @@ -74,3 +83,158 @@ def test_get_loader() -> None: assert loader.serv_types == loader2.serv_types assert get_loader() == loader + + +# Accesses only admin controllers may use, not part of the HAP representation. +ADMIN_ONLY_ACCESS = "AdminOnlyAccess" + +# Camera services and characteristics of the HomeKit Secure Video 17.99 guide: +# name -> (short UUID, format, permissions, AdminOnlyAccess). +HKSV3_CHARS = { + "SensorUUID": ("805B", "data", {"pr"}, None), + "MotionEnabled": ("8087", "bool", {"pr", "pw", "ev", "tw"}, ["write"]), + "SupportedVideoStreamTiers": ("8043", "tlv8", {"pr", "ev"}, None), + "SupportedAudioStreamTiers": ("8044", "tlv8", {"pr", "ev"}, None), + "CameraCapabilities": ("8011", "tlv8", {"pr"}, None), + "ContributingSensors": ("8086", "tlv8", {"pr", "ev"}, None), + "CameraKey": ("8051", "tlv8", {"pw", "tw"}, None), + "CameraKeyID": ("8052", "tlv8", {"pr", "ev"}, None), + "BufferUploadCommand": ("8013", "tlv8", {"pr", "pw", "wr"}, None), + "BufferActivityCommand": ("8017", "tlv8", {"pw"}, None), + "BufferEventCommand": ("8014", "tlv8", {"pr", "pw", "wr"}, None), + "BufferEventSequenceNumber": ("8015", "uint32", {"pr", "ev"}, None), + "CameraRecordingPublishingPoint": ("8016", "tlv8", {"pr", "pw"}, None), + "CameraZones": ("8022", "tlv8", {"pr", "pw"}, None), + "StreamingEnabled": ("8041", "bool", {"pr", "pw", "ev", "tw"}, ["write"]), + "RTPStreamingControl": ("8045", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCSolicitOffer": ("8053", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCProvideAnswer": ("8054", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCStreamingControl": ("8056", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCNumberOfActiveSessions": ("8057", "uint8", {"pr", "ev"}, None), + "WebRTCReoffer": ("8058", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCUpdateSession": ("805C", "tlv8", {"pr", "pw", "wr"}, None), + "WebRTCSupportedVideoStreamTiers": ("8059", "tlv8", {"pr", "ev"}, None), + "WebRTCSupportedAudioStreamTiers": ("805A", "tlv8", {"pr", "ev"}, None), + "CameraClientCSR": ("8081", "tlv8", {"pr", "pw", "wr"}, None), + "CameraClientCertificate": ("8082", "tlv8", {"pr", "pw", "tw"}, None), + "CameraClientCertificateStatus": ("8083", "tlv8", {"pr", "ev"}, None), +} + +# name -> (short UUID, required, optional) +HKSV3_SERVICES = { + "CameraCapabilities": ( + "8010", + ["Version", "CameraCapabilities"], + ["ManuallyDisabled"], + ), + "CameraGlobalOperatingMode": ( + "8032", + ["HomeKitCameraActive", "StreamingEnabled", "CameraOperatingModeIndicator"], + ["ManuallyDisabled", "NightVision", "ThirdPartyCameraActive"], + ), + "CameraMotionZones": ("8021", ["Version", "Active", "CameraZones"], []), + "CameraBufferManagement": ( + "8000", + [ + "BufferUploadCommand", + "BufferActivityCommand", + "BufferEventCommand", + "BufferEventSequenceNumber", + "CameraRecordingPublishingPoint", + ], + [], + ), + "CameraMultiTierRTPStreamManagement": ( + "8031", + [ + "StreamingEnabled", + "StatusActive", + "SupportedVideoStreamTiers", + "SupportedAudioStreamTiers", + "SupportedRTPConfiguration", + "SetupEndpoints", + "RTPStreamingControl", + "SensorUUID", + ], + [], + ), + "CameraWebRTCStreamManagement": ( + "8033", + [ + "WebRTCSolicitOffer", + "WebRTCProvideAnswer", + "WebRTCStreamingControl", + "WebRTCNumberOfActiveSessions", + "WebRTCReoffer", + "WebRTCUpdateSession", + "WebRTCSupportedVideoStreamTiers", + "WebRTCSupportedAudioStreamTiers", + "StreamingEnabled", + "SensorUUID", + ], + [], + ), + "CameraKeyManagement": ("8050", ["CameraKey", "CameraKeyID"], []), + "CameraClientCertificateManagement": ( + "8080", + [ + "CameraClientCSR", + "CameraClientCertificate", + "CameraClientCertificateStatus", + ], + [], + ), +} + + +def _hap_uuid(short: str) -> UUID: + return UUID(f"0000{short}-0000-1000-8000-0026BB765291") + + +@pytest.mark.parametrize("name", HKSV3_CHARS) +def test_loader_hksv3_char(name: str) -> None: + """Test the HKSV 17.99 characteristics match the guide.""" + short, char_format, permissions, admin_only_access = HKSV3_CHARS[name] + char = Loader().get_char(name) + assert char.type_id == _hap_uuid(short) + assert char.properties[PROP_FORMAT] == char_format + assert set(char.properties[PROP_PERMISSIONS]) == permissions + assert char.properties.get(ADMIN_ONLY_ACCESS) == admin_only_access + + +def test_loader_hksv3_admin_only_not_advertised() -> None: + """Test the admin only marker stays out of the HAP representation.""" + char = Loader().get_char("StreamingEnabled") + with patch.object(char, "broker") as mock_broker: + mock_broker.iid_manager.get_iid.return_value = 2 + hap_repr = char.to_HAP() + assert hap_repr["perms"] == ["pr", "pw", "ev", "tw"] + assert ADMIN_ONLY_ACCESS not in hap_repr + + +def test_loader_hksv3_active_sessions_range() -> None: + """Test the WebRTC session count carries the guide's value range.""" + char = Loader().get_char("WebRTCNumberOfActiveSessions") + assert char.properties[PROP_MIN_VALUE] == 0 + assert char.properties[PROP_MAX_VALUE] == 255 + + +@pytest.mark.parametrize("name", HKSV3_SERVICES) +def test_loader_hksv3_service(name: str) -> None: + """Test the HKSV 17.99 services match the guide.""" + short, required, optional = HKSV3_SERVICES[name] + loader = Loader() + service = loader.get_service(name) + assert service.type_id == _hap_uuid(short) + assert [char.display_name for char in service.characteristics] == required + assert loader.serv_types[name]["OptionalCharacteristics"] == optional + assert set(optional) <= set(loader.char_types) + + +def test_loader_motion_sensor_hksv3_optional() -> None: + """Test the motion sensor gains the HKSV 17.99 chars only as optional.""" + loader = Loader() + optional = loader.serv_types["MotionSensor"]["OptionalCharacteristics"] + assert {"ContributingSensors", "MotionEnabled"} <= set(optional) + service = loader.get_service("MotionSensor") + assert [char.display_name for char in service.characteristics] == ["MotionDetected"] From 207041ce2e7017ff31e3a965a09578e0ae929ce7 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 15:48:49 +0100 Subject: [PATCH 2/7] Fix the camera operating mode characteristic permissions Third Party Camera Active is state the camera reports, so controllers read and subscribe to it but do not write it. Night Vision and Camera Operating Mode Indicator are written with timed writes. The Camera Global Operating Mode service of HKSV 17.99 uses all three. This matches the definitions in HAP-NodeJS. --- pyhap/resources/characteristics.json | 7 ++++--- tests/test_loader.py | 16 ++++++++++++++++ 2 files changed, 20 insertions(+), 3 deletions(-) diff --git a/pyhap/resources/characteristics.json b/pyhap/resources/characteristics.json index f185818d..639b04ff 100644 --- a/pyhap/resources/characteristics.json +++ b/pyhap/resources/characteristics.json @@ -220,7 +220,8 @@ "Permissions": [ "pr", "pw", - "ev" + "ev", + "tw" ], "UUID": "0000021D-0000-1000-8000-0026BB765291" }, @@ -965,7 +966,8 @@ "Permissions": [ "pr", "pw", - "ev" + "ev", + "tw" ], "UUID": "0000011B-0000-1000-8000-0026BB765291" }, @@ -1872,7 +1874,6 @@ "Format": "bool", "Permissions": [ "pr", - "pw", "ev" ], "UUID": "0000021C-0000-1000-8000-0026BB765291" diff --git a/tests/test_loader.py b/tests/test_loader.py index 8714f65c..a38f1426 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -238,3 +238,19 @@ def test_loader_motion_sensor_hksv3_optional() -> None: assert {"ContributingSensors", "MotionEnabled"} <= set(optional) service = loader.get_service("MotionSensor") assert [char.display_name for char in service.characteristics] == ["MotionDetected"] + + +@pytest.mark.parametrize( + ("name", "permissions"), + [ + ("CameraOperatingModeIndicator", {"pr", "pw", "ev", "tw"}), + ("NightVision", {"pr", "pw", "ev", "tw"}), + ("ThirdPartyCameraActive", {"pr", "ev"}), + ], +) +def test_loader_camera_operating_mode_permissions( + name: str, permissions: set[str] +) -> None: + """Test the operating mode settings shared with the HKSV 17.99 services.""" + char = Loader().get_char(name) + assert set(char.properties[PROP_PERMISSIONS]) == permissions From 7b5d09204d6e88a8eae81f51317d917832b0979b Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sun, 8 Nov 2026 19:32:58 +0100 Subject: [PATCH 3/7] Enforce admin-only characteristic access The catalog can mark a characteristic with "AdminOnlyAccess" (Motion Enabled and Streaming Enabled list "write"), but nothing checked it. Pass the controller of the session from the HAP handler to the driver and refuse a write, or a read when the list contains "read", with -70401 unless the controller is a paired admin. A request without a known controller is refused too. This follows adminOnlyAccess in HAP-NodeJS. Characteristics without the property are not affected. --- pyhap/accessory_driver.py | 56 ++++++- pyhap/characteristic.py | 5 + pyhap/hap_handler.py | 6 +- tests/test_admin_only.py | 298 ++++++++++++++++++++++++++++++++++++++ 4 files changed, 359 insertions(+), 6 deletions(-) create mode 100644 tests/test_admin_only.py diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index a68eae6c..668bcd3e 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -30,13 +30,20 @@ import threading import time from typing import Any +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 Characteristic, CharacteristicError +from pyhap.characteristic import ( + ADMIN_ONLY_READ, + ADMIN_ONLY_WRITE, + PROP_ADMIN_ONLY_ACCESS, + Characteristic, + CharacteristicError, +) from pyhap.const import ( HAP_PERMISSION_NOTIFY, HAP_PROTOCOL_SHORT_VERSION, @@ -85,6 +92,11 @@ MAX_PREPARED_WRITE_TTL_SECONDS = 30.0 +def _admin_only(char: Characteristic, access: str) -> bool: + """Return whether ``access`` to ``char`` is restricted to admin controllers.""" + return access in char.properties.get(PROP_ADMIN_ONLY_ACCESS, ()) + + def _wrap_char_setter(char: Any, value: Any, client_addr: Any) -> tuple[int, Any]: """Process an characteristic setter callback trapping and logging all exceptions.""" try: @@ -905,12 +917,19 @@ def get_accessories(self, include_value: bool = True) -> dict[str, Any]: logger.debug("Get accessories response: %s", hap_rep) return {HAP_REPR_ACCS: hap_rep} - def get_characteristics(self, char_ids: Any) -> dict[str, Any]: + def get_characteristics( + self, char_ids: Any, client_uuid: UUID | None = None + ) -> dict[str, Any]: """Returns values for the required characteristics. :param char_ids: A list of characteristic "paths", e.g. "1.2" is aid 1, iid 2. :type char_ids: list + :param client_uuid: The paired controller that sent the request. Reads + of characteristics restricted to admins are refused with -70401 + unless it is an admin. + :type client_uuid: uuid.UUID + :return: Status success for each required characteristic. For example: .. code-block:: python @@ -926,6 +945,7 @@ def get_characteristics(self, char_ids: Any) -> dict[str, Any]: :rtype: dict """ assert self.accessory is not None # noqa: S101 + is_admin = self._is_admin(client_uuid) chars = [] for aid_iid in char_ids: aid, iid = (int(i) for i in aid_iid.split(".")) @@ -956,7 +976,13 @@ def get_characteristics(self, char_ids: Any) -> dict[str, Any]: # A characteristic IID may resolve to a Service or be unknown; # only a Characteristic can report a value. - if available and isinstance(char, Characteristic): + if ( + isinstance(char, Characteristic) + and not is_admin + and _admin_only(char, ADMIN_ONLY_READ) + ): + rep[HAP_REPR_STATUS] = HAP_SERVER_STATUS.INSUFFICIENT_PRIVILEGES + elif available and isinstance(char, Characteristic): rep[HAP_REPR_VALUE] = char.get_value() rep[HAP_REPR_STATUS] = HAP_SERVER_STATUS.SUCCESS except CharacteristicError: @@ -994,7 +1020,10 @@ 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. @@ -1013,6 +1042,11 @@ 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 """ # TODO: Add support for chars that do no support notifications. @@ -1062,6 +1096,16 @@ def set_characteristics( if char is None: continue + if ( + value is not None + and _admin_only(char, ADMIN_ONLY_WRITE) + and not self._is_admin(client_uuid) + ): + results[aid][iid] = { + HAP_REPR_STATUS: HAP_SERVER_STATUS.INSUFFICIENT_PRIVILEGES + } + continue + set_result = HAP_SERVER_STATUS.INVALID_VALUE_IN_REQUEST set_result_value = None @@ -1120,6 +1164,10 @@ def set_characteristics( return {HAP_REPR_CHARS: characteristics} if nonempty_results_exist else None + def _is_admin(self, client_uuid: UUID | None) -> bool: + """Return whether ``client_uuid`` is a paired admin controller.""" + return client_uuid is not None and self.state.is_admin(client_uuid) + def prepare( self, prepare_query: dict[str, Any], client_addr: tuple[str, int] ) -> dict[str, int]: diff --git a/pyhap/characteristic.py b/pyhap/characteristic.py index bcbdd3e2..5788135f 100644 --- a/pyhap/characteristic.py +++ b/pyhap/characteristic.py @@ -95,6 +95,11 @@ def _setter_wants_client_addr(setter: Callable[..., Any]) -> bool: HAP_UNIT_SECONDS = "seconds" # ### Properties ### +# Accesses ("read", "write") only admin controllers may use. Not part of the +# HAP representation, the accessory driver enforces it. +PROP_ADMIN_ONLY_ACCESS = "AdminOnlyAccess" +ADMIN_ONLY_READ = "read" +ADMIN_ONLY_WRITE = "write" PROP_FORMAT = "Format" PROP_MAX_VALUE = "maxValue" PROP_MIN_STEP = "minStep" diff --git a/pyhap/hap_handler.py b/pyhap/hap_handler.py index 83a47380..ec702f94 100644 --- a/pyhap/hap_handler.py +++ b/pyhap/hap_handler.py @@ -679,7 +679,7 @@ def handle_get_characteristics(self) -> None: assert self.parsed_url is not None # noqa: S101 params = parse_qs(self.parsed_url.query) response = self.accessory_handler.get_characteristics( - params["id"][0].split(",") + params["id"][0].split(","), self.client_uuid ) chars = response[HAP_REPR_CHARS] @@ -712,7 +712,9 @@ def handle_set_characteristics(self) -> None: ) response = self.accessory_handler.set_characteristics( - requested_chars, cast("tuple[str, int]", self.client_address) + requested_chars, + cast("tuple[str, int]", self.client_address), + self.client_uuid, ) if response is None: self.send_response(HTTPStatus.NO_CONTENT) diff --git a/tests/test_admin_only.py b/tests/test_admin_only.py new file mode 100644 index 00000000..063bcd0a --- /dev/null +++ b/tests/test_admin_only.py @@ -0,0 +1,298 @@ +"""Tests for characteristics restricted to admin controllers.""" + +from collections.abc import Callable, Iterator +import io +import json +from typing import Any, NamedTuple +from unittest.mock import MagicMock, patch +from urllib.parse import urlparse +from uuid import UUID + +import pytest + +from pyhap import hap_handler +from pyhap.accessory import Accessory +from pyhap.accessory_driver import AccessoryDriver +from pyhap.characteristic import ( + HAP_FORMAT_BOOL, + HAP_PERMISSION_READ, + PROP_ADMIN_ONLY_ACCESS, + PROP_FORMAT, + PROP_PERMISSIONS, + Characteristic, +) +from pyhap.const import ( + HAP_PERMISSION_WRITE, + HAP_PERMISSIONS, + HAP_REPR_AID, + HAP_REPR_CHARS, + HAP_REPR_IID, + HAP_REPR_STATUS, + HAP_REPR_VALUE, + HAP_SERVER_STATUS, +) +from pyhap.encoder import AccessoryEncoder +from pyhap.service import Service + +CLIENT = ("1.2.3.4", 5000) +ADMIN = UUID("7d0d1ee9-46fe-4a56-a115-69df3f6860c1") +USER = UUID("7d0d1ee9-46fe-4a56-a115-69df3f6860c2") +UNPAIRED = UUID("7d0d1ee9-46fe-4a56-a115-69df3f6860c3") +OTHER_ADMIN = UUID("7d0d1ee9-46fe-4a56-a115-69df3f6860c4") +REFUSED = HAP_SERVER_STATUS.INSUFFICIENT_PRIVILEGES + + +class SensorChar(NamedTuple): + """A characteristic of the test accessory and its setter.""" + + setter: MagicMock + aid: int + iid: int + + +def _with_setter(acc: Accessory, service: Service, name: str) -> SensorChar: + """Give the char ``name`` of ``service`` a mock setter.""" + setter = MagicMock() + char = service.configure_char(name, setter_callback=setter) + iid = acc.iid_manager.get_iid(char) + assert acc.aid is not None + assert iid is not None + return SensorChar(setter, acc.aid, iid) + + +def _write(char: SensorChar) -> dict[str, Any]: + """Return a write of True to ``char``.""" + return { + HAP_REPR_CHARS: [ + {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_VALUE: True} + ] + } + + +def _response(char: SensorChar, status: int | None) -> dict[str, Any] | None: + """Return the expected write response, None for a plain success.""" + if status is None: + return None + return { + HAP_REPR_CHARS: [ + {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_STATUS: status} + ] + } + + +@pytest.fixture +def chars(driver: AccessoryDriver) -> Iterator[dict[str, SensorChar]]: + """Add a motion sensor and pair an admin and a regular controller. + + Motion Enabled may only be written by admins. Secret is a made-up + characteristic that only admins may read. Open is a made-up characteristic + anyone may read and write. + """ + acc = Accessory(driver, "Sensor") + service = driver.loader.get_service("MotionSensor") + service.add_characteristic( + driver.loader.get_char("MotionEnabled"), + Characteristic( + "Open", + UUID("2d7f7c38-1d4c-4e0b-9d0e-4c1f1d2c7a11"), + { + PROP_FORMAT: HAP_FORMAT_BOOL, + PROP_PERMISSIONS: [HAP_PERMISSION_READ, HAP_PERMISSION_WRITE], + }, + ), + Characteristic( + "Secret", + UUID("2d7f7c38-1d4c-4e0b-9d0e-4c1f1d2c7a10"), + { + PROP_FORMAT: HAP_FORMAT_BOOL, + PROP_PERMISSIONS: [HAP_PERMISSION_READ], + PROP_ADMIN_ONLY_ACCESS: ["read"], + }, + ), + ) + acc.add_service(service) + driver.add_accessory(acc) + driver.pair(str(ADMIN).encode(), b"admin-key", HAP_PERMISSIONS.ADMIN) + driver.pair(str(USER).encode(), b"user-key", HAP_PERMISSIONS.USER) + with patch.object(driver, "publish"), patch.object(driver, "async_persist"): + yield { + "MotionEnabled": _with_setter(acc, service, "MotionEnabled"), + "Open": _with_setter(acc, service, "Open"), + "Secret": _with_setter(acc, service, "Secret"), + } + + +@pytest.mark.parametrize( + ("name", "client_uuid", "status", "calls"), + [ + ("MotionEnabled", ADMIN, None, 1), + ("MotionEnabled", USER, REFUSED, 0), + ("MotionEnabled", UNPAIRED, REFUSED, 0), + ("MotionEnabled", None, REFUSED, 0), + ("Open", USER, None, 1), + ("Open", None, None, 1), + ], + ids=[ + "admin", + "user", + "unpaired", + "unknown_controller", + "unrestricted_user", + "unrestricted_unknown_controller", + ], +) +def test_admin_only_write( + driver: AccessoryDriver, + chars: dict[str, SensorChar], + name: str, + client_uuid: UUID | None, + status: int | None, + calls: int, +) -> None: + """Test only admins may write a characteristic restricted to them.""" + char = chars[name] + + response = driver.set_characteristics(_write(char), CLIENT, client_uuid) + + assert response == _response(char, status) + assert char.setter.call_count == calls + + +def _demote(driver: AccessoryDriver) -> None: + driver.pair(str(ADMIN).encode(), b"admin-key", HAP_PERMISSIONS.USER) + + +def _remove(driver: AccessoryDriver) -> None: + driver.unpair(ADMIN) + + +@pytest.mark.parametrize("change", [_demote, _remove], ids=["demoted", "removed"]) +def test_admin_change_applies_to_open_session( + driver: AccessoryDriver, + chars: dict[str, SensorChar], + change: Callable[[AccessoryDriver], None], +) -> None: + """Test a controller that loses admin is refused on its next write.""" + driver.pair(str(OTHER_ADMIN).encode(), b"other-key", HAP_PERMISSIONS.ADMIN) + char = chars["MotionEnabled"] + assert driver.set_characteristics(_write(char), CLIENT, ADMIN) is None + + change(driver) + + assert driver.set_characteristics(_write(char), CLIENT, ADMIN) == _response( + char, REFUSED + ) + char.setter.assert_called_once_with(True) + + +def test_legacy_state_pairs_everyone_as_admin( + driver: AccessoryDriver, chars: dict[str, SensorChar] +) -> None: + """Test controllers from a state file without permissions are admins.""" + stored = io.StringIO() + AccessoryEncoder.persist(stored, driver.state) + legacy = json.loads(stored.getvalue()) + del legacy["client_properties"] + AccessoryEncoder.load_into(io.StringIO(json.dumps(legacy)), driver.state) + char = chars["MotionEnabled"] + + assert driver.set_characteristics(_write(char), CLIENT, USER) is None + char.setter.assert_called_once_with(True) + + +@pytest.mark.parametrize( + ("name", "client_uuid", "expected"), + [ + ("Secret", ADMIN, {HAP_REPR_STATUS: 0, HAP_REPR_VALUE: False}), + ("Secret", USER, {HAP_REPR_STATUS: REFUSED}), + ("Secret", None, {HAP_REPR_STATUS: REFUSED}), + ("MotionEnabled", USER, {HAP_REPR_STATUS: 0, HAP_REPR_VALUE: False}), + ("MotionEnabled", None, {HAP_REPR_STATUS: 0, HAP_REPR_VALUE: False}), + ], + ids=[ + "admin", + "user", + "unknown_controller", + "write_restricted_user", + "write_restricted_unknown_controller", + ], +) +def test_admin_only_read( + driver: AccessoryDriver, + chars: dict[str, SensorChar], + name: str, + client_uuid: UUID | None, + expected: dict[str, Any], +) -> None: + """Test reads are only refused when the restriction lists read.""" + char = chars[name] + + response = driver.get_characteristics([f"{char.aid}.{char.iid}"], client_uuid) + + assert response == { + HAP_REPR_CHARS: [{HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, **expected}] + } + + +@pytest.mark.parametrize( + ("client_uuid", "status_code", "body"), + [ + (ADMIN, 204, b""), + (USER, 207, b'{"characteristics":[{"status":-70401,"aid":1,"iid":10}]}'), + ], + ids=["admin", "user"], +) +def test_handler_passes_the_controller_on_write( + driver: AccessoryDriver, + chars: dict[str, SensorChar], + client_uuid: UUID, + status_code: int, + body: bytes, +) -> None: + """Test the HAP handler writes as the controller of its session.""" + char = chars["MotionEnabled"] + handler = hap_handler.HAPServerHandler(driver, CLIENT) + handler.is_encrypted = True + handler.client_uuid = client_uuid + handler.response = hap_handler.HAPResponse() + handler.request_body = json.dumps( + { + HAP_REPR_CHARS: [ + {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_VALUE: True} + ] + } + ).encode() + + handler.handle_set_characteristics() + + assert handler.response.status_code == status_code + assert handler.response.body == body + + +@pytest.mark.parametrize( + ("client_uuid", "status_code", "body"), + [ + (ADMIN, 200, b'{"characteristics":[{"aid":1,"iid":12,"value":false}]}'), + (USER, 207, b'{"characteristics":[{"aid":1,"iid":12,"status":-70401}]}'), + ], + ids=["admin", "user"], +) +def test_handler_passes_the_controller_on_read( + driver: AccessoryDriver, + chars: dict[str, SensorChar], + client_uuid: UUID, + status_code: int, + body: bytes, +) -> None: + """Test the HAP handler reads as the controller of its session.""" + char = chars["Secret"] + handler = hap_handler.HAPServerHandler(driver, CLIENT) + handler.is_encrypted = True + handler.client_uuid = client_uuid + handler.response = hap_handler.HAPResponse() + handler.parsed_url = urlparse(f"/characteristics?id={char.aid}.{char.iid}") + + handler.handle_get_characteristics() + + assert handler.response.status_code == status_code + assert handler.response.body == body From 26642252eeb380ac57f040c95da16b418c7459a8 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:39:08 +0100 Subject: [PATCH 4/7] Pass client_uuid by keyword only #535 adds client_addr as the second argument of get_characteristics, so a positional client_uuid could end up as the address once both are in. Also note why privileges are checked before availability. --- pyhap/accessory_driver.py | 7 +++++-- pyhap/hap_handler.py | 4 ++-- tests/test_admin_only.py | 16 +++++++++------- 3 files changed, 16 insertions(+), 11 deletions(-) diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index 668bcd3e..efc7106a 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -918,7 +918,7 @@ def get_accessories(self, include_value: bool = True) -> dict[str, Any]: return {HAP_REPR_ACCS: hap_rep} def get_characteristics( - self, char_ids: Any, client_uuid: UUID | None = None + self, char_ids: Any, *, client_uuid: UUID | None = None ) -> dict[str, Any]: """Returns values for the required characteristics. @@ -975,7 +975,9 @@ def get_characteristics( available = False # A characteristic IID may resolve to a Service or be unknown; - # only a Characteristic can report a value. + # only a Characteristic can report a value. Privileges are + # checked before availability, so a controller that may not + # read the characteristic learns nothing about its accessory. if ( isinstance(char, Characteristic) and not is_admin @@ -1023,6 +1025,7 @@ def set_characteristics( 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. diff --git a/pyhap/hap_handler.py b/pyhap/hap_handler.py index ec702f94..96698353 100644 --- a/pyhap/hap_handler.py +++ b/pyhap/hap_handler.py @@ -679,7 +679,7 @@ def handle_get_characteristics(self) -> None: assert self.parsed_url is not None # noqa: S101 params = parse_qs(self.parsed_url.query) response = self.accessory_handler.get_characteristics( - params["id"][0].split(","), self.client_uuid + params["id"][0].split(","), client_uuid=self.client_uuid ) chars = response[HAP_REPR_CHARS] @@ -714,7 +714,7 @@ def handle_set_characteristics(self) -> None: response = self.accessory_handler.set_characteristics( requested_chars, cast("tuple[str, int]", self.client_address), - self.client_uuid, + client_uuid=self.client_uuid, ) if response is None: self.send_response(HTTPStatus.NO_CONTENT) diff --git a/tests/test_admin_only.py b/tests/test_admin_only.py index 063bcd0a..eca2e8bb 100644 --- a/tests/test_admin_only.py +++ b/tests/test_admin_only.py @@ -152,7 +152,7 @@ def test_admin_only_write( """Test only admins may write a characteristic restricted to them.""" char = chars[name] - response = driver.set_characteristics(_write(char), CLIENT, client_uuid) + response = driver.set_characteristics(_write(char), CLIENT, client_uuid=client_uuid) assert response == _response(char, status) assert char.setter.call_count == calls @@ -175,13 +175,13 @@ def test_admin_change_applies_to_open_session( """Test a controller that loses admin is refused on its next write.""" driver.pair(str(OTHER_ADMIN).encode(), b"other-key", HAP_PERMISSIONS.ADMIN) char = chars["MotionEnabled"] - assert driver.set_characteristics(_write(char), CLIENT, ADMIN) is None + assert driver.set_characteristics(_write(char), CLIENT, client_uuid=ADMIN) is None change(driver) - assert driver.set_characteristics(_write(char), CLIENT, ADMIN) == _response( - char, REFUSED - ) + assert driver.set_characteristics( + _write(char), CLIENT, client_uuid=ADMIN + ) == _response(char, REFUSED) char.setter.assert_called_once_with(True) @@ -196,7 +196,7 @@ def test_legacy_state_pairs_everyone_as_admin( AccessoryEncoder.load_into(io.StringIO(json.dumps(legacy)), driver.state) char = chars["MotionEnabled"] - assert driver.set_characteristics(_write(char), CLIENT, USER) is None + assert driver.set_characteristics(_write(char), CLIENT, client_uuid=USER) is None char.setter.assert_called_once_with(True) @@ -227,7 +227,9 @@ def test_admin_only_read( """Test reads are only refused when the restriction lists read.""" char = chars[name] - response = driver.get_characteristics([f"{char.aid}.{char.iid}"], client_uuid) + response = driver.get_characteristics( + [f"{char.aid}.{char.iid}"], client_uuid=client_uuid + ) assert response == { HAP_REPR_CHARS: [{HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, **expected}] From 1313b729269dc16914d82622192993b6f59e6a64 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 14:39:08 +0100 Subject: [PATCH 5/7] Keep only the admin-only test in the loader tests The catalog tests came with the old catalog commits and are not about admin-only access. --- tests/test_loader.py | 176 +------------------------------------------ 1 file changed, 4 insertions(+), 172 deletions(-) diff --git a/tests/test_loader.py b/tests/test_loader.py index a38f1426..62f57e64 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -1,18 +1,11 @@ """Tests for pyhap.loader.""" from unittest.mock import patch -from uuid import UUID import pytest from pyhap import CHARACTERISTICS_FILE, SERVICES_FILE -from pyhap.characteristic import ( - PROP_FORMAT, - PROP_MAX_VALUE, - PROP_MIN_VALUE, - PROP_PERMISSIONS, - Characteristic, -) +from pyhap.characteristic import PROP_ADMIN_ONLY_ACCESS, Characteristic from pyhap.loader import Loader, get_loader from pyhap.service import Service @@ -85,172 +78,11 @@ def test_get_loader() -> None: assert get_loader() == loader -# Accesses only admin controllers may use, not part of the HAP representation. -ADMIN_ONLY_ACCESS = "AdminOnlyAccess" - -# Camera services and characteristics of the HomeKit Secure Video 17.99 guide: -# name -> (short UUID, format, permissions, AdminOnlyAccess). -HKSV3_CHARS = { - "SensorUUID": ("805B", "data", {"pr"}, None), - "MotionEnabled": ("8087", "bool", {"pr", "pw", "ev", "tw"}, ["write"]), - "SupportedVideoStreamTiers": ("8043", "tlv8", {"pr", "ev"}, None), - "SupportedAudioStreamTiers": ("8044", "tlv8", {"pr", "ev"}, None), - "CameraCapabilities": ("8011", "tlv8", {"pr"}, None), - "ContributingSensors": ("8086", "tlv8", {"pr", "ev"}, None), - "CameraKey": ("8051", "tlv8", {"pw", "tw"}, None), - "CameraKeyID": ("8052", "tlv8", {"pr", "ev"}, None), - "BufferUploadCommand": ("8013", "tlv8", {"pr", "pw", "wr"}, None), - "BufferActivityCommand": ("8017", "tlv8", {"pw"}, None), - "BufferEventCommand": ("8014", "tlv8", {"pr", "pw", "wr"}, None), - "BufferEventSequenceNumber": ("8015", "uint32", {"pr", "ev"}, None), - "CameraRecordingPublishingPoint": ("8016", "tlv8", {"pr", "pw"}, None), - "CameraZones": ("8022", "tlv8", {"pr", "pw"}, None), - "StreamingEnabled": ("8041", "bool", {"pr", "pw", "ev", "tw"}, ["write"]), - "RTPStreamingControl": ("8045", "tlv8", {"pr", "pw", "wr"}, None), - "WebRTCSolicitOffer": ("8053", "tlv8", {"pr", "pw", "wr"}, None), - "WebRTCProvideAnswer": ("8054", "tlv8", {"pr", "pw", "wr"}, None), - "WebRTCStreamingControl": ("8056", "tlv8", {"pr", "pw", "wr"}, None), - "WebRTCNumberOfActiveSessions": ("8057", "uint8", {"pr", "ev"}, None), - "WebRTCReoffer": ("8058", "tlv8", {"pr", "pw", "wr"}, None), - "WebRTCUpdateSession": ("805C", "tlv8", {"pr", "pw", "wr"}, None), - "WebRTCSupportedVideoStreamTiers": ("8059", "tlv8", {"pr", "ev"}, None), - "WebRTCSupportedAudioStreamTiers": ("805A", "tlv8", {"pr", "ev"}, None), - "CameraClientCSR": ("8081", "tlv8", {"pr", "pw", "wr"}, None), - "CameraClientCertificate": ("8082", "tlv8", {"pr", "pw", "tw"}, None), - "CameraClientCertificateStatus": ("8083", "tlv8", {"pr", "ev"}, None), -} - -# name -> (short UUID, required, optional) -HKSV3_SERVICES = { - "CameraCapabilities": ( - "8010", - ["Version", "CameraCapabilities"], - ["ManuallyDisabled"], - ), - "CameraGlobalOperatingMode": ( - "8032", - ["HomeKitCameraActive", "StreamingEnabled", "CameraOperatingModeIndicator"], - ["ManuallyDisabled", "NightVision", "ThirdPartyCameraActive"], - ), - "CameraMotionZones": ("8021", ["Version", "Active", "CameraZones"], []), - "CameraBufferManagement": ( - "8000", - [ - "BufferUploadCommand", - "BufferActivityCommand", - "BufferEventCommand", - "BufferEventSequenceNumber", - "CameraRecordingPublishingPoint", - ], - [], - ), - "CameraMultiTierRTPStreamManagement": ( - "8031", - [ - "StreamingEnabled", - "StatusActive", - "SupportedVideoStreamTiers", - "SupportedAudioStreamTiers", - "SupportedRTPConfiguration", - "SetupEndpoints", - "RTPStreamingControl", - "SensorUUID", - ], - [], - ), - "CameraWebRTCStreamManagement": ( - "8033", - [ - "WebRTCSolicitOffer", - "WebRTCProvideAnswer", - "WebRTCStreamingControl", - "WebRTCNumberOfActiveSessions", - "WebRTCReoffer", - "WebRTCUpdateSession", - "WebRTCSupportedVideoStreamTiers", - "WebRTCSupportedAudioStreamTiers", - "StreamingEnabled", - "SensorUUID", - ], - [], - ), - "CameraKeyManagement": ("8050", ["CameraKey", "CameraKeyID"], []), - "CameraClientCertificateManagement": ( - "8080", - [ - "CameraClientCSR", - "CameraClientCertificate", - "CameraClientCertificateStatus", - ], - [], - ), -} - - -def _hap_uuid(short: str) -> UUID: - return UUID(f"0000{short}-0000-1000-8000-0026BB765291") - - -@pytest.mark.parametrize("name", HKSV3_CHARS) -def test_loader_hksv3_char(name: str) -> None: - """Test the HKSV 17.99 characteristics match the guide.""" - short, char_format, permissions, admin_only_access = HKSV3_CHARS[name] - char = Loader().get_char(name) - assert char.type_id == _hap_uuid(short) - assert char.properties[PROP_FORMAT] == char_format - assert set(char.properties[PROP_PERMISSIONS]) == permissions - assert char.properties.get(ADMIN_ONLY_ACCESS) == admin_only_access - - -def test_loader_hksv3_admin_only_not_advertised() -> None: +def test_loader_admin_only_not_advertised() -> None: """Test the admin only marker stays out of the HAP representation.""" char = Loader().get_char("StreamingEnabled") + assert char.properties[PROP_ADMIN_ONLY_ACCESS] == ["write"] with patch.object(char, "broker") as mock_broker: mock_broker.iid_manager.get_iid.return_value = 2 hap_repr = char.to_HAP() - assert hap_repr["perms"] == ["pr", "pw", "ev", "tw"] - assert ADMIN_ONLY_ACCESS not in hap_repr - - -def test_loader_hksv3_active_sessions_range() -> None: - """Test the WebRTC session count carries the guide's value range.""" - char = Loader().get_char("WebRTCNumberOfActiveSessions") - assert char.properties[PROP_MIN_VALUE] == 0 - assert char.properties[PROP_MAX_VALUE] == 255 - - -@pytest.mark.parametrize("name", HKSV3_SERVICES) -def test_loader_hksv3_service(name: str) -> None: - """Test the HKSV 17.99 services match the guide.""" - short, required, optional = HKSV3_SERVICES[name] - loader = Loader() - service = loader.get_service(name) - assert service.type_id == _hap_uuid(short) - assert [char.display_name for char in service.characteristics] == required - assert loader.serv_types[name]["OptionalCharacteristics"] == optional - assert set(optional) <= set(loader.char_types) - - -def test_loader_motion_sensor_hksv3_optional() -> None: - """Test the motion sensor gains the HKSV 17.99 chars only as optional.""" - loader = Loader() - optional = loader.serv_types["MotionSensor"]["OptionalCharacteristics"] - assert {"ContributingSensors", "MotionEnabled"} <= set(optional) - service = loader.get_service("MotionSensor") - assert [char.display_name for char in service.characteristics] == ["MotionDetected"] - - -@pytest.mark.parametrize( - ("name", "permissions"), - [ - ("CameraOperatingModeIndicator", {"pr", "pw", "ev", "tw"}), - ("NightVision", {"pr", "pw", "ev", "tw"}), - ("ThirdPartyCameraActive", {"pr", "ev"}), - ], -) -def test_loader_camera_operating_mode_permissions( - name: str, permissions: set[str] -) -> None: - """Test the operating mode settings shared with the HKSV 17.99 services.""" - char = Loader().get_char(name) - assert set(char.properties[PROP_PERMISSIONS]) == permissions + assert PROP_ADMIN_ONLY_ACCESS not in hap_repr From db56978d2e06aaa9b5cefab38afe922f00f6fe7b Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:05:31 +0100 Subject: [PATCH 6/7] Enforce AdminOnlyAccess for writes only --- pyhap/accessory_driver.py | 37 ++----- pyhap/characteristic.py | 6 +- pyhap/hap_handler.py | 4 +- pyhap/state.py | 2 +- tests/test_admin_only.py | 223 +++++++++++++------------------------- tests/test_loader.py | 11 -- 6 files changed, 86 insertions(+), 197 deletions(-) diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index 7553bc9f..b32e1f7b 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -38,7 +38,6 @@ from pyhap import util from pyhap.accessory import Accessory, Bridge, get_topic from pyhap.characteristic import ( - ADMIN_ONLY_READ, ADMIN_ONLY_WRITE, PROP_ADMIN_ONLY_ACCESS, PROP_PERMISSIONS, @@ -97,11 +96,6 @@ MAX_PREPARED_WRITE_TTL_SECONDS = 30.0 -def _admin_only(char: Characteristic, access: str) -> bool: - """Return whether ``access`` to ``char`` is restricted to admin controllers.""" - return access in char.properties.get(PROP_ADMIN_ONLY_ACCESS, ()) - - def _stored_value(char: Characteristic, value: HAPValue) -> HAPValue: """Return the value ``client_update_value`` stores for ``value``.""" try: @@ -940,11 +934,7 @@ def get_accessories(self, include_value: bool = True) -> dict[str, Any]: return {HAP_REPR_ACCS: hap_rep} def get_characteristics( - self, - char_ids: Any, - client_addr: tuple[str, int] | None = None, - *, - client_uuid: UUID | None = None, + self, char_ids: Any, client_addr: tuple[str, int] | None = None ) -> dict[str, Any]: """Returns values for the required characteristics. @@ -955,11 +945,6 @@ def get_characteristics( responses. :type client_addr: tuple - :param client_uuid: The paired controller that sent the request. Reads - of characteristics restricted to admins are refused with -70401 - unless it is an admin. - :type client_uuid: uuid.UUID - :return: Status success for each required characteristic. For example: .. code-block:: python @@ -975,7 +960,6 @@ def get_characteristics( :rtype: dict """ assert self.accessory is not None # noqa: S101 - is_admin = self._is_admin(client_uuid) write_responses = ( self._write_responses.get(client_addr, {}) if client_addr else {} ) @@ -1009,13 +993,9 @@ def get_characteristics( # A characteristic IID may resolve to a Service or be unknown; # only a Characteristic can report a value, and only when it - # has the paired read permission. Privileges are checked first, - # before availability, so a controller that may not read the - # characteristic learns nothing about its accessory. + # has the paired read permission. if isinstance(char, Characteristic): - if not is_admin and _admin_only(char, ADMIN_ONLY_READ): - rep[HAP_REPR_STATUS] = HAP_SERVER_STATUS.INSUFFICIENT_PRIVILEGES - elif HAP_PERMISSION_READ not in char.properties[PROP_PERMISSIONS]: + if HAP_PERMISSION_READ not in char.properties[PROP_PERMISSIONS]: rep[HAP_REPR_STATUS] = ( HAP_SERVER_STATUS.WRITE_ONLY_CHARACTERISTIC ) @@ -1117,6 +1097,7 @@ def set_characteristics( 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] @@ -1143,10 +1124,12 @@ def set_characteristics( if char is None: continue + # Privileges are checked before the permissions, so a controller + # that may not write the characteristic learns nothing else. if ( value is not None - and _admin_only(char, ADMIN_ONLY_WRITE) - and not self._is_admin(client_uuid) + and 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 @@ -1284,10 +1267,6 @@ def _store_write_response( char.value = valid return valid - def _is_admin(self, client_uuid: UUID | None) -> bool: - """Return whether ``client_uuid`` is a paired admin controller.""" - return client_uuid is not None and self.state.is_admin(client_uuid) - def prepare( self, prepare_query: dict[str, Any], client_addr: tuple[str, int] ) -> dict[str, int]: diff --git a/pyhap/characteristic.py b/pyhap/characteristic.py index 24de89fb..8770c8a0 100644 --- a/pyhap/characteristic.py +++ b/pyhap/characteristic.py @@ -98,10 +98,10 @@ def _setter_wants_client_addr(setter: Callable[..., Any]) -> bool: HAP_UNIT_SECONDS = "seconds" # ### Properties ### -# Accesses ("read", "write") only admin controllers may use. Not part of the -# HAP representation, the accessory driver enforces it. +# 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_READ = "read" ADMIN_ONLY_WRITE = "write" PROP_FORMAT = "Format" PROP_MAX_VALUE = "maxValue" diff --git a/pyhap/hap_handler.py b/pyhap/hap_handler.py index d41a38ec..7361429c 100644 --- a/pyhap/hap_handler.py +++ b/pyhap/hap_handler.py @@ -679,9 +679,7 @@ def handle_get_characteristics(self) -> None: assert self.parsed_url is not None # noqa: S101 params = parse_qs(self.parsed_url.query) response = self.accessory_handler.get_characteristics( - params["id"][0].split(","), - cast("tuple[str, int]", self.client_address), - client_uuid=self.client_uuid, + params["id"][0].split(","), cast("tuple[str, int]", self.client_address) ) chars = response[HAP_REPR_CHARS] diff --git a/pyhap/state.py b/pyhap/state.py index 8922f358..428c1884 100644 --- a/pyhap/state.py +++ b/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 diff --git a/tests/test_admin_only.py b/tests/test_admin_only.py index 0d4571e1..49284548 100644 --- a/tests/test_admin_only.py +++ b/tests/test_admin_only.py @@ -4,14 +4,14 @@ import io import json from typing import Any, NamedTuple -from unittest.mock import MagicMock, PropertyMock, patch +from unittest.mock import MagicMock, patch from urllib.parse import urlparse from uuid import UUID import pytest from pyhap import hap_handler -from pyhap.accessory import Accessory +from pyhap.accessory import Accessory, get_topic from pyhap.accessory_driver import AccessoryDriver from pyhap.characteristic import ( HAP_FORMAT_BOOL, @@ -22,6 +22,7 @@ Characteristic, ) from pyhap.const import ( + HAP_PERMISSION_NOTIFY, HAP_PERMISSION_WRITE, HAP_PERMISSIONS, HAP_REPR_AID, @@ -85,9 +86,9 @@ def chars(driver: AccessoryDriver) -> Iterator[dict[str, SensorChar]]: """Add a motion sensor and pair an admin and a regular controller. Motion Enabled may only be written by admins. Secret is a made-up - characteristic that only admins may read. Open is a made-up characteristic - anyone may read and write. Sealed is made up, write only and readable only - by admins, Locked is made up, read only and writable only by admins. + characteristic whose restriction lists read, which is not enforced. Open is + a made-up characteristic anyone may read and write. Locked is made up, read + only and writable only by admins. """ acc = Accessory(driver, "Sensor") service = driver.loader.get_service("MotionSensor") @@ -110,15 +111,6 @@ def chars(driver: AccessoryDriver) -> Iterator[dict[str, SensorChar]]: PROP_ADMIN_ONLY_ACCESS: ["read"], }, ), - Characteristic( - "Sealed", - UUID("2d7f7c38-1d4c-4e0b-9d0e-4c1f1d2c7a12"), - { - PROP_FORMAT: HAP_FORMAT_BOOL, - PROP_PERMISSIONS: [HAP_PERMISSION_WRITE], - PROP_ADMIN_ONLY_ACCESS: ["read"], - }, - ), Characteristic( "Locked", UUID("2d7f7c38-1d4c-4e0b-9d0e-4c1f1d2c7a13"), @@ -138,7 +130,6 @@ def chars(driver: AccessoryDriver) -> Iterator[dict[str, SensorChar]]: "MotionEnabled": _with_setter(acc, service, "MotionEnabled"), "Open": _with_setter(acc, service, "Open"), "Secret": _with_setter(acc, service, "Secret"), - "Sealed": _with_setter(acc, service, "Sealed"), "Locked": _with_setter(acc, service, "Locked"), } @@ -221,183 +212,115 @@ def test_legacy_state_pairs_everyone_as_admin( char.setter.assert_called_once_with(True) -@pytest.mark.parametrize( - ("name", "client_uuid", "expected"), - [ - ("Secret", ADMIN, {HAP_REPR_STATUS: 0, HAP_REPR_VALUE: False}), - ("Secret", USER, {HAP_REPR_STATUS: REFUSED}), - ("Secret", None, {HAP_REPR_STATUS: REFUSED}), - ("MotionEnabled", USER, {HAP_REPR_STATUS: 0, HAP_REPR_VALUE: False}), - ("MotionEnabled", None, {HAP_REPR_STATUS: 0, HAP_REPR_VALUE: False}), - ], - ids=[ - "admin", - "user", - "unknown_controller", - "write_restricted_user", - "write_restricted_unknown_controller", - ], -) -def test_admin_only_read( +@pytest.mark.parametrize("name", ["Secret", "MotionEnabled"]) +@pytest.mark.parametrize("client_uuid", [ADMIN, USER], ids=["admin", "user"]) +def test_admin_only_read_not_refused( driver: AccessoryDriver, chars: dict[str, SensorChar], name: str, - client_uuid: UUID | None, - expected: dict[str, Any], + client_uuid: UUID, ) -> None: - """Test reads are only refused when the restriction lists read.""" + """Test reads are not restricted, also not when the restriction lists read.""" char = chars[name] + handler = hap_handler.HAPServerHandler(driver, CLIENT) + handler.is_encrypted = True + handler.client_uuid = client_uuid + handler.response = hap_handler.HAPResponse() + handler.parsed_url = urlparse(f"/characteristics?id={char.aid}.{char.iid}") - response = driver.get_characteristics( - [f"{char.aid}.{char.iid}"], client_uuid=client_uuid - ) + handler.handle_get_characteristics() - assert response == { - HAP_REPR_CHARS: [{HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, **expected}] + assert handler.response.status_code == 200 + assert json.loads(handler.response.body) == { + HAP_REPR_CHARS: [ + {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_VALUE: False} + ] } @pytest.mark.parametrize( - ("client_uuid", "status"), - [ - (USER, REFUSED), - (ADMIN, HAP_SERVER_STATUS.READ_ONLY_CHARACTERISTIC), - ], - ids=["user", "admin"], -) -def test_admin_only_write_refused_before_read_only( - driver: AccessoryDriver, - chars: dict[str, SensorChar], - client_uuid: UUID, - status: int, -) -> None: - """Test a write to a read only admin char is refused for privileges first.""" - char = chars["Locked"] - - response = driver.set_characteristics(_write(char), CLIENT, client_uuid=client_uuid) - - assert response == _response(char, status) - char.setter.assert_not_called() - - -@pytest.mark.parametrize( - ("client_uuid", "status"), - [ - (USER, REFUSED), - (ADMIN, HAP_SERVER_STATUS.WRITE_ONLY_CHARACTERISTIC), - ], - ids=["user", "admin"], + "client_uuid", [ADMIN, USER, None], ids=["admin", "user", "unknown"] ) -def test_admin_only_read_refused_before_write_only( - driver: AccessoryDriver, - chars: dict[str, SensorChar], - client_uuid: UUID, - status: int, +def test_admin_only_event_subscription_allowed( + driver: AccessoryDriver, chars: dict[str, SensorChar], client_uuid: UUID | None ) -> None: - """Test a read of a write only admin char is refused for privileges first.""" - char = chars["Sealed"] - - response = driver.get_characteristics( - [f"{char.aid}.{char.iid}"], CLIENT, client_uuid=client_uuid - ) - - assert response == { + """Test a write without a value only subscribes, whoever sends it.""" + char = chars["MotionEnabled"] + query = { HAP_REPR_CHARS: [ - {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_STATUS: status} + { + HAP_REPR_AID: char.aid, + HAP_REPR_IID: char.iid, + HAP_PERMISSION_NOTIFY: True, + } ] } + with patch.object(driver, "async_subscribe_client_topic") as subscribe: + response = driver.set_characteristics(query, CLIENT, client_uuid=client_uuid) + + assert response is None + subscribe.assert_called_once_with(CLIENT, get_topic(char.aid, char.iid), True) + char.setter.assert_not_called() + @pytest.mark.parametrize( ("client_uuid", "status"), [ (USER, REFUSED), - (ADMIN, HAP_SERVER_STATUS.SERVICE_COMMUNICATION_FAILURE), + (ADMIN, HAP_SERVER_STATUS.READ_ONLY_CHARACTERISTIC), ], ids=["user", "admin"], ) -def test_admin_only_read_refused_before_availability( +def test_admin_only_write_refused_before_read_only( driver: AccessoryDriver, chars: dict[str, SensorChar], client_uuid: UUID, status: int, ) -> None: - """Test a read of an unavailable admin char is refused for privileges first.""" - char = chars["Secret"] + """Test a write to a read only admin char is refused for privileges first.""" + char = chars["Locked"] - with patch.object( - Accessory, "available", new_callable=PropertyMock, return_value=False - ): - response = driver.get_characteristics( - [f"{char.aid}.{char.iid}"], CLIENT, client_uuid=client_uuid - ) + response = driver.set_characteristics(_write(char), CLIENT, client_uuid=client_uuid) - assert response == { - HAP_REPR_CHARS: [ - {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_STATUS: status} - ] - } + assert response == _response(char, status) + char.setter.assert_not_called() -@pytest.mark.parametrize( - ("client_uuid", "status_code", "body"), - [ - (ADMIN, 204, b""), - (USER, 207, b'{"characteristics":[{"status":-70401,"aid":1,"iid":10}]}'), - ], - ids=["admin", "user"], -) -def test_handler_passes_the_controller_on_write( - driver: AccessoryDriver, - chars: dict[str, SensorChar], - client_uuid: UUID, - status_code: int, - body: bytes, -) -> None: - """Test the HAP handler writes as the controller of its session.""" - char = chars["MotionEnabled"] +def _write_through_handler( + driver: AccessoryDriver, char: SensorChar, client_uuid: UUID +) -> hap_handler.HAPResponse: + """Write True to ``char`` through a HAP handler of ``client_uuid``.""" handler = hap_handler.HAPServerHandler(driver, CLIENT) handler.is_encrypted = True handler.client_uuid = client_uuid handler.response = hap_handler.HAPResponse() - handler.request_body = json.dumps( - { - HAP_REPR_CHARS: [ - {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_VALUE: True} - ] - } - ).encode() - + handler.request_body = json.dumps(_write(char)).encode() handler.handle_set_characteristics() + return handler.response - assert handler.response.status_code == status_code - assert handler.response.body == body +def test_handler_writes_as_admin( + driver: AccessoryDriver, chars: dict[str, SensorChar] +) -> None: + """Test the HAP handler writes as the admin controller of its session.""" + char = chars["MotionEnabled"] -@pytest.mark.parametrize( - ("client_uuid", "status_code", "body"), - [ - (ADMIN, 200, b'{"characteristics":[{"aid":1,"iid":12,"value":false}]}'), - (USER, 207, b'{"characteristics":[{"aid":1,"iid":12,"status":-70401}]}'), - ], - ids=["admin", "user"], -) -def test_handler_passes_the_controller_on_read( - driver: AccessoryDriver, - chars: dict[str, SensorChar], - client_uuid: UUID, - status_code: int, - body: bytes, + response = _write_through_handler(driver, char, ADMIN) + + assert response.status_code == 204 + assert response.body == b"" + char.setter.assert_called_once_with(True) + + +def test_handler_writes_as_user( + driver: AccessoryDriver, chars: dict[str, SensorChar] ) -> None: - """Test the HAP handler reads as the controller of its session.""" - char = chars["Secret"] - handler = hap_handler.HAPServerHandler(driver, CLIENT) - handler.is_encrypted = True - handler.client_uuid = client_uuid - handler.response = hap_handler.HAPResponse() - handler.parsed_url = urlparse(f"/characteristics?id={char.aid}.{char.iid}") + """Test the HAP handler writes as the regular controller of its session.""" + char = chars["MotionEnabled"] - handler.handle_get_characteristics() + response = _write_through_handler(driver, char, USER) - assert handler.response.status_code == status_code - assert handler.response.body == body + assert response.status_code == 207 + assert json.loads(response.body) == _response(char, REFUSED) + char.setter.assert_not_called() diff --git a/tests/test_loader.py b/tests/test_loader.py index 21ba6f3a..a38f1426 100644 --- a/tests/test_loader.py +++ b/tests/test_loader.py @@ -7,7 +7,6 @@ from pyhap import CHARACTERISTICS_FILE, SERVICES_FILE from pyhap.characteristic import ( - PROP_ADMIN_ONLY_ACCESS, PROP_FORMAT, PROP_MAX_VALUE, PROP_MIN_VALUE, @@ -86,16 +85,6 @@ def test_get_loader() -> None: assert get_loader() == loader -def test_loader_admin_only_not_advertised() -> None: - """Test the admin only marker stays out of the HAP representation.""" - char = Loader().get_char("StreamingEnabled") - assert char.properties[PROP_ADMIN_ONLY_ACCESS] == ["write"] - with patch.object(char, "broker") as mock_broker: - mock_broker.iid_manager.get_iid.return_value = 2 - hap_repr = char.to_HAP() - assert PROP_ADMIN_ONLY_ACCESS not in hap_repr - - # Accesses only admin controllers may use, not part of the HAP representation. ADMIN_ONLY_ACCESS = "AdminOnlyAccess" From f1a182ef81068541e1fcf94f58a38e58cb827e34 Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Tue, 10 Nov 2026 18:28:48 +0100 Subject: [PATCH 7/7] Refuse an explicit null from a regular controller before any setter Also say in the catalog generator that the driver enforces admin-only writes. --- pyhap/accessory_driver.py | 9 ++++----- scripts/gen_hap_types.py | 4 ++-- tests/test_admin_only.py | 24 ++++++++++++++++++++++++ 3 files changed, 30 insertions(+), 7 deletions(-) diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index 592e9932..37c490b2 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -1153,7 +1153,7 @@ def set_characteristics( :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 + :type client_uuid: uuid.UUID or None """ results, pending = self._apply_characteristic_writes( chars_query, client_addr, client_uuid @@ -1293,10 +1293,9 @@ def _run_characteristic_writes( # Privileges are checked before the permissions, so a controller # that may not write the characteristic learns nothing else. - if ( - value is not None - and not is_admin - and ADMIN_ONLY_WRITE in char.properties.get(PROP_ADMIN_ONLY_ACCESS, ()) + # 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 diff --git a/scripts/gen_hap_types.py b/scripts/gen_hap_types.py index 907c17a4..2477baad 100755 --- a/scripts/gen_hap_types.py +++ b/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 diff --git a/tests/test_admin_only.py b/tests/test_admin_only.py index 49284548..f63f2832 100644 --- a/tests/test_admin_only.py +++ b/tests/test_admin_only.py @@ -287,6 +287,30 @@ def test_admin_only_write_refused_before_read_only( char.setter.assert_not_called() +def test_admin_only_null_write_reaches_no_setter( + driver: AccessoryDriver, chars: dict[str, SensorChar] +) -> None: + """Test an explicit null from a regular controller is refused before setters.""" + char = chars["MotionEnabled"] + assert driver.accessory is not None + service = driver.accessory.get_service("MotionSensor") + assert service is not None + service.setter_callback = MagicMock() + driver.accessory.setter_callback = MagicMock() + query = { + HAP_REPR_CHARS: [ + {HAP_REPR_AID: char.aid, HAP_REPR_IID: char.iid, HAP_REPR_VALUE: None} + ] + } + + response = driver.set_characteristics(query, CLIENT, client_uuid=USER) + + assert response == _response(char, REFUSED) + char.setter.assert_not_called() + service.setter_callback.assert_not_called() + driver.accessory.setter_callback.assert_not_called() + + def _write_through_handler( driver: AccessoryDriver, char: SensorChar, client_uuid: UUID ) -> hap_handler.HAPResponse: