diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index 7f7326f2..37c490b2 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -32,6 +32,7 @@ import threading import time from typing import Any, NamedTuple, TypeGuard +from uuid import UUID from zeroconf import ServiceInfo from zeroconf.asyncio import AsyncZeroconf @@ -39,6 +40,8 @@ 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,7 +1195,9 @@ 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 @@ -1187,12 +1205,17 @@ def _set_characteristics_for_controller( 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] diff --git a/pyhap/characteristic.py b/pyhap/characteristic.py index 3dd44c9f..19a2730e 100644 --- a/pyhap/characteristic.py +++ b/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" diff --git a/pyhap/hap_handler.py b/pyhap/hap_handler.py index 8c7844cd..b3a26bfb 100644 --- a/pyhap/hap_handler.py +++ b/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): 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/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 new file mode 100644 index 00000000..f63f2832 --- /dev/null +++ b/tests/test_admin_only.py @@ -0,0 +1,350 @@ +"""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, get_topic +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_NOTIFY, + 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 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") + 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"], + }, + ), + Characteristic( + "Locked", + UUID("2d7f7c38-1d4c-4e0b-9d0e-4c1f1d2c7a13"), + { + PROP_FORMAT: HAP_FORMAT_BOOL, + PROP_PERMISSIONS: [HAP_PERMISSION_READ], + PROP_ADMIN_ONLY_ACCESS: ["write"], + }, + ), + ) + 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"), + "Locked": _with_setter(acc, service, "Locked"), + } + + +@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=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, client_uuid=ADMIN) is None + + change(driver) + + assert driver.set_characteristics( + _write(char), CLIENT, client_uuid=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, client_uuid=USER) is None + char.setter.assert_called_once_with(True) + + +@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: + """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}") + + handler.handle_get_characteristics() + + 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", [ADMIN, USER, None], ids=["admin", "user", "unknown"] +) +def test_admin_only_event_subscription_allowed( + driver: AccessoryDriver, chars: dict[str, SensorChar], client_uuid: UUID | None +) -> None: + """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_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.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() + + +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: + """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(_write(char)).encode() + handler.handle_set_characteristics() + return handler.response + + +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"] + + 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 writes as the regular controller of its session.""" + char = chars["MotionEnabled"] + + response = _write_through_handler(driver, char, USER) + + assert response.status_code == 207 + assert json.loads(response.body) == _response(char, REFUSED) + char.setter.assert_not_called()