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()