From d92069f0ce193927a77c6fa9c803cb73b0e92b3e Mon Sep 17 00:00:00 2001 From: Raphael Hehl <7577984+RaHehl@users.noreply.github.com> Date: Sat, 4 Jul 2026 20:38:06 +0200 Subject: [PATCH] Add classic HomeKit Secure Video recording Implements HomeKit Secure Video (classic HKSV) so an accessory can record event-triggered fragmented-MP4 clips to a HomeKit Home Hub over the HomeKit Data Stream (HDS). Verified end-to-end against a real iOS 26.5 Home Hub: motion triggers the hub to open a recording stream and pull fMP4 fragments over the encrypted HDS connection. Contents: - HDS transport: framing + ChaCha20-Poly1305 key derivation (pyhap/hds.py), the message/opack protocol codec (hds_protocol.py), the TCP listener and connection dispatch (hds_server.py), and the dataSend recording fragment transfer (hds_recording.py). - Recording configuration TLV codecs matching the HAP spec / HAP-NodeJS (hksv_recording.py): supported vs. selected encoding, list separators, per-attribute resolutions; encode/decode are faithful inverses in both directions (single value <-> scalar, several <-> list). - Camera wiring (camera.py): CameraRecordingManagement, the Camera Operating Mode, the DataStreamTransportManagement service, an HKSV event-trigger MotionSensor, and the recording-stream delegate hook. Stream management is set up only when the options describe video, so a recording-only camera is possible (streaming_status is None for it). - Recording eligibility characteristics iOS validates before managing recording: Active (0x0B0) on CameraRTPStreamManagement, StatusActive on the MotionSensor, PeriodicSnapshotsActive on the operating mode, and uint8 timed-write HomeKitCameraActive/RecordingAudioActive. The shared Version characteristic becomes paired-read only (per the HAP spec; it previously also advertised events, which the accessory never emitted - existing to_HAP expectations updated accordingly). - SetupDataStreamTransport is a write-response characteristic: the setter returns the encoded response (TCP listening port + accessory key salt) so the HAP layer delivers it in the write response and the controller can open the HDS connection. - Integration helpers: fmp4_recording_packets() turns any fragmented-MP4 byte source (ffmpeg stdout, an aiohttp/go2rtc response, ...) into the init packet + media fragments a recording delegate yields, and set_motion_detected() drives the event trigger - so a consumer supplies a keyframe-aligned fMP4 source and a motion signal rather than reimplementing the framing. - Framework hooks: the pair-verify shared secret is exposed per session for HDS key derivation (and popped on disconnect), and setters can opt into the sender's client address (introspection cached per setter, not per write). - accessory_driver guards the asyncio.SafeChildWatcher usage which was removed in Python 3.14. Robustness (asyncio / resource management): - The pre-handshake receive buffer is capped and connections that never complete the control handshake time out and close, so an unauthenticated peer on the advertised port cannot make the process buffer unboundedly. Frames announcing more than the 1 MiB payload cap are rejected. - Pending transport registrations are bounded; malformed frames/messages and handler errors are contained (bounded opack recursion, guarded dispatch) rather than tearing down the link uncaught. A stale dataSend close for an earlier stream cannot stop the current one. - The recording task and its delegate are torn down exactly once on dataSend close and/or connection loss (a close-callback list, not a single slot; a second cancel cannot abort the in-flight generator aclose()), and the delegate's async generator is aclose()d so its subprocess is released deterministically. - Fragment delivery honours transport write back-pressure (pause/resume) so a slow controller cannot make the accessory buffer a whole fMP4 fragment. Tests: HDS fragment transfer and chunking, connection-loss cleanup, the close-then-disconnect double-stop, write back-pressure, pre-bind buffer cap and pending bounds, recording configuration round-trips (incl. multi-value), recording management setup/eligibility, and the SetupDataStreamTransport write-response handshake (through the HAP path, plus a single-argument-setter regression guard). --- pyhap/accessory_driver.py | 7 + pyhap/camera.py | 235 +++++++++++++- pyhap/characteristic.py | 44 ++- pyhap/hap_protocol.py | 5 + pyhap/hds.py | 185 +++++++++++ pyhap/hds_protocol.py | 307 ++++++++++++++++++ pyhap/hds_recording.py | 262 ++++++++++++++++ pyhap/hds_server.py | 307 ++++++++++++++++++ pyhap/hksv_recording.py | 364 ++++++++++++++++++++++ pyhap/resources/characteristics.json | 132 +++++++- pyhap/resources/services.json | 37 +++ tests/test_accessory.py | 4 +- tests/test_hds_recording.py | 449 +++++++++++++++++++++++++++ tests/test_hksv_recording.py | 280 +++++++++++++++++ 14 files changed, 2607 insertions(+), 11 deletions(-) create mode 100644 pyhap/hds.py create mode 100644 pyhap/hds_protocol.py create mode 100644 pyhap/hds_recording.py create mode 100644 pyhap/hds_server.py create mode 100644 pyhap/hksv_recording.py create mode 100644 tests/test_hds_recording.py create mode 100644 tests/test_hksv_recording.py diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index 782f492a..4824af18 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -292,6 +292,9 @@ def __init__( self.persist_file = os.path.expanduser(persist_file) self.encoder = encoder or AccessoryEncoder() self.topics = {} # topic: set of (address, port) of subscribed clients + # Per-session HAP shared keys, keyed by client (address, port). Needed + # by transports derived from the session, e.g. HomeKit Data Stream. + self.session_shared_keys: Dict[Tuple[str, int], bytes] = {} self.loader = loader or Loader() self.aio_stop_event = None self.stop_event = threading.Event() @@ -322,6 +325,9 @@ def start(self): if ( threading.current_thread() is threading.main_thread() and os.name != "nt" + # The child watcher API was removed in Python 3.14; there the + # default event loop reaps subprocesses without one. + and hasattr(asyncio, "SafeChildWatcher") ): logger.debug("Setting child watcher") watcher = asyncio.SafeChildWatcher() # pylint: disable=deprecated-class @@ -537,6 +543,7 @@ def connection_lost(self, client): for topic in client_topics: self.async_subscribe_client_topic(client, topic, subscribe=False) self.prepared_writes.pop(client, None) + self.session_shared_keys.pop(client, None) def publish(self, data, sender_client_addr=None, immediate=False): """Publishes an event to the client. diff --git a/pyhap/camera.py b/pyhap/camera.py index f77de394..cd9e0625 100644 --- a/pyhap/camera.py +++ b/pyhap/camera.py @@ -25,10 +25,10 @@ import sys from uuid import UUID -from pyhap import RESOURCE_DIR, tlv +from pyhap import RESOURCE_DIR, hds, hds_recording, hds_server, hksv_recording, tlv from pyhap.accessory import Accessory from pyhap.const import CATEGORY_CAMERA -from pyhap.util import byte_bool, to_base64_str +from pyhap.util import base64_to_bytes, byte_bool, to_base64_str if sys.version_info >= (3, 11): from asyncio import timeout as async_timeout @@ -450,12 +450,25 @@ def __init__(self, options, *args, **kwargs): self.add_preload_service("Microphone") self._streaming_status = [] self._management = [] - self._setup_stream_management(options) + # All HKSV state defaults are initialized before any setup runs so the + # setup methods below never follow (and reset) them. + self._status_active_char = None + self._recording_service = None + self._selected_recording_config = None + self._motion_detected_char = None + self._hds_listener = None + self._hds_connections = set() + if options.get("video"): + self._setup_stream_management(options) + # Classic HomeKit Secure Video recording (HDS + fragmented MP4). + if options.get("recording"): + self._setup_recording_management(options) @property def streaming_status(self): """For backwards compatibility.""" - return self._streaming_status[0] + # A recording-only camera has no RTP stream management service. + return self._streaming_status[0] if self._streaming_status else None def _setup_stream_management(self, options): """Create stream management.""" @@ -467,8 +480,12 @@ def _setup_stream_management(self, options): def _create_stream_management(self, stream_idx, options): """Create a stream management service.""" management = self.add_preload_service( - "CameraRTPStreamManagement", unique_id=stream_idx + "CameraRTPStreamManagement", unique_id=stream_idx, chars=["Active"] ) + # HAP-NodeJS exposes an Active (0x0B0) characteristic on every stream + # management service (value 1 = active); iOS' camera validation expects + # it, and it is required for Secure Video eligibility. + management.configure_char("Active", value=1) management.configure_char( "StreamingStatus", getter_callback=lambda: self._get_streaming_status(stream_idx), @@ -497,6 +514,203 @@ def _create_stream_management(self, stream_idx, options): ) return management + def _default_recording_configs(self, options): + """Derive supported recording configurations from the streaming options. + + A widely-compatible default: H.264 (Baseline/Main/High, levels 3.1/3.2/ + 4.0) at the streaming resolutions, AAC-LC audio. Accessories can override + this to advertise their own recording capabilities. + """ + video_opts = options.get("video") or {} + resolutions = video_opts.get("resolutions") or [[1920, 1080, 30]] + video = [ + hksv_recording.VideoCodecConfiguration( + codec_type=hksv_recording.VideoCodecType.H264, + profile=[ + VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["BASELINE"][0], + VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["MAIN"][0], + VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["HIGH"][0], + ], + level=[ + VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE3_1"][0], + VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE3_2"][0], + VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE4_0"][0], + ], + bitrate_kbps=2000, + iframe_interval_ms=4000, + attributes=[ + hksv_recording.VideoAttributes(width, height, fps) + for width, height, fps in resolutions + ], + ) + ] + audio = [hksv_recording.AudioCodecConfiguration()] + general = hksv_recording.SupportedRecordingConfiguration() + return general, video, audio + + def _setup_recording_management(self, options): + """Create the Camera Recording Management service (classic HKSV). + + Advertises the supported general/video/audio recording configurations + and accepts the controller's selection. Recording stays inactive + (Active=0) unless the ``recording`` option enables it; the fragment + data flow over HDS is wired by the recording transport. + """ + # HKSV requires the Camera Operating Mode service (0000021A); the + # controller validates its presence before enabling recording. + operating_mode = self.add_preload_service( + "CameraOperatingMode", chars=["PeriodicSnapshotsActive"] + ) + operating_mode.configure_char("EventSnapshotsActive", value=1) + operating_mode.configure_char("HomeKitCameraActive", value=1) + operating_mode.configure_char("PeriodicSnapshotsActive", value=1) + + # HKSV records on an event trigger; the camera owns a motion sensor and + # links it (and the data stream transport) to the recording management. + motion = self.add_preload_service("MotionSensor", chars=["StatusActive"]) + self._motion_detected_char = motion.configure_char( + "MotionDetected", value=False + ) + # HAP-NodeJS marks the HKSV event-trigger sensor as active; iOS expects it. + motion.configure_char("StatusActive", value=True) + + general, video, audio = self._default_recording_configs(options) + service = self.add_preload_service( + "CameraRecordingManagement", chars=["RecordingAudioActive"] + ) + service.add_linked_service(motion) + service.configure_char("Active", value=0) + service.configure_char( + "RecordingAudioActive", value=1 if options.get("recording_audio") else 0 + ) + service.configure_char( + "SupportedCameraRecordingConfiguration", + value=to_base64_str(general.encode()), + ) + service.configure_char( + "SupportedVideoRecordingConfiguration", + value=to_base64_str(hksv_recording.encode_supported_video(video)), + ) + service.configure_char( + "SupportedAudioRecordingConfiguration", + value=to_base64_str(hksv_recording.encode_supported_audio(audio)), + ) + service.configure_char( + "SelectedCameraRecordingConfiguration", + setter_callback=self.set_selected_recording_configuration, + ) + self._recording_service = service + self._setup_data_stream_transport() + + def _setup_data_stream_transport(self): + """Create the Data Stream Transport Management service used by HDS. + + SetupDataStreamTransport needs the HAP session shared secret, so its + setter receives the sender's client address to look the secret up. + """ + transport = self.add_preload_service("DataStreamTransportManagement") + transport.configure_char("Version", value="1.0") + transport.configure_char( + "SupportedDataStreamTransportConfiguration", + value=to_base64_str(self._supported_data_stream_transport()), + ) + transport.configure_char( + "SetupDataStreamTransport", + setter_callback=self.set_data_stream_transport, + ) + # HKSV requires the recording management service to link the data stream + # transport it records over; without the link the controller rejects + # recording during validation, before writing any configuration. + self._recording_service.add_linked_service(transport) + + @staticmethod + def _supported_data_stream_transport(): + # A single supported transport configuration: TCP. + transport_configuration = tlv.encode(b"\x01", hds.TRANSPORT_TYPE_TCP) + return tlv.encode(b"\x01", transport_configuration) + + async def _ensure_hds_listener(self): + if self._hds_listener is None: + self._hds_listener = hds_server.HDSListener() + self._hds_listener.on_connection = self._on_hds_connection + await self._hds_listener.start() + + def _on_hds_connection(self, connection): + self._hds_connections.add(connection) + connection.add_close_callback(self._hds_connections.discard) + hds_recording.RecordingStreamManager(connection, self._recording_delegate) + + def set_data_stream_transport(self, value, sender_client_addr=None): + """Handle a write to SetupDataStreamTransport (HDS session setup). + + SetupDataStreamTransport is a write-response ('wr') characteristic: the + controller writes with ``r: true`` and reads the encoded setup response + (TCP listening port + accessory key salt) straight back from the write + response. That means the value MUST be RETURNED from the setter so the + HAP layer includes it in the write response — setting it on the + characteristic is not delivered to the controller and leaves it without + the port, so it can never open the HDS connection. + """ + request = hds.SetupRequest.decode(base64_to_bytes(value)) + shared_secret = self.driver.session_shared_keys.get(sender_client_addr) + if shared_secret is None or self._hds_listener is None: + response = hds.encode_setup_response( + 0, b"", status=hds.SETUP_STATUS_GENERIC_ERROR + ) + else: + accessory_salt = self._hds_listener.register_transport( + shared_secret, request.controller_key_salt + ) + response = hds.encode_setup_response( + self._hds_listener.port, accessory_salt + ) + return to_base64_str(response) + + async def _recording_delegate(self, stream_id): + """Yield recording fragments for ``stream_id``. + + Override ``handle_recording_stream`` to produce the fragmented MP4 + packets; this adapter exists so the transport can await an async + generator. + """ + async for packet in self.handle_recording_stream(stream_id): + yield packet + + async def handle_recording_stream(self, stream_id): + """Produce the fragmented-MP4 recording for ``stream_id``. + + Override to yield :class:`pyhap.hds_recording.RecordingPacket` objects, + the first being the MP4 initialization segment. The default yields + nothing (no recording). + """ + return + yield # pragma: no cover - makes this an async generator + + @property + def selected_recording_configuration(self): + """The controller's selected recording configuration, if any.""" + return self._selected_recording_config + + def set_selected_recording_configuration(self, value): + """Handle a write to Selected Camera Recording Configuration (spec HKSV).""" + self._selected_recording_config = ( + hksv_recording.SelectedRecordingConfiguration.decode(base64_to_bytes(value)) + ) + self.recording_configuration_selected(self._selected_recording_config) + + def recording_configuration_selected(self, configuration): + """React to the controller selecting a recording configuration. Override.""" + + def set_motion_detected(self, detected): + """Fire (``True``) or clear (``False``) the recording motion trigger. + + This is the entry point an integration calls when its own motion + detector changes state: it drives the HKSV event-trigger MotionSensor, + which is what makes the Home Hub start (and later finish) a recording. + """ + if self._motion_detected_char is not None: + self._motion_detected_char.set_value(bool(detected)) + async def _start_stream(self, objs, reconfigure): # pylint: disable=unused-argument """Start or reconfigure video streaming for the given session. @@ -832,11 +1046,20 @@ def set_endpoints(self, value, stream_idx=None): response_tlv ) + async def run(self): + """Start the HDS listener when recording transport is available.""" + await super().run() + if self._recording_service is not None: + await self._ensure_hds_listener() + async def stop(self): - """Stop all streaming sessions.""" + """Stop all streaming sessions and the HDS listener.""" await asyncio.gather( *(self.stop_stream(session_info) for session_info in self.sessions.values()) ) + if self._hds_listener is not None: + await self._hds_listener.stop() + self._hds_listener = None # ### For client extensions ### diff --git a/pyhap/characteristic.py b/pyhap/characteristic.py index ee768fe8..b3ac8e42 100644 --- a/pyhap/characteristic.py +++ b/pyhap/characteristic.py @@ -5,6 +5,7 @@ a temperature measuring or a device status. """ +import inspect import logging from typing import TYPE_CHECKING, Any, Callable, Dict, Optional, Tuple from uuid import UUID @@ -28,6 +29,23 @@ logger = logging.getLogger(__name__) + +def _setter_wants_client_addr(setter: Callable) -> bool: + """Whether ``setter`` opts in to receiving the sender's client address. + + A setter opts in by declaring a parameter literally named + ``sender_client_addr`` (used e.g. for HDS transport setup that needs the HAP + session). Every other setter - including the common two-argument + default-value closures ``lambda value, option=option: ...`` - is called with + the value alone, unchanged. + """ + try: + parameters = inspect.signature(setter).parameters + except (TypeError, ValueError): + return False + return "sender_client_addr" in parameters + + # ### HAP Format ### HAP_FORMAT_BOOL = "bool" HAP_FORMAT_INT = "int" @@ -140,6 +158,8 @@ class Characteristic: "_to_hap_cache_with_value", "_to_hap_cache", "_always_null", + "_setter_addr_cache", + "_setter_addr_cache_for", ) def __init__( @@ -181,6 +201,10 @@ def __init__( self._value = self._get_default_value() self.getter_callback: Optional[Callable[[], Any]] = None self.setter_callback: Optional[Callable[[Any], None]] = None + # Cache of whether ``setter_callback`` opts into the client address, + # recomputed only when the setter object itself changes. + self._setter_addr_cache: bool = False + self._setter_addr_cache_for: Optional[Callable] = None self.service: Optional["Service"] = None self.unique_id = unique_id self._uuid_str = uuid_to_hap_type(type_id) @@ -383,7 +407,10 @@ def client_update_value( response = None if self.setter_callback: # pylint: disable=not-callable - response = self.setter_callback(value) + if self._setter_wants_addr(): + response = self.setter_callback(value, sender_client_addr) + else: + response = self.setter_callback(value) changed = self._value != previous_value if changed: self.notify(sender_client_addr) @@ -391,6 +418,21 @@ def client_update_value( self.value = None return response + def _setter_wants_addr(self) -> bool: + """Whether the current setter opts into the sender's client address. + + Signature introspection is done once per setter object and cached, so + the common write path does not call :func:`inspect.signature` on every + client write. + """ + setter = self.setter_callback + if setter is not self._setter_addr_cache_for: + self._setter_addr_cache_for = setter + self._setter_addr_cache = setter is not None and _setter_wants_client_addr( + setter + ) + return self._setter_addr_cache + def notify(self, sender_client_addr: Optional[Tuple[str, int]] = None) -> None: """Notify clients about a value change. Sends the value. diff --git a/pyhap/hap_protocol.py b/pyhap/hap_protocol.py index 0fda6995..d89f7a6a 100644 --- a/pyhap/hap_protocol.py +++ b/pyhap/hap_protocol.py @@ -291,6 +291,11 @@ def _process_response(self, response: HAPResponse) -> None: # If we get a shared key, upgrade to encrypted if response.shared_key: self.hap_crypto = HAPCrypto(response.shared_key) + # Publish the session key so transports derived from this session + # (e.g. HomeKit Data Stream) can be set up from a characteristic write. + self.accessory_driver.session_shared_keys[self.peername] = ( + response.shared_key + ) # Only update mDNS after sending the response if response.pairing_changed: async_create_background_task( diff --git a/pyhap/hds.py b/pyhap/hds.py new file mode 100644 index 00000000..09594e70 --- /dev/null +++ b/pyhap/hds.py @@ -0,0 +1,185 @@ +"""HomeKit Data Stream (HDS) transport primitives. + +HDS is the TCP side channel HomeKit Secure Video uses to move fMP4 recording +fragments (and other bulk payloads) off the HAP control connection. A controller +sets up the transport with SetupDataStreamTransport, connects to the advertised +TCP port and exchanges frames encrypted with keys derived from the HAP session's +shared secret plus a per-transport salt. + +This module is protocol-only: key derivation, frame encryption/decryption and +the setup-request/response TLV shapes. The listener and the higher-level HDS +protocol (control/dataSend topics) build on top of it. +""" + +from __future__ import annotations + +from dataclasses import dataclass +import os +import struct +from typing import Optional, Tuple + +from cryptography.hazmat.primitives.ciphers.aead import ChaCha20Poly1305 + +from pyhap import tlv +from pyhap.hap_crypto import hap_hkdf + +# --- Setup Data Stream Transport (HAP R17 section 12) --- + +TRANSPORT_TYPE_TCP = b"\x00" + +# SetupDataStreamTransport write (Transfer Transport Configuration). +SETUP_TYPES = { + "SESSION_COMMAND_TYPE": b"\x01", + "TRANSPORT_TYPE": b"\x02", + "CONTROLLER_KEY_SALT": b"\x03", +} + +SESSION_COMMAND_START = b"\x00" + +# SetupDataStreamTransport response. +SETUP_RESPONSE_TYPES = { + "STATUS": b"\x01", + "TRANSPORT_TYPE_SESSION_PARAMETERS": b"\x02", + "ACCESSORY_KEY_SALT": b"\x03", +} + +TRANSPORT_SESSION_PARAM_TCP_LISTENING_PORT = b"\x01" + +SETUP_STATUS_SUCCESS = b"\x00" +SETUP_STATUS_GENERIC_ERROR = b"\x01" +SETUP_STATUS_BUSY = b"\x02" + +# HDS derives its two directional keys from the HAP session shared secret salted +# with both key salts. The info strings are named from the controller's point of +# view: the controller READS (decrypts) what the accessory encrypts, so the +# accessory encrypts outgoing frames with the "Read" key and decrypts incoming +# frames with the "Write" key. +_KEY_INFO_ACCESSORY_ENCRYPT = b"HDS-Read-Encryption-Key" +_KEY_INFO_ACCESSORY_DECRYPT = b"HDS-Write-Encryption-Key" + +_NONCE_LENGTH = 12 +_TAG_LENGTH = 16 +_FRAME_HEADER_LENGTH = 4 +# The 24-bit length field allows up to 16 MiB, but real HDS frames are far +# smaller (recording chunks are 256 KiB, control frames tiny). Cap well above +# that so both the encoder and the decoder reject/bound anything larger, +# limiting how much a peer can make us buffer for a single frame. +_MAX_PAYLOAD_LENGTH = 0x100000 # 1 MiB + + +@dataclass +class SetupRequest: + """Decoded SetupDataStreamTransport controller request.""" + + controller_key_salt: bytes + transport_type: bytes = TRANSPORT_TYPE_TCP + session_command: bytes = SESSION_COMMAND_START + + @classmethod + def decode(cls, data: bytes) -> "SetupRequest": + objs = tlv.decode(data) + return cls( + controller_key_salt=objs[SETUP_TYPES["CONTROLLER_KEY_SALT"]], + transport_type=objs.get(SETUP_TYPES["TRANSPORT_TYPE"], TRANSPORT_TYPE_TCP), + session_command=objs.get( + SETUP_TYPES["SESSION_COMMAND_TYPE"], SESSION_COMMAND_START + ), + ) + + +def encode_setup_response( + listening_port: int, + accessory_key_salt: bytes, + status: bytes = SETUP_STATUS_SUCCESS, +) -> bytes: + """Encode a SetupDataStreamTransport response advertising the TCP port.""" + if status != SETUP_STATUS_SUCCESS: + return tlv.encode(SETUP_RESPONSE_TYPES["STATUS"], status) + session_params = tlv.encode( + TRANSPORT_SESSION_PARAM_TCP_LISTENING_PORT, + struct.pack(" Tuple[bytes, bytes]: + """Derive the (encrypt, decrypt) HDS keys from the HAP session shared secret. + + ``encrypt`` seals accessory->controller frames; ``decrypt`` opens + controller->accessory frames. The salt is the concatenation of the two key + salts, matching the controller's derivation. + """ + salt = controller_key_salt + accessory_key_salt + encrypt_key = hap_hkdf(shared_secret, salt, _KEY_INFO_ACCESSORY_ENCRYPT) + decrypt_key = hap_hkdf(shared_secret, salt, _KEY_INFO_ACCESSORY_DECRYPT) + return encrypt_key, decrypt_key + + +def new_key_salt() -> bytes: + """Return a fresh 32-byte accessory key salt.""" + return os.urandom(32) + + +def _nonce(counter: int) -> bytes: + # 96-bit nonce: four zero bytes then the 64-bit little-endian counter + # (right-justified), the same layout the HAP control channel uses. + return b"\x00\x00\x00\x00" + struct.pack(" None: + self._encrypt_cipher = ChaCha20Poly1305(encrypt_key) + self._decrypt_cipher = ChaCha20Poly1305(decrypt_key) + self._encrypt_count = 0 + self._decrypt_count = 0 + + def encrypt_frame(self, payload: bytes, frame_type: int = 1) -> bytes: + """Encrypt a payload into a full HDS frame (header + ciphertext + tag).""" + if len(payload) > _MAX_PAYLOAD_LENGTH: + raise ValueError("HDS payload too large for a single frame") + header = bytes([frame_type]) + len(payload).to_bytes(3, "big") + nonce = _nonce(self._encrypt_count) + self._encrypt_count += 1 + ciphertext = self._encrypt_cipher.encrypt(nonce, payload, header) + return header + ciphertext + + def decrypt_frame(self, buffer: bytearray) -> Optional[bytes]: + """Decrypt one complete frame from the front of ``buffer``. + + Returns the plaintext payload and removes the frame from ``buffer``, or + ``None`` when a full frame is not yet buffered (leaving ``buffer`` + untouched so the caller can retry once more data arrives). + """ + if len(buffer) < _FRAME_HEADER_LENGTH: + return None + header = bytes(buffer[:_FRAME_HEADER_LENGTH]) + payload_length = int.from_bytes(header[1:4], "big") + if payload_length > _MAX_PAYLOAD_LENGTH: + # A hostile/broken peer must not be able to make us buffer an + # arbitrarily large frame; reject well beyond any real HDS frame. + raise ValueError("HDS frame exceeds the maximum payload length") + frame_length = _FRAME_HEADER_LENGTH + payload_length + _TAG_LENGTH + if len(buffer) < frame_length: + return None + ciphertext = bytes(buffer[_FRAME_HEADER_LENGTH:frame_length]) + nonce = _nonce(self._decrypt_count) + plaintext = self._decrypt_cipher.decrypt(nonce, ciphertext, header) + self._decrypt_count += 1 + del buffer[:frame_length] + return plaintext diff --git a/pyhap/hds_protocol.py b/pyhap/hds_protocol.py new file mode 100644 index 00000000..80832a6c --- /dev/null +++ b/pyhap/hds_protocol.py @@ -0,0 +1,307 @@ +"""HomeKit Data Stream message protocol (payload codec + message framing). + +Inside each decrypted HDS frame HomeKit carries a binary-serialized header +dictionary and a message dictionary. The serialization is a compact, +self-describing format (a length byte / tag per value); this module implements +it in pure Python together with the request/response/event message layer that +rides on top of :mod:`pyhap.hds`. + +The byte-level format is not published by Apple; the tag values below match the +de-facto reference behaviour so the encoding is wire-compatible with HomeKit +controllers. +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import IntEnum +import struct +from typing import Any, Optional + + +class HDSStatus(IntEnum): + """Header status of an HDS response message.""" + + SUCCESS = 0 + OUT_OF_MEMORY = 1 + TIMEOUT = 2 + HEADER_ERROR = 3 + PAYLOAD_ERROR = 4 + MISSING_PROTOCOL = 5 + PROTOCOL_SPECIFIC_ERROR = 6 + + +class Int64(int): + """An integer that always serializes as a 64-bit HDS value. + + The message header ``id`` and ``status`` are wire-encoded as 64-bit integers + regardless of magnitude; wrapping them keeps that exact-width encoding. + """ + + +# --- payload format tags --- + +_TRUE = 0x01 +_FALSE = 0x02 +_TERMINATOR = 0x03 +_NULL = 0x04 +_UUID = 0x05 +_DATE = 0x06 +_INT_MINUS_ONE = 0x07 +_INT_0 = 0x08 # 0x08..0x2E encode 0..38 inline +_INT_38 = 0x2E +_INT8 = 0x30 +_INT16 = 0x31 +_INT32 = 0x32 +_INT64 = 0x33 +_FLOAT32 = 0x35 +_FLOAT64 = 0x36 +_UTF8_0 = 0x40 # 0x40..0x60 encode utf8 of length 0..32 inline +_UTF8_32 = 0x60 +_UTF8_LEN8 = 0x61 +_UTF8_LEN16 = 0x62 +_UTF8_LEN32 = 0x63 +_UTF8_LEN64 = 0x64 +_DATA_0 = 0x70 # 0x70..0x90 encode data of length 0..32 inline +_DATA_32 = 0x90 +_DATA_LEN8 = 0x91 +_DATA_LEN16 = 0x92 +_DATA_LEN32 = 0x93 +_DATA_LEN64 = 0x94 +_ARRAY_0 = 0xD0 # 0xD0..0xDE encode 0..14 elements inline +_ARRAY_14 = 0xDE +_ARRAY_TERMINATED = 0xDF +_DICT_0 = 0xE0 # 0xE0..0xEE encode 0..14 pairs inline +_DICT_14 = 0xEE +_DICT_TERMINATED = 0xEF + + +def _encode_int(value: int) -> bytes: + if isinstance(value, Int64): + return bytes([_INT64]) + struct.pack(" bytes: + length = len(value) + if length <= 32: + return bytes([base + length]) + value + if length <= 0xFF: + return bytes([len8]) + struct.pack(" bytes: + """Serialize a python value into the HDS payload format.""" + if value is True: + return bytes([_TRUE]) + if value is False: + return bytes([_FALSE]) + if value is None: + return bytes([_NULL]) + if isinstance(value, int): + return _encode_int(value) + if isinstance(value, float): + return bytes([_FLOAT64]) + struct.pack(" None: + self._data = data + self._offset = 0 + + def _take(self, length: int) -> bytes: + chunk = self._data[self._offset : self._offset + length] + if len(chunk) != length: + raise ValueError("Truncated HDS payload") + self._offset += length + return chunk + + def _read_length(self, size: int) -> int: + return int.from_bytes(self._take(size), "little") + + def decode(self, depth: int = 0) -> Any: + if depth > _MAX_NESTING_DEPTH: + # A hostile peer must not be able to blow the Python stack with + # deeply nested containers (RecursionError). + raise ValueError("HDS payload nesting too deep") + tag = self._take(1)[0] + if tag == _TRUE: + return True + if tag == _FALSE: + return False + if tag == _NULL: + return None + if tag == _INT_MINUS_ONE: + return -1 + if _INT_0 <= tag <= _INT_38: + return tag - _INT_0 + if tag == _INT8: + return struct.unpack(" int: + return self._offset + + +def decode(data: bytes) -> Any: + """Deserialize a single value from an HDS payload buffer.""" + return _Decoder(data).decode() + + +# --- message layer --- + +REQUEST = "request" +RESPONSE = "response" +EVENT = "event" + + +@dataclass +class Message: + """A decoded HDS protocol message (header + message dictionaries).""" + + protocol: str + kind: str # REQUEST / RESPONSE / EVENT + topic: str + message: dict = field(default_factory=dict) + id: Optional[int] = None + status: Optional[int] = None + + def encode(self) -> bytes: + """Return the plaintext HDS frame payload for this message. + + The frame payload is a header-length byte, the serialized header + dictionary and the serialized message dictionary. + """ + header: dict = {"protocol": self.protocol, self.kind: self.topic} + if self.id is not None: + header["id"] = Int64(self.id) + if self.status is not None: + header["status"] = Int64(self.status) + header_bytes = encode(header) + if len(header_bytes) > 0xFF: + raise ValueError("HDS message header too large") + return bytes([len(header_bytes)]) + header_bytes + encode(self.message) + + @classmethod + def decode(cls, payload: bytes) -> "Message": + header_length = payload[0] + header = decode(payload[1 : 1 + header_length]) + message = decode(payload[1 + header_length :]) + for kind in (REQUEST, RESPONSE, EVENT): + if kind in header: + topic = header[kind] + break + else: + raise ValueError("HDS message header has no request/response/event") + return cls( + protocol=header["protocol"], + kind=kind, + topic=topic, + message=message, + id=header.get("id"), + status=header.get("status"), + ) + + +def request(protocol: str, topic: str, message: dict, request_id: int) -> Message: + """Build a request message.""" + return Message(protocol, REQUEST, topic, message, id=request_id) + + +def response( + protocol: str, topic: str, message: dict, request_id: int, status: int = 0 +) -> Message: + """Build a response message.""" + return Message(protocol, RESPONSE, topic, message, id=request_id, status=status) + + +def event(protocol: str, topic: str, message: dict) -> Message: + """Build an event message.""" + return Message(protocol, EVENT, topic, message) diff --git a/pyhap/hds_recording.py b/pyhap/hds_recording.py new file mode 100644 index 00000000..866fc53a --- /dev/null +++ b/pyhap/hds_recording.py @@ -0,0 +1,262 @@ +"""Classic HomeKit Secure Video recording transfer over HDS ``dataSend``. + +Once a controller has set up an HDS connection and completed the ``control`` +handshake, it opens a ``dataSend`` stream of type ``ipcamera.recording``. The +accessory answers and then streams fragmented-MP4 recording fragments as +``dataSend``/``data`` events: the first fragment is the MP4 initialization +segment (``moov``), the rest are media fragments. Fragments larger than a chunk +are split across several events and reassembled by the controller using the +per-chunk metadata. + +The fragments themselves come from a delegate: an async generator yielding +:class:`RecordingPacket` objects. This module owns the protocol; producing the +MP4 bytes is the accessory's job. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +from enum import IntEnum +import logging +from typing import AsyncIterator, Callable, Optional + +from pyhap.hds_protocol import HDSStatus, Message + +logger = logging.getLogger(__name__) + +_DATA_SEND = "dataSend" +_TYPE_RECORDING = "ipcamera.recording" +_TARGET_CONTROLLER = "controller" + +# The controller reassembles chunks; this only bounds a single HDS frame. +DEFAULT_CHUNK_SIZE = 0x40000 + +MEDIA_INITIALIZATION = "mediaInitialization" +MEDIA_FRAGMENT = "mediaFragment" + + +class RecordingReason(IntEnum): + NORMAL = 0 + NOT_ALLOWED = 1 + BUSY = 2 + CANCELLED = 3 + UNSUPPORTED = 4 + UNEXPECTED_FAILURE = 5 + TIMEOUT = 6 + BAD_DATA = 7 + PROTOCOL_ERROR = 8 + INVALID_CONFIGURATION = 9 + + +@dataclass +class RecordingPacket: + """One recording fragment produced by the delegate.""" + + data: bytes + is_last: bool = False + + +# A delegate is called with the stream id and yields RecordingPackets. The first +# yielded packet is the MP4 initialization segment. +RecordingDelegate = Callable[[int], AsyncIterator[RecordingPacket]] + + +class RecordingStreamManager: + """Handle ``dataSend`` recording streams on one HDS connection.""" + + def __init__( + self, + connection, + delegate: RecordingDelegate, + chunk_size: int = DEFAULT_CHUNK_SIZE, + ) -> None: + self._connection = connection + self._delegate = delegate + self._chunk_size = chunk_size + self._stream_id: Optional[int] = None + self._task: Optional[asyncio.Task] = None + self._closed = False + self._stopping = False + connection.add_request_handler(_DATA_SEND, "open", self._handle_open) + connection.add_request_handler(_DATA_SEND, "close", self._handle_close) + connection.add_request_handler(_DATA_SEND, "ack", self._handle_ack) + # Stop the recording task (and its delegate/subprocess) if the controller + # drops the connection instead of sending a clean dataSend/close. + connection.add_close_callback(lambda _connection: self._stop()) + + @staticmethod + def _reject(reason: RecordingReason): + # A rejected request carries HDSStatus.PROTOCOL_SPECIFIC_ERROR in the + # header and the specific reason in the message body. + return HDSStatus.PROTOCOL_SPECIFIC_ERROR, {"status": reason} + + def _handle_open(self, message: Message): + body = message.message + if not isinstance(body, dict) or "streamId" not in body: + return self._reject(RecordingReason.UNEXPECTED_FAILURE) + if ( + body.get("target") != _TARGET_CONTROLLER + or body.get("type") != _TYPE_RECORDING + ): + return self._reject(RecordingReason.UNEXPECTED_FAILURE) + if self._task is not None and not self._task.done(): + return self._reject(RecordingReason.BUSY) + self._stream_id = body["streamId"] + self._closed = False + self._stopping = False + self._task = asyncio.get_running_loop().create_task(self._stream()) + return HDSStatus.SUCCESS, {} + + def _handle_close(self, message: Message): + body = message.message + # A stale or duplicate close for an earlier stream must not stop the + # currently running one. + if isinstance(body, dict) and body.get("streamId") != self._stream_id: + return self._reject(RecordingReason.UNEXPECTED_FAILURE) + self._stop() + return HDSStatus.SUCCESS, {} + + def _handle_ack(self, message: Message): + return HDSStatus.SUCCESS, {} + + def _stop(self) -> None: + # Cancel at most once: _stop() is reached from both dataSend/close and + # connection loss, and a second cancel() landing inside the stream's + # `finally: await generator.aclose()` would abort the delegate's own + # cleanup (e.g. reaping its ffmpeg subprocess) with a CancelledError. + self._closed = True + task = self._task + if task is not None and not task.done() and not self._stopping: + self._stopping = True + task.cancel() + + async def _stream(self) -> None: + generator = self._delegate(self._stream_id) + try: + sequence_number = 1 + async for packet in generator: + if self._closed: + break + await self._send_fragment( + packet.data, + sequence_number, + initialization=sequence_number == 1, + is_last=packet.is_last, + ) + if packet.is_last: + break + sequence_number += 1 + except Exception: # pylint: disable=broad-exception-caught + logger.warning("Recording stream %s failed", self._stream_id, exc_info=True) + finally: + # Deterministically run the delegate's cleanup (e.g. terminate its + # ffmpeg subprocess) even when the controller dropped the connection. + aclose = getattr(generator, "aclose", None) + if aclose is not None: + try: + await aclose() + except Exception: # pylint: disable=broad-exception-caught + pass + + async def _send_fragment( + self, fragment: bytes, sequence_number: int, initialization: bool, is_last: bool + ) -> None: + offset = 0 + chunk_sequence_number = 1 + total = len(fragment) + # A zero-length fragment still needs a single event to carry its metadata. + while True: + # Respect transport write back-pressure so a slow controller cannot + # make us buffer an entire multi-megabyte fragment in memory. + await self._connection.drain() + if self._closed: + return + chunk = fragment[offset : offset + self._chunk_size] + offset += len(chunk) + last_chunk = offset >= total + metadata = { + "dataType": MEDIA_INITIALIZATION if initialization else MEDIA_FRAGMENT, + "dataSequenceNumber": sequence_number, + "dataChunkSequenceNumber": chunk_sequence_number, + "isLastDataChunk": last_chunk, + } + if chunk_sequence_number == 1: + metadata["dataTotalSize"] = total + event = { + "streamId": self._stream_id, + "packets": [{"data": chunk, "metadata": metadata}], + } + if last_chunk and is_last: + event["endOfStream"] = True + self._connection.send_event(_DATA_SEND, "data", event) + chunk_sequence_number += 1 + if last_chunk: + break + + +async def _read_mp4_box(reader: asyncio.StreamReader) -> Optional[bytes]: + """Read one top-level MP4 box (header + body) from ``reader``. + + Returns ``None`` at end of stream. + """ + try: + header = await reader.readexactly(8) + except asyncio.IncompleteReadError: + return None + size = int.from_bytes(header[0:4], "big") + if size == 1: # 64-bit extended size + ext = await reader.readexactly(8) + size = int.from_bytes(ext, "big") + body = await reader.readexactly(size - 16) + return header + ext + body + if size < 8: + return None + body = await reader.readexactly(size - 8) + return header + body + + +async def fmp4_recording_packets( + reader: asyncio.StreamReader, +) -> AsyncIterator["RecordingPacket"]: + """Turn a fragmented-MP4 byte stream into HKSV recording packets. + + Reads MP4 boxes from ``reader`` - any object with an ``readexactly`` + coroutine, e.g. an ffmpeg subprocess' ``stdout`` or an aiohttp/go2rtc + response body. The leading ``ftyp``+``moov`` boxes are yielded as the single + initialization packet; each following ``moof``(+``mdat``) fragment is + yielded as a media fragment, and the fragment produced right before end of + stream is flagged ``is_last`` so the recording ends gracefully. + + This is source-agnostic: point it at whatever produces keyframe-aligned + fragmented MP4 whose fragment length matches the selected recording + configuration. It never owns the source - the caller starts and stops it + (e.g. terminates the ffmpeg subprocess in its own ``finally``), which the + recording stream lifecycle already drives via the delegate's ``aclose()``. + """ + init = bytearray() + sent_init = False + pending: Optional[bytearray] = None + while True: + box = await _read_mp4_box(reader) + if box is None: + break + box_type = bytes(box[4:8]) + if box_type in (b"ftyp", b"moov"): + init += box + elif box_type == b"moof": + if not sent_init: + yield RecordingPacket(bytes(init)) + sent_init = True + # A new fragment starts; flush the previous one (not the last). + if pending is not None: + yield RecordingPacket(bytes(pending)) + pending = bytearray(box) + elif pending is not None: + # mdat (and any styp / other boxes) belong to the current fragment. + pending += box + elif not sent_init: + init += box + # End of stream: the buffered fragment is the last one. + if pending is not None: + yield RecordingPacket(bytes(pending), is_last=True) diff --git a/pyhap/hds_server.py b/pyhap/hds_server.py new file mode 100644 index 00000000..67499afa --- /dev/null +++ b/pyhap/hds_server.py @@ -0,0 +1,307 @@ +"""HomeKit Data Stream TCP listener and connection dispatch. + +Ties the HDS transport crypto (:mod:`pyhap.hds`) and message codec +(:mod:`pyhap.hds_protocol`) into an asyncio TCP server. A controller sets up a +transport over HAP (SetupDataStreamTransport), connects to the advertised port +and completes a ``control``/``hello`` handshake; from there both sides exchange +request/response/event messages on named protocols (e.g. ``dataSend`` for +recording fragment transfer). + +The listener does not know the HAP session shared secret on its own; the camera +accessory registers a transport with the secret when it answers +SetupDataStreamTransport, and the first frame from the controller is matched +against the registered transports by trial decryption. +""" + +from __future__ import annotations + +import asyncio +from dataclasses import dataclass +import logging +from typing import Callable, Dict, List, Optional, Tuple + +from pyhap import hds +from pyhap.hds_protocol import EVENT, REQUEST, RESPONSE, HDSStatus, Message + +logger = logging.getLogger(__name__) + +RequestHandler = Callable[[Message], Tuple[int, dict]] # returns (HDSStatus, body) +EventHandler = Callable[[Message], None] + +# A connection that has not completed the control handshake buffers only the +# tiny hello frame; cap it hard and time it out so an unauthenticated peer +# cannot make us buffer indefinitely. +_MAX_UNBOUND_BYTES = 4096 +_BIND_TIMEOUT = 10.0 +# Controllers re-run SetupDataStreamTransport and may abandon transports; bound +# the registrations we keep (and trial-decrypt against) so they cannot pile up. +_MAX_PENDING_TRANSPORTS = 16 + + +@dataclass +class _PendingTransport: + encrypt_key: bytes + decrypt_key: bytes + + +class HDSConnection(asyncio.Protocol): + """One HDS TCP connection: key binding, framing and message dispatch.""" + + def __init__(self, listener: "HDSListener") -> None: + self._listener = listener + self._buffer = bytearray() + self._crypto: Optional[hds.HDSCrypto] = None + self._transport: Optional[asyncio.Transport] = None + self._request_handlers: Dict[Tuple[str, str], RequestHandler] = {} + self._event_handlers: Dict[Tuple[str, str], EventHandler] = {} + self._pending_requests: Dict[int, asyncio.Future] = {} + self._next_request_id = 1 + self._bind_timeout_handle: Optional[asyncio.TimerHandle] = None + self._close_callbacks: List[Callable[["HDSConnection"], None]] = [] + # Writable unless the transport has asked us to pause (back-pressure). + self._can_write = asyncio.Event() + self._can_write.set() + # The mandatory control handshake is answered by default. + self.add_request_handler( + "control", "hello", lambda message: (HDSStatus.SUCCESS, {}) + ) + + # --- registration API --- + + def add_request_handler( + self, protocol: str, topic: str, handler: RequestHandler + ) -> None: + """Register a handler returning ``(status, response_message)``.""" + self._request_handlers[(protocol, topic)] = handler + + def add_event_handler( + self, protocol: str, topic: str, handler: EventHandler + ) -> None: + """Register a handler for a fire-and-forget event.""" + self._event_handlers[(protocol, topic)] = handler + + def add_close_callback(self, callback: Callable[["HDSConnection"], None]) -> None: + """Register a callback invoked once when the connection is lost. + + Several owners (the listener's connection set, any recording stream + manager) need to react to connection loss, so this is a list rather than + a single slot. + """ + self._close_callbacks.append(callback) + + # --- send API --- + + def send_event(self, protocol: str, topic: str, message: dict) -> None: + """Send an event message (no response expected).""" + self._send(Message(protocol, EVENT, topic, message)) + + def send_request( + self, protocol: str, topic: str, message: dict + ) -> "asyncio.Future[Message]": + """Send a request; the future resolves with the controller's response.""" + request_id = self._next_request_id + self._next_request_id += 1 + future: asyncio.Future = asyncio.get_running_loop().create_future() + self._pending_requests[request_id] = future + self._send(Message(protocol, REQUEST, topic, message, id=request_id)) + return future + + def close(self) -> None: + if self._transport is not None: + self._transport.close() + + # --- asyncio.Protocol --- + + def connection_made(self, transport: asyncio.BaseTransport) -> None: + self._transport = transport # type: ignore[assignment] + self._bind_timeout_handle = asyncio.get_running_loop().call_later( + _BIND_TIMEOUT, self._on_bind_timeout + ) + + def _on_bind_timeout(self) -> None: + if self._crypto is None: + logger.warning("Closing HDS connection: control handshake timed out") + self.close() + + def data_received(self, data: bytes) -> None: + self._buffer += data + if self._crypto is None: + if len(self._buffer) > _MAX_UNBOUND_BYTES: + logger.warning( + "Closing HDS connection: no valid handshake within %d bytes", + _MAX_UNBOUND_BYTES, + ) + self.close() + return + if not self._bind(): + return + self._drain_frames() + + def pause_writing(self) -> None: + # Transport buffer above the high-water mark: stop producing. + self._can_write.clear() + + def resume_writing(self) -> None: + self._can_write.set() + + async def drain(self) -> None: + """Wait until the transport can accept more data (write back-pressure).""" + await self._can_write.wait() + + def connection_lost(self, exc: Optional[Exception]) -> None: + if self._bind_timeout_handle is not None: + self._bind_timeout_handle.cancel() + self._bind_timeout_handle = None + # Unblock anyone waiting on write back-pressure so their task can unwind. + self._can_write.set() + for future in self._pending_requests.values(): + if not future.done(): + future.cancel() + self._pending_requests.clear() + self._listener._connections.discard(self) # pylint: disable=protected-access + for callback in self._close_callbacks: + callback(self) + self._close_callbacks.clear() + + # --- internals --- + + def _bind(self) -> bool: + """Identify the transport by trial-decrypting the first frame.""" + if len(self._buffer) < 4: + return False + # The listener and its connections cooperate within this module. + # pylint: disable=protected-access + for pending in list(self._listener._pending): + trial = hds.HDSCrypto(pending.encrypt_key, pending.decrypt_key) + probe = bytearray(self._buffer) + try: + payload = trial.decrypt_frame(probe) + # Wrong keys fail the auth tag; try the next pending transport. + except Exception: # pylint: disable=broad-exception-caught + continue + if payload is None: + # Right keys but the frame is not complete yet; wait for more. + return False + self._crypto = trial + self._buffer = probe + if self._bind_timeout_handle is not None: + self._bind_timeout_handle.cancel() + self._bind_timeout_handle = None + self._listener._pending.remove(pending) + self._listener._connections.add(self) + self._dispatch(payload) + return True + return False + + def _drain_frames(self) -> None: + while True: + try: + payload = self._crypto.decrypt_frame(self._buffer) + except Exception: # pylint: disable=broad-exception-caught + logger.warning("Dropping HDS connection on frame decrypt failure") + self.close() + return + if payload is None: + return + self._dispatch(payload) + + def _dispatch(self, payload: bytes) -> None: + try: + message = Message.decode(payload) + except (ValueError, IndexError, KeyError, TypeError): + logger.warning("Ignoring malformed HDS message") + return + if message.kind == REQUEST: + self._handle_request(message) + elif message.kind == RESPONSE: + future = self._pending_requests.pop(message.id, None) + if future is not None and not future.done(): + future.set_result(message) + elif message.kind == EVENT: + handler = self._event_handlers.get((message.protocol, message.topic)) + if handler is not None: + handler(message) + + def _handle_request(self, message: Message) -> None: + handler = self._request_handlers.get((message.protocol, message.topic)) + if handler is None: + status, response_message = HDSStatus.PROTOCOL_SPECIFIC_ERROR, {} + else: + try: + status, response_message = handler(message) + except Exception: # pylint: disable=broad-exception-caught + logger.warning( + "HDS request handler for %s/%s failed", + message.protocol, + message.topic, + exc_info=True, + ) + status, response_message = HDSStatus.PROTOCOL_SPECIFIC_ERROR, {} + self._send( + Message( + message.protocol, + RESPONSE, + message.topic, + response_message, + id=message.id, + status=status, + ) + ) + + def _send(self, message: Message) -> None: + if self._transport is None or self._crypto is None: + raise RuntimeError("HDS connection is not ready to send") + self._transport.write(self._crypto.encrypt_frame(message.encode())) + + +class HDSListener: + """An asyncio TCP server accepting HDS connections.""" + + def __init__(self) -> None: + self._server: Optional[asyncio.AbstractServer] = None + self._pending = [] + self._connections = set() + self.on_connection: Optional[Callable[[HDSConnection], None]] = None + + async def start(self, host: str = "0.0.0.0") -> int: + """Bind the server to an ephemeral port and return it.""" + loop = asyncio.get_running_loop() + self._server = await loop.create_server(self._make_connection, host, 0) + return self.port + + def _make_connection(self) -> HDSConnection: + connection = HDSConnection(self) + if self.on_connection is not None: + self.on_connection(connection) + return connection + + @property + def port(self) -> int: + return self._server.sockets[0].getsockname()[1] + + def register_transport( + self, shared_secret: bytes, controller_key_salt: bytes + ) -> bytes: + """Register an expected transport, returning the accessory key salt. + + The camera calls this when it answers SetupDataStreamTransport, passing + the HAP session shared secret and the controller's key salt. The + returned accessory key salt goes into the setup response. + """ + accessory_key_salt = hds.new_key_salt() + encrypt_key, decrypt_key = hds.derive_keys( + shared_secret, controller_key_salt, accessory_key_salt + ) + self._pending.append(_PendingTransport(encrypt_key, decrypt_key)) + # Evict the oldest stale registration if the controller keeps setting up + # transports it never connects to. + while len(self._pending) > _MAX_PENDING_TRANSPORTS: + self._pending.pop(0) + return accessory_key_salt + + async def stop(self) -> None: + for connection in list(self._connections): + connection.close() + if self._server is not None: + self._server.close() + await self._server.wait_closed() diff --git a/pyhap/hksv_recording.py b/pyhap/hksv_recording.py new file mode 100644 index 00000000..59a0a2b1 --- /dev/null +++ b/pyhap/hksv_recording.py @@ -0,0 +1,364 @@ +"""TLV codecs for classic HomeKit Secure Video recording configuration. + +The Camera Recording Management service advertises the container, video and +audio configurations the accessory supports and receives the controller's +selection, using the standard HAP TLV8 layout (repeated items separated by a +zero-length type-0 TLV). +""" + +from __future__ import annotations + +from dataclasses import dataclass, field +from enum import IntEnum +import struct +from typing import List, Union + +from pyhap import tlv + +_SEPARATOR = b"\x00\x00" + + +class EventTrigger(IntEnum): + MOTION = 0x01 + DOORBELL = 0x02 + + +class MediaContainerType(IntEnum): + FRAGMENTED_MP4 = 0x00 + + +class VideoCodecType(IntEnum): + H264 = 0x00 + H265 = 0x01 + + +class AudioCodecType(IntEnum): + AAC_LC = 0x00 + AAC_ELD = 0x01 + + +class AudioSampleRate(IntEnum): + KHZ_8 = 0x00 + KHZ_16 = 0x01 + KHZ_24 = 0x02 + KHZ_32 = 0x03 + KHZ_44_1 = 0x04 + KHZ_48 = 0x05 + + +class BitRateMode(IntEnum): + VARIABLE = 0x00 + CONSTANT = 0x01 + + +def _u8(value: int) -> bytes: + return struct.pack(" bytes: + return struct.pack(" bytes: + return struct.pack(" int: + return int.from_bytes(data, "little") + + +def _join(items: List[bytes]) -> bytes: + return _SEPARATOR.join(items) + + +def _iter(data: bytes): + offset = 0 + while offset < len(data): + tag = data[offset] + length = data[offset + 1] + value = data[offset + 2 : offset + 2 + length] + offset += 2 + length + while length == 255 and offset < len(data) and data[offset] == tag: + length = data[offset + 1] + value += data[offset + 2 : offset + 2 + length] + offset += 2 + length + yield tag, value + + +def _decode(data: bytes) -> dict: + out: dict = {} + for tag, value in _iter(data): + out.setdefault(tag, value) + return out + + +def _split(data: bytes) -> list: + items = [] + current = b"" + for tag, value in _iter(data): + if tag == 0 and not value: + if current: + items.append(current) + current = b"" + continue + current += tlv.encode(_u8(tag), value) + if current: + items.append(current) + return items + + +@dataclass +class MediaContainerConfiguration: + fragment_length_ms: int = 4000 + container_type: MediaContainerType = MediaContainerType.FRAGMENTED_MP4 + + def encode(self) -> bytes: + params = tlv.encode(b"\x01", _u32(self.fragment_length_ms)) + return tlv.encode(b"\x01", _u8(self.container_type), b"\x02", params) + + @classmethod + def decode(cls, data: bytes) -> "MediaContainerConfiguration": + d = _decode(data) + params = _decode(d[2]) + return cls( + container_type=MediaContainerType(_int(d[1])), + fragment_length_ms=_int(params[1]), + ) + + +@dataclass +class SupportedRecordingConfiguration: + prebuffer_length_ms: int = 4000 + event_triggers: int = EventTrigger.MOTION + media_containers: List[MediaContainerConfiguration] = field( + default_factory=lambda: [MediaContainerConfiguration()] + ) + + def encode(self) -> bytes: + return tlv.encode( + b"\x01", + _u32(self.prebuffer_length_ms), + b"\x02", + struct.pack(" "SupportedRecordingConfiguration": + d = _decode(data) + return cls( + prebuffer_length_ms=_int(d[1]), + event_triggers=_int(d[2]), + media_containers=[ + MediaContainerConfiguration.decode(i) for i in _split(d[3]) + ], + ) + + +@dataclass +class VideoAttributes: + width: int + height: int + frame_rate: int + + def encode(self) -> bytes: + return tlv.encode( + b"\x01", + _u16(self.width), + b"\x02", + _u16(self.height), + b"\x03", + _u8(self.frame_rate), + ) + + @classmethod + def decode(cls, data: bytes) -> "VideoAttributes": + d = _decode(data) + return cls(width=_int(d[1]), height=_int(d[2]), frame_rate=_int(d[3])) + + +@dataclass +class VideoCodecConfiguration: + codec_type: VideoCodecType + profile: Union[int, List[int]] + level: Union[int, List[int]] + bitrate_kbps: int + iframe_interval_ms: int + attributes: List[VideoAttributes] + + def encode(self) -> bytes: + # Matches HAP-NodeJS' SupportedVideoRecordingConfiguration exactly: + # * one ProfileID (0x01) and one Level (0x02) TLV entry per supported + # value, with a zero-length 0x00 separator between consecutive + # entries of the SAME type (HAP list convention); + # * NO bitrate/iframe-interval in the *supported* advertisement (those + # only appear in the controller-written *selected* configuration); + # * each VideoAttributes as its OWN 0x03 TLV, separated by 0x00 — + # not bundled inside a single 0x03. + profiles = ( + self.profile if isinstance(self.profile, (list, tuple)) else [self.profile] + ) + levels = self.level if isinstance(self.level, (list, tuple)) else [self.level] + prof_bytes = _SEPARATOR.join(tlv.encode(b"\x01", _u8(p)) for p in profiles) + lvl_bytes = _SEPARATOR.join(tlv.encode(b"\x02", _u8(lvl)) for lvl in levels) + params = prof_bytes + lvl_bytes + attrs = _SEPARATOR.join( + tlv.encode(b"\x03", a.encode()) for a in self.attributes + ) + return ( + tlv.encode(b"\x01", _u8(self.codec_type)) + + tlv.encode(b"\x02", params) + + attrs + ) + + def encode_selected(self) -> bytes: + """Encode the single negotiated configuration the controller writes. + + One profile/level plus bitrate + iframe interval and one resolution. + """ + profile = ( + self.profile[0] if isinstance(self.profile, (list, tuple)) else self.profile + ) + level = self.level[0] if isinstance(self.level, (list, tuple)) else self.level + params = tlv.encode( + b"\x01", + _u8(profile), + b"\x02", + _u8(level), + b"\x03", + _u32(self.bitrate_kbps), + b"\x04", + _u32(self.iframe_interval_ms), + ) + return tlv.encode( + b"\x01", + _u8(self.codec_type), + b"\x02", + params, + b"\x03", + self.attributes[0].encode(), + ) + + @classmethod + def decode(cls, data: bytes) -> "VideoCodecConfiguration": + d = _decode(data) + # Collect ALL profile (0x01) and level (0x02) entries: the supported + # advertisement lists several (0x00-separated), the selected config a + # single one. A lone value decodes to a scalar, several to a list, so + # decode is the faithful inverse of encode in both directions. + profiles = [_int(v) for t, v in _iter(d[2]) if t == 1] + levels = [_int(v) for t, v in _iter(d[2]) if t == 2] + params = _decode(d[2]) + # Attributes are separate top-level 0x03 TLVs; gather them all. Bitrate + # (0x03) and iframe interval (0x04) are only present in the controller's + # SELECTED config, absent from the supported advertisement. + attributes = [VideoAttributes.decode(v) for t, v in _iter(data) if t == 3] + return cls( + codec_type=VideoCodecType(_int(d[1])), + profile=profiles[0] if len(profiles) == 1 else profiles, + level=levels[0] if len(levels) == 1 else levels, + bitrate_kbps=_int(params[3]) if 3 in params else 0, + iframe_interval_ms=_int(params[4]) if 4 in params else 0, + attributes=attributes, + ) + + +def encode_supported_video(configs: List[VideoCodecConfiguration]) -> bytes: + # Each configuration is its own 0x01 TLV; multiple configs are separated by + # a zero-length 0x00 TLV (a config's own body already contains 0x00 list + # separators, so we must not flatten them under a single 0x01). + return _SEPARATOR.join(tlv.encode(b"\x01", c.encode()) for c in configs) + + +def decode_supported_video(data: bytes) -> List[VideoCodecConfiguration]: + return [VideoCodecConfiguration.decode(v) for t, v in _iter(data) if t == 1] + + +@dataclass +class AudioCodecConfiguration: + codec_type: AudioCodecType = AudioCodecType.AAC_LC + channels: int = 1 + bitrate_mode: BitRateMode = BitRateMode.VARIABLE + sample_rate: AudioSampleRate = AudioSampleRate.KHZ_32 + max_audio_bitrate_kbps: int = 64 + + def encode(self) -> bytes: + # Matches HAP-NodeJS' SupportedAudioRecordingConfiguration: channels, + # bit-rate mode and sample rate only — no max-bitrate field. + params = tlv.encode( + b"\x01", + _u8(self.channels), + b"\x02", + _u8(self.bitrate_mode), + b"\x03", + _u8(self.sample_rate), + ) + return tlv.encode(b"\x01", _u8(self.codec_type), b"\x02", params) + + def encode_selected(self) -> bytes: + """Encode the single negotiated audio configuration. + + Unlike the supported advertisement it includes the max-bitrate field. + """ + params = tlv.encode( + b"\x01", + _u8(self.channels), + b"\x02", + _u8(self.bitrate_mode), + b"\x03", + _u8(self.sample_rate), + b"\x04", + _u32(self.max_audio_bitrate_kbps), + ) + return tlv.encode(b"\x01", _u8(self.codec_type), b"\x02", params) + + @classmethod + def decode(cls, data: bytes) -> "AudioCodecConfiguration": + d = _decode(data) + params = _decode(d[2]) + return cls( + codec_type=AudioCodecType(_int(d[1])), + channels=_int(params[1]), + bitrate_mode=BitRateMode(_int(params[2])), + sample_rate=AudioSampleRate(_int(params[3])), + max_audio_bitrate_kbps=_int(params[4]) if 4 in params else 0, + ) + + +def encode_supported_audio(configs: List[AudioCodecConfiguration]) -> bytes: + return _SEPARATOR.join(tlv.encode(b"\x01", c.encode()) for c in configs) + + +def decode_supported_audio(data: bytes) -> List[AudioCodecConfiguration]: + return [AudioCodecConfiguration.decode(v) for t, v in _iter(data) if t == 1] + + +@dataclass +class SelectedRecordingConfiguration: + """The controller's chosen recording configuration.""" + + recording: SupportedRecordingConfiguration + video: VideoCodecConfiguration + audio: AudioCodecConfiguration + + def encode(self) -> bytes: + # The selected configuration carries the single negotiated codec params + # (with bitrate/iframe), not the multi-value supported advertisement. + return tlv.encode( + b"\x01", + self.recording.encode(), + b"\x02", + self.video.encode_selected(), + b"\x03", + self.audio.encode_selected(), + ) + + @classmethod + def decode(cls, data: bytes) -> "SelectedRecordingConfiguration": + d = _decode(data) + return cls( + recording=SupportedRecordingConfiguration.decode(d[1]), + video=VideoCodecConfiguration.decode(d[2]), + audio=AudioCodecConfiguration.decode(d[3]), + ) diff --git a/pyhap/resources/characteristics.json b/pyhap/resources/characteristics.json index b14dc654..d539fd1f 100644 --- a/pyhap/resources/characteristics.json +++ b/pyhap/resources/characteristics.json @@ -122,6 +122,15 @@ "minValue": 0, "unit": "percentage" }, + "CameraOperatingModeIndicator": { + "Format": "bool", + "Permissions": [ + "pr", + "pw", + "ev" + ], + "UUID": "0000021D-0000-1000-8000-0026BB765291" + }, "CarbonDioxideDetected": { "Format": "uint8", "Permissions": [ @@ -482,6 +491,19 @@ ], "UUID": "00000136-0000-1000-8000-0026BB765291" }, + "EventSnapshotsActive": { + "Format": "uint8", + "Permissions": [ + "pr", + "pw", + "ev" + ], + "UUID": "00000223-0000-1000-8000-0026BB765291", + "ValidValues": { + "Disable": 0, + "Enable": 1 + } + }, "FilterChangeIndication": { "Format": "uint8", "Permissions": [ @@ -539,6 +561,20 @@ ], "UUID": "0000006F-0000-1000-8000-0026BB765291" }, + "HomeKitCameraActive": { + "Format": "uint8", + "Permissions": [ + "pr", + "pw", + "ev", + "tw" + ], + "UUID": "0000021B-0000-1000-8000-0026BB765291", + "ValidValues": { + "Disable": 0, + "Enable": 1 + } + }, "Hue": { "Format": "float", "Permissions": [ @@ -748,6 +784,14 @@ ], "UUID": "0000001F-0000-1000-8000-0026BB765291" }, + "ManuallyDisabled": { + "Format": "bool", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00000227-0000-1000-8000-0026BB765291" + }, "Manufacturer": { "Format": "string", "Permissions": [ @@ -916,6 +960,19 @@ ], "UUID": "00000050-0000-1000-8000-0026BB765291" }, + "PeriodicSnapshotsActive": { + "Format": "uint8", + "Permissions": [ + "pr", + "pw", + "ev" + ], + "UUID": "00000225-0000-1000-8000-0026BB765291", + "ValidValues": { + "Disable": 0, + "Enable": 1 + } + }, "PictureMode":{ "Format": "uint8", "Permissions": [ @@ -987,6 +1044,20 @@ "SinglePress": 0 } }, + "RecordingAudioActive": { + "Format": "uint8", + "Permissions": [ + "pr", + "pw", + "ev", + "tw" + ], + "UUID": "00000226-0000-1000-8000-0026BB765291", + "ValidValues": { + "Disable": 0, + "Enable": 1 + } + }, "RelativeHumidityDehumidifierThreshold": { "Format": "float", "Permissions": [ @@ -1138,6 +1209,15 @@ "StayArm": 0 } }, + "SelectedCameraRecordingConfiguration": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "ev" + ], + "UUID": "00000209-0000-1000-8000-0026BB765291" + }, "SelectedRTPStreamConfiguration": { "Format": "tlv8", "Permissions": [ @@ -1187,6 +1267,15 @@ "minStep": 1, "minValue": 0 }, + "SetupDataStreamTransport": { + "Format": "tlv8", + "Permissions": [ + "pr", + "pw", + "wr" + ], + "UUID": "00000131-0000-1000-8000-0026BB765291" + }, "SetupEndpoints": { "Format": "tlv8", "Permissions": [ @@ -1305,6 +1394,14 @@ "minStep": 1, "minValue": 0 }, + "SupportedAudioRecordingConfiguration": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00000207-0000-1000-8000-0026BB765291" + }, "SupportedAudioStreamConfiguration": { "Format": "tlv8", "Permissions": [ @@ -1312,6 +1409,21 @@ ], "UUID": "00000115-0000-1000-8000-0026BB765291" }, + "SupportedCameraRecordingConfiguration": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00000205-0000-1000-8000-0026BB765291" + }, + "SupportedDataStreamTransportConfiguration": { + "Format": "tlv8", + "Permissions": [ + "pr" + ], + "UUID": "00000130-0000-1000-8000-0026BB765291" + }, "SupportedRTPConfiguration": { "Format": "tlv8", "Permissions": [ @@ -1319,6 +1431,14 @@ ], "UUID": "00000116-0000-1000-8000-0026BB765291" }, + "SupportedVideoRecordingConfiguration": { + "Format": "tlv8", + "Permissions": [ + "pr", + "ev" + ], + "UUID": "00000206-0000-1000-8000-0026BB765291" + }, "SupportedVideoStreamConfiguration": { "Format": "tlv8", "Permissions": [ @@ -1573,6 +1693,15 @@ "Fahrenheit": 1 } }, + "ThirdPartyCameraActive": { + "Format": "bool", + "Permissions": [ + "pr", + "pw", + "ev" + ], + "UUID": "0000021C-0000-1000-8000-0026BB765291" + }, "TransitionControl": { "Format": "tlv8", "Permissions": [ @@ -1611,8 +1740,7 @@ "Format": "string", "MaximumLength": 64, "Permissions": [ - "pr", - "ev" + "pr" ], "UUID": "00000037-0000-1000-8000-0026BB765291" }, diff --git a/pyhap/resources/services.json b/pyhap/resources/services.json index 7c386279..34322521 100644 --- a/pyhap/resources/services.json +++ b/pyhap/resources/services.json @@ -61,8 +61,23 @@ ], "UUID": "00000096-0000-1000-8000-0026BB765291" }, + "CameraOperatingMode": { + "OptionalCharacteristics": [ + "CameraOperatingModeIndicator", + "ManuallyDisabled", + "NightVision", + "PeriodicSnapshotsActive", + "ThirdPartyCameraActive" + ], + "RequiredCharacteristics": [ + "EventSnapshotsActive", + "HomeKitCameraActive" + ], + "UUID": "0000021A-0000-1000-8000-0026BB765291" + }, "CameraRTPStreamManagement": { "OptionalCharacteristics": [ + "Active", "Name" ], "RequiredCharacteristics": [ @@ -75,6 +90,19 @@ ], "UUID": "00000110-0000-1000-8000-0026BB765291" }, + "CameraRecordingManagement": { + "OptionalCharacteristics": [ + "RecordingAudioActive" + ], + "RequiredCharacteristics": [ + "Active", + "SupportedCameraRecordingConfiguration", + "SupportedVideoRecordingConfiguration", + "SupportedAudioRecordingConfiguration", + "SelectedCameraRecordingConfiguration" + ], + "UUID": "00000204-0000-1000-8000-0026BB765291" + }, "CarbonDioxideSensor": { "OptionalCharacteristics": [ "StatusActive", @@ -118,6 +146,15 @@ ], "UUID": "00000080-0000-1000-8000-0026BB765291" }, + "DataStreamTransportManagement": { + "OptionalCharacteristics": [], + "RequiredCharacteristics": [ + "SetupDataStreamTransport", + "SupportedDataStreamTransportConfiguration", + "Version" + ], + "UUID": "00000129-0000-1000-8000-0026BB765291" + }, "Door": { "OptionalCharacteristics": [ "HoldPosition", diff --git a/tests/test_accessory.py b/tests/test_accessory.py index 5e766759..f6cfc9e5 100644 --- a/tests/test_accessory.py +++ b/tests/test_accessory.py @@ -303,7 +303,7 @@ def test_to_hap_bridge(mock_driver): { "format": "string", "iid": 9, - "perms": ["pr", "ev"], + "perms": ["pr"], "type": "37", "value": "01.01.00", } @@ -527,7 +527,7 @@ def test_to_hap_standalone(mock_driver): { "format": "string", "iid": 9, - "perms": ["pr", "ev"], + "perms": ["pr"], "type": "37", "value": "01.01.00", } diff --git a/tests/test_hds_recording.py b/tests/test_hds_recording.py new file mode 100644 index 00000000..2822674e --- /dev/null +++ b/tests/test_hds_recording.py @@ -0,0 +1,449 @@ +"""Tests for the HDS dataSend recording transfer.""" + +# pylint: disable=protected-access + +import asyncio +import os + +import pytest + +from pyhap import hds, hds_recording, hds_server +from pyhap.hds_protocol import EVENT, REQUEST, RESPONSE, HDSStatus, Message + + +def _paired(listener): + secret, salt = os.urandom(32), os.urandom(32) + accessory_salt = listener.register_transport(secret, salt) + acc_read, acc_write = hds.derive_keys(secret, salt, accessory_salt) + return hds.HDSCrypto(acc_write, acc_read) + + +class _CaptureTransport(asyncio.Transport): # pylint: disable=abstract-method + def __init__(self): + super().__init__() + self.buffer = bytearray() + self.closed = False + + def write(self, data): + self.buffer += data + + def close(self): + self.closed = True + + +def _drain_events(controller, transport): + """Decrypt and return every buffered accessory->controller message.""" + buffer = bytearray(transport.buffer) + transport.buffer = bytearray() + messages = [] + while True: + payload = controller.decrypt_frame(buffer) + if payload is None: + break + messages.append(Message.decode(payload)) + return messages + + +async def _run_recording(delegate, chunk_size=hds_recording.DEFAULT_CHUNK_SIZE): + listener = hds_server.HDSListener() + controller = _paired(listener) + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + hds_recording.RecordingStreamManager(connection, delegate, chunk_size=chunk_size) + + # Bind with the control handshake. + connection.data_received( + controller.encrypt_frame( + Message("control", REQUEST, "hello", {}, id=1).encode() + ) + ) + _drain_events(controller, transport) + + # Open the recording data stream. + connection.data_received( + controller.encrypt_frame( + Message( + "dataSend", + REQUEST, + "open", + {"streamId": 99, "type": "ipcamera.recording", "target": "controller"}, + id=2, + ).encode() + ) + ) + # Let the streaming task run. + await asyncio.sleep(0.05) + return controller, transport + + +@pytest.mark.asyncio +async def test_recording_stream_open_and_fragments(): + async def delegate(stream_id): + assert stream_id == 99 + yield hds_recording.RecordingPacket(b"moov-init-segment") + yield hds_recording.RecordingPacket(b"fragment-1") + yield hds_recording.RecordingPacket(b"fragment-2", is_last=True) + + controller, transport = await _run_recording(delegate) + messages = _drain_events(controller, transport) + + # First message is the open response, then three data events. + assert messages[0].kind == RESPONSE + assert messages[0].topic == "open" + assert messages[0].status == HDSStatus.SUCCESS + + data_events = [m for m in messages if m.kind == EVENT and m.topic == "data"] + assert len(data_events) == 3 + metas = [e.message["packets"][0]["metadata"] for e in data_events] + assert metas[0]["dataType"] == hds_recording.MEDIA_INITIALIZATION + assert metas[1]["dataType"] == hds_recording.MEDIA_FRAGMENT + assert [m["dataSequenceNumber"] for m in metas] == [1, 2, 3] + assert all(m["isLastDataChunk"] for m in metas) + assert data_events[-1].message["endOfStream"] is True + assert data_events[0].message["packets"][0]["data"] == b"moov-init-segment" + + +@pytest.mark.asyncio +async def test_recording_fragment_chunking(): + fragment = os.urandom(1000) + + async def delegate(stream_id): + yield hds_recording.RecordingPacket(b"init") + yield hds_recording.RecordingPacket(fragment, is_last=True) + + controller, transport = await _run_recording(delegate, chunk_size=256) + messages = _drain_events(controller, transport) + data_events = [m for m in messages if m.kind == EVENT and m.topic == "data"] + + # init (1 chunk) + fragment split into ceil(1000/256)=4 chunks. + fragment_events = [ + e + for e in data_events + if e.message["packets"][0]["metadata"]["dataSequenceNumber"] == 2 + ] + assert len(fragment_events) == 4 + metas = [e.message["packets"][0]["metadata"] for e in fragment_events] + assert [m["dataChunkSequenceNumber"] for m in metas] == [1, 2, 3, 4] + # dataTotalSize only on the first chunk. + assert metas[0]["dataTotalSize"] == 1000 + assert "dataTotalSize" not in metas[1] + assert [m["isLastDataChunk"] for m in metas] == [False, False, False, True] + # Reassembling the chunks reproduces the fragment. + reassembled = b"".join(e.message["packets"][0]["data"] for e in fragment_events) + assert reassembled == fragment + # endOfStream only on the very last chunk. + assert fragment_events[-1].message["endOfStream"] is True + assert "endOfStream" not in fragment_events[0].message + + +@pytest.mark.asyncio +async def test_recording_open_rejects_wrong_type(): + async def delegate(stream_id): + yield hds_recording.RecordingPacket(b"x", is_last=True) + + listener = hds_server.HDSListener() + controller = _paired(listener) + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + hds_recording.RecordingStreamManager(connection, delegate) + + connection.data_received( + controller.encrypt_frame( + Message("control", REQUEST, "hello", {}, id=1).encode() + ) + ) + _drain_events(controller, transport) + + connection.data_received( + controller.encrypt_frame( + Message( + "dataSend", + REQUEST, + "open", + {"streamId": 1, "type": "wrong", "target": "controller"}, + id=2, + ).encode() + ) + ) + await asyncio.sleep(0.02) + response = _drain_events(controller, transport)[0] + assert response.status == HDSStatus.PROTOCOL_SPECIFIC_ERROR + assert ( + response.message["status"] == hds_recording.RecordingReason.UNEXPECTED_FAILURE + ) + + +@pytest.mark.asyncio +async def test_recording_close_stops_stream(): + started = asyncio.Event() + release = asyncio.Event() + + async def delegate(stream_id): + yield hds_recording.RecordingPacket(b"init") + started.set() + await release.wait() # block until the controller closes + yield hds_recording.RecordingPacket(b"never", is_last=True) + + listener = hds_server.HDSListener() + controller = _paired(listener) + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + hds_recording.RecordingStreamManager(connection, delegate) + + connection.data_received( + controller.encrypt_frame( + Message("control", REQUEST, "hello", {}, id=1).encode() + ) + ) + _drain_events(controller, transport) + connection.data_received( + controller.encrypt_frame( + Message( + "dataSend", + REQUEST, + "open", + {"streamId": 7, "type": "ipcamera.recording", "target": "controller"}, + id=2, + ).encode() + ) + ) + await started.wait() + _drain_events(controller, transport) + + connection.data_received( + controller.encrypt_frame( + Message("dataSend", REQUEST, "close", {"streamId": 7}, id=3).encode() + ) + ) + await asyncio.sleep(0.02) + messages = _drain_events(controller, transport) + assert messages[0].topic == "close" + assert messages[0].status == HDSStatus.SUCCESS + # The blocked "never" fragment was never sent. + assert not [m for m in messages if m.kind == EVENT] + + +@pytest.mark.asyncio +async def test_unbound_connection_is_capped(): + """A connection that never completes the control handshake must not buffer + unboundedly: once it exceeds the pre-bind cap it is closed. + """ + listener = hds_server.HDSListener() + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + connection.data_received(b"\x00" * (hds_server._MAX_UNBOUND_BYTES + 1)) + assert transport.closed + + +def test_pending_transports_are_bounded(): + """Abandoned SetupDataStreamTransport registrations must not accumulate.""" + listener = hds_server.HDSListener() + for _ in range(hds_server._MAX_PENDING_TRANSPORTS + 5): + listener.register_transport(os.urandom(32), os.urandom(32)) + assert len(listener._pending) == hds_server._MAX_PENDING_TRANSPORTS + + +@pytest.mark.asyncio +async def test_connection_loss_cancels_stream_and_runs_delegate_cleanup(): + """Dropping the TCP connection mid-recording must cancel the stream task and + run the delegate's cleanup (e.g. terminate its ffmpeg subprocess). + """ + started = asyncio.Event() + cleaned = asyncio.Event() + + async def delegate(stream_id): + try: + yield hds_recording.RecordingPacket(b"init") + started.set() + await asyncio.Event().wait() # block until cancelled + yield hds_recording.RecordingPacket(b"never", is_last=True) + finally: + cleaned.set() + + listener = hds_server.HDSListener() + controller = _paired(listener) + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + manager = hds_recording.RecordingStreamManager(connection, delegate) + connection.data_received( + controller.encrypt_frame( + Message("control", REQUEST, "hello", {}, id=1).encode() + ) + ) + connection.data_received( + controller.encrypt_frame( + Message( + "dataSend", + REQUEST, + "open", + {"streamId": 5, "type": "ipcamera.recording", "target": "controller"}, + id=2, + ).encode() + ) + ) + await started.wait() + + connection.connection_lost(None) + await asyncio.sleep(0.02) + assert cleaned.is_set() + assert manager._task.done() + + +@pytest.mark.asyncio +async def test_write_backpressure_gates_fragments(): + """While the transport is paused, no data events are produced; resuming lets + them flow. + """ + + async def delegate(stream_id): + yield hds_recording.RecordingPacket(b"init") + yield hds_recording.RecordingPacket(b"fragment", is_last=True) + + listener = hds_server.HDSListener() + controller = _paired(listener) + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + hds_recording.RecordingStreamManager(connection, delegate) + connection.data_received( + controller.encrypt_frame( + Message("control", REQUEST, "hello", {}, id=1).encode() + ) + ) + connection.pause_writing() + connection.data_received( + controller.encrypt_frame( + Message( + "dataSend", + REQUEST, + "open", + {"streamId": 8, "type": "ipcamera.recording", "target": "controller"}, + id=2, + ).encode() + ) + ) + await asyncio.sleep(0.02) + paused = [ + m + for m in _drain_events(controller, transport) + if m.kind == EVENT and m.topic == "data" + ] + assert paused == [] # nothing sent while paused + + connection.resume_writing() + await asyncio.sleep(0.02) + resumed = [ + m + for m in _drain_events(controller, transport) + if m.kind == EVENT and m.topic == "data" + ] + assert len(resumed) == 2 # init + fragment now delivered + + +@pytest.mark.asyncio +async def test_double_stop_does_not_abort_delegate_cleanup(): + """A dataSend/close followed by a TCP drop (two _stop calls) must not cancel + the stream twice and abort the delegate's in-flight async cleanup. + """ + started = asyncio.Event() + release_cleanup = asyncio.Event() + cleanup_done = asyncio.Event() + + async def delegate(stream_id): + try: + yield hds_recording.RecordingPacket(b"init") + started.set() + await asyncio.Event().wait() # block until cancelled + finally: + await release_cleanup.wait() # async teardown that must complete + cleanup_done.set() + + listener = hds_server.HDSListener() + controller = _paired(listener) + connection = hds_server.HDSConnection(listener) + transport = _CaptureTransport() + connection.connection_made(transport) + manager = hds_recording.RecordingStreamManager(connection, delegate) + connection.data_received( + controller.encrypt_frame( + Message("control", REQUEST, "hello", {}, id=1).encode() + ) + ) + connection.data_received( + controller.encrypt_frame( + Message( + "dataSend", + REQUEST, + "open", + {"streamId": 9, "type": "ipcamera.recording", "target": "controller"}, + id=2, + ).encode() + ) + ) + await started.wait() + + manager._stop() # cancel #1 (dataSend/close) -> enters aclose cleanup + await asyncio.sleep(0.01) + connection.connection_lost(None) # cancel #2 (TCP drop) must be suppressed + await asyncio.sleep(0.01) + assert not cleanup_done.is_set() # still blocked in the delegate's cleanup + release_cleanup.set() + await asyncio.sleep(0.01) + assert cleanup_done.is_set() # cleanup ran to completion despite double stop + + +@pytest.mark.asyncio +async def test_fmp4_recording_packets_splits_init_and_fragments(): + """Yield ftyp+moov as init and each moof+mdat as a fragment, last flagged. + + Exercises the source-agnostic framer end to end over a synthetic stream. + """ + + def box(box_type, body=b"\x00" * 8): + return (8 + len(body)).to_bytes(4, "big") + box_type + body + + stream = ( + box(b"ftyp") + + box(b"moov") # initialization + + box(b"moof") + + box(b"mdat") # fragment 1 + + box(b"moof") + + box(b"mdat") # fragment 2 + ) + reader = asyncio.StreamReader() + reader.feed_data(stream) + reader.feed_eof() + + packets = [p async for p in hds_recording.fmp4_recording_packets(reader)] + assert len(packets) == 3 + # init packet carries both ftyp and moov, not flagged last + assert packets[0].data[4:8] == b"ftyp" + assert b"moov" in packets[0].data + assert not packets[0].is_last + assert packets[1].data[4:8] == b"moof" and not packets[1].is_last + # the fragment right before EOF is the last one + assert packets[2].data[4:8] == b"moof" and packets[2].is_last + + +@pytest.mark.asyncio +async def test_fmp4_recording_packets_handles_64bit_box_size(): + """A box using the 64-bit extended-size form is read whole.""" + body = b"\x11" * 8 + ftyp = (16).to_bytes(4, "big") + b"ftyp" + b"\x00" * 8 + moov = (16).to_bytes(4, "big") + b"moov" + b"\x00" * 8 + # moof with 64-bit size: size32 == 1, then 64-bit size = 16 + 8 + len(body) + big_moof = ( + (1).to_bytes(4, "big") + b"moof" + (16 + len(body)).to_bytes(8, "big") + body + ) + reader = asyncio.StreamReader() + reader.feed_data(ftyp + moov + big_moof) + reader.feed_eof() + packets = [p async for p in hds_recording.fmp4_recording_packets(reader)] + assert len(packets) == 2 + assert packets[1].data[4:8] == b"moof" and packets[1].data.endswith(body) + assert packets[1].is_last diff --git a/tests/test_hksv_recording.py b/tests/test_hksv_recording.py new file mode 100644 index 00000000..ed0a8f5b --- /dev/null +++ b/tests/test_hksv_recording.py @@ -0,0 +1,280 @@ +"""Tests for classic HomeKit Secure Video recording. + +Covers the recording configuration TLV codecs, the recording management service +setup (eligibility characteristics), and the HDS SetupDataStreamTransport +write-response handshake. The HDS fragment transfer itself is covered in +test_hds_recording.py. +""" + +# pylint: disable=protected-access + +import os +from unittest.mock import MagicMock, patch + +import pytest + +from pyhap import camera as camera_module, hds, hksv_recording as hr, tlv +from pyhap.accessory_driver import AccessoryDriver +from pyhap.camera import Camera +from pyhap.util import base64_to_bytes, to_base64_str + + +@pytest.fixture(name="recording_camera") +def recording_camera_fixture(): + with ( + patch("pyhap.accessory_driver.AccessoryDriver.persist"), + patch("pyhap.accessory_driver.AccessoryDriver.load"), + ): + driver = AccessoryDriver(loop=MagicMock(), listen_address="127.0.0.1") + options = { + "video": { + "codec": { + "profiles": [ + camera_module.VIDEO_CODEC_PARAM_PROFILE_ID_TYPES["BASELINE"] + ], + "levels": [camera_module.VIDEO_CODEC_PARAM_LEVEL_TYPES["TYPE3_1"]], + }, + "resolutions": [[1920, 1080, 30], [1280, 720, 30]], + }, + "audio": {"codecs": [{"type": "OPUS", "samplerate": 24}]}, + "srtp": True, + "address": "127.0.0.1", + "recording": True, + } + return Camera(options, driver, "Cam") + + +# --- Recording configuration TLV codecs ------------------------------------ + + +def test_recording_supported_config_roundtrip(): + config = hr.SupportedRecordingConfiguration( + prebuffer_length_ms=4000, + event_triggers=hr.EventTrigger.MOTION | hr.EventTrigger.DOORBELL, + media_containers=[hr.MediaContainerConfiguration(fragment_length_ms=4000)], + ) + decoded = hr.SupportedRecordingConfiguration.decode(config.encode()) + assert decoded == config + assert decoded.event_triggers == 3 + + +def test_recording_supported_video_roundtrip(): + configs = [ + hr.VideoCodecConfiguration( + codec_type=hr.VideoCodecType.H265, + profile=2, + level=2, + bitrate_kbps=2000, + iframe_interval_ms=4000, + attributes=[ + hr.VideoAttributes(3840, 2160, 30), + hr.VideoAttributes(1920, 1080, 30), + ], + ), + hr.VideoCodecConfiguration( + codec_type=hr.VideoCodecType.H264, + profile=0, + level=0, + bitrate_kbps=800, + iframe_interval_ms=4000, + attributes=[hr.VideoAttributes(1280, 720, 24)], + ), + ] + decoded = hr.decode_supported_video(hr.encode_supported_video(configs)) + assert len(decoded) == 2 + assert decoded[0].codec_type == hr.VideoCodecType.H265 + assert decoded[0].profile == 2 and decoded[0].level == 2 + assert [(a.width, a.height, a.frame_rate) for a in decoded[0].attributes] == [ + (3840, 2160, 30), + (1920, 1080, 30), + ] + assert decoded[1].codec_type == hr.VideoCodecType.H264 + assert [(a.width, a.height, a.frame_rate) for a in decoded[1].attributes] == [ + (1280, 720, 24) + ] + # The supported advertisement omits per-config bitrate / iframe interval; + # those only appear in the controller-selected configuration. + assert decoded[0].bitrate_kbps == 0 and decoded[0].iframe_interval_ms == 0 + + +def test_recording_supported_audio_roundtrip(): + configs = [ + hr.AudioCodecConfiguration( + codec_type=hr.AudioCodecType.AAC_LC, + channels=1, + bitrate_mode=hr.BitRateMode.VARIABLE, + sample_rate=hr.AudioSampleRate.KHZ_32, + max_audio_bitrate_kbps=64, + ), + hr.AudioCodecConfiguration( + codec_type=hr.AudioCodecType.AAC_ELD, + sample_rate=hr.AudioSampleRate.KHZ_24, + ), + ] + decoded = hr.decode_supported_audio(hr.encode_supported_audio(configs)) + assert len(decoded) == 2 + assert decoded[0].codec_type == hr.AudioCodecType.AAC_LC + assert decoded[0].sample_rate == hr.AudioSampleRate.KHZ_32 + assert decoded[1].codec_type == hr.AudioCodecType.AAC_ELD + # max-bitrate is omitted from the supported advertisement. + assert decoded[0].max_audio_bitrate_kbps == 0 + + +def test_recording_supported_video_multi_value_roundtrip(): + # A supported config advertising several profiles/levels must round-trip + # faithfully: encode -> decode preserves the full lists (not just the first). + config = hr.VideoCodecConfiguration( + codec_type=hr.VideoCodecType.H264, + profile=[0, 1, 2], + level=[0, 1, 2], + bitrate_kbps=2000, + iframe_interval_ms=4000, + attributes=[ + hr.VideoAttributes(1920, 1080, 30), + hr.VideoAttributes(1280, 720, 30), + ], + ) + decoded = hr.decode_supported_video(hr.encode_supported_video([config]))[0] + assert decoded.profile == [0, 1, 2] + assert decoded.level == [0, 1, 2] + assert [(a.width, a.height, a.frame_rate) for a in decoded.attributes] == [ + (1920, 1080, 30), + (1280, 720, 30), + ] + + +def test_recording_selected_config_roundtrip(): + selected = hr.SelectedRecordingConfiguration( + recording=hr.SupportedRecordingConfiguration( + prebuffer_length_ms=4000, event_triggers=hr.EventTrigger.MOTION + ), + video=hr.VideoCodecConfiguration( + codec_type=hr.VideoCodecType.H265, + profile=2, + level=2, + bitrate_kbps=2000, + iframe_interval_ms=4000, + attributes=[hr.VideoAttributes(1920, 1080, 30)], + ), + audio=hr.AudioCodecConfiguration(), + ) + decoded = hr.SelectedRecordingConfiguration.decode(selected.encode()) + assert decoded == selected + + +# --- Recording management service setup / eligibility ---------------------- + + +def test_recording_management_service_present(recording_camera): + recording = recording_camera.get_service("CameraRecordingManagement") + assert recording is not None + # Recording starts inactive. + assert recording.get_characteristic("Active").get_value() == 0 + for char in ( + "SupportedCameraRecordingConfiguration", + "SupportedVideoRecordingConfiguration", + "SupportedAudioRecordingConfiguration", + ): + assert recording.get_characteristic(char).get_value() + + +def test_recording_eligibility_characteristics(recording_camera): + # HAP-NodeJS-parity characteristics iOS validates before managing recording. + stream = recording_camera.get_service("CameraRTPStreamManagement") + assert stream.get_characteristic("Active").get_value() == 1 + + operating_mode = recording_camera.get_service("CameraOperatingMode") + assert operating_mode.get_characteristic("HomeKitCameraActive").get_value() == 1 + assert operating_mode.get_characteristic("PeriodicSnapshotsActive").get_value() == 1 + + motion = recording_camera.get_service("MotionSensor") + assert motion.get_characteristic("StatusActive").get_value() is True + + +# --- HDS SetupDataStreamTransport write-response handshake ------------------ + + +def test_data_stream_transport_service_present(recording_camera): + service = recording_camera.get_service("DataStreamTransportManagement") + assert service.get_characteristic("Version").get_value() == "1.0" + assert ( + service.get_characteristic( + "SupportedDataStreamTransportConfiguration" + ).get_value() + is not None + ) + + +def test_setup_data_stream_transport_without_session_errors(recording_camera): + camera = recording_camera + # No session key for this client, and no listener started. + request = tlv.encode(hds.SETUP_TYPES["CONTROLLER_KEY_SALT"], b"\x11" * 32) + # SetupDataStreamTransport is a write-response characteristic: the encoded + # response is RETURNED from the setter (so HAP puts it in the write + # response), not stored on the characteristic. + value = camera.set_data_stream_transport( + to_base64_str(request), sender_client_addr=("10.0.0.9", 5000) + ) + objs = tlv.decode(base64_to_bytes(value)) + assert objs[hds.SETUP_RESPONSE_TYPES["STATUS"]] == hds.SETUP_STATUS_GENERIC_ERROR + + +@pytest.mark.asyncio +async def test_setup_data_stream_transport_write_response_plumbing(recording_camera): + # The HAP write-response path calls client_update_value and forwards the + # setter's RETURN value. Exercise that path (not the setter directly) and + # confirm the address opt-in is detected for this two-argument setter. + camera = recording_camera + await camera._ensure_hds_listener() + try: + client = ("10.0.0.9", 5000) + camera.driver.session_shared_keys[client] = os.urandom(32) + char = camera.get_service("DataStreamTransportManagement").get_characteristic( + "SetupDataStreamTransport" + ) + assert char._setter_wants_addr() is True + request = tlv.encode(hds.SETUP_TYPES["CONTROLLER_KEY_SALT"], os.urandom(32)) + value = char.client_update_value(to_base64_str(request), client) + objs = tlv.decode(base64_to_bytes(value)) + assert objs[hds.SETUP_RESPONSE_TYPES["STATUS"]] == hds.SETUP_STATUS_SUCCESS + finally: + await camera.stop() + + +def test_single_argument_setter_still_called_with_one_argument(recording_camera): + # Regression guard for the characteristic change: a plain single-argument + # setter must still be invoked with the value alone even when a client + # address is available. + mute = recording_camera.get_service("Microphone").get_characteristic("Mute") + seen = [] + mute.setter_callback = seen.append + assert mute._setter_wants_addr() is False + mute.client_update_value(True, ("10.0.0.9", 5000)) + assert seen == [True] + + +@pytest.mark.asyncio +async def test_setup_data_stream_transport_with_session(recording_camera): + camera = recording_camera + await camera._ensure_hds_listener() + try: + client = ("10.0.0.9", 5000) + camera.driver.session_shared_keys[client] = os.urandom(32) + request = tlv.encode(hds.SETUP_TYPES["CONTROLLER_KEY_SALT"], os.urandom(32)) + value = camera.set_data_stream_transport( + to_base64_str(request), sender_client_addr=client + ) + objs = tlv.decode(base64_to_bytes(value)) + assert objs[hds.SETUP_RESPONSE_TYPES["STATUS"]] == hds.SETUP_STATUS_SUCCESS + assert objs[hds.SETUP_RESPONSE_TYPES["ACCESSORY_KEY_SALT"]] + # The write response must carry the accessory's TCP listening port so + # the controller can open the HDS connection. + session_params = tlv.decode( + objs[hds.SETUP_RESPONSE_TYPES["TRANSPORT_TYPE_SESSION_PARAMETERS"]] + ) + port = int.from_bytes( + session_params[hds.TRANSPORT_SESSION_PARAM_TCP_LISTENING_PORT], "little" + ) + assert port == camera._hds_listener.port + finally: + await camera.stop()