Add classic HomeKit Secure Video recording

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.

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 ###
 

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,14 +407,32 @@ 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)
         if self._always_null:
             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.