diff --git a/pyhap/accessory_driver.py b/pyhap/accessory_driver.py index c093391f..9ab0f2ba 100644 --- a/pyhap/accessory_driver.py +++ b/pyhap/accessory_driver.py @@ -1124,7 +1124,7 @@ def derive_session_key( return None return connection.derive_key(salt, info) - def local_address( + def _local_address( self, client_addr: tuple[str, int] | None, ipv6: bool ) -> str | None: """The accessory's address on a controller's connection. diff --git a/pyhap/camera.py b/pyhap/camera.py index ae9fa8b0..c9660658 100644 --- a/pyhap/camera.py +++ b/pyhap/camera.py @@ -633,7 +633,7 @@ def __init__(self, options: dict[str, Any], *args: Any, **kwargs: Any) -> None: ``audio``: iOS 27 shows a camera without the classic RTP stream management service as not responding. Its services come before the RTP stream management services, which then get a new - ``unique_id``; for a paired camera this changes instance IDs. + ``unique_id``. For a paired camera this changes instance IDs. - webrtc - :class:`pyhap.camera_webrtc.WebRTCConfig`, or a dict of its fields. Developer Preview of HomeKit Secure Video 17.99, experimental. Adds a Camera WebRTC Stream Management service per @@ -875,7 +875,7 @@ def _update_stream_status(self) -> None: for controller in self._stream_controllers(): controller.update_status() - def stream_address_for( + def _stream_address_for( self, client_addr: tuple[str, int] | None, is_ipv6: bool ) -> str | None: """The address to stream from to a controller, in the given IP version. @@ -883,37 +883,13 @@ def stream_address_for( The accessory's address on the controller's connection, else the ``address`` option if it has that version. """ - address = self.driver.local_address(client_addr, is_ipv6) + address = self.driver._local_address(client_addr, is_ipv6) if address is not None: return address if (self.stream_address_isv6 == b"\x01") == is_ipv6: return str(self.stream_address) return None - async def end_session(self, session_id: UUID) -> bool: - """End a multi-tier or WebRTC session (17.99) on the camera's side. - - E.g. when its media source stopped. Ends the sessions with that ID of - every sensor and service and waits until their delegates released - them. Returns False if there is no such session. - """ - ended = await asyncio.gather( - *( - controller.end_session(session_id) - for controller in self._stream_controllers() - ) - ) - return any(ended) - - async def end_all_sessions(self) -> None: - """End every multi-tier and WebRTC session (17.99) and wait for them.""" - await asyncio.gather( - *( - controller.end_all_sessions() - for controller in self._stream_controllers() - ) - ) - def client_disconnected(self, client_addr: tuple[str, int]) -> None: """End the multi-tier and WebRTC sessions of the connection.""" for controller in self._stream_controllers(): diff --git a/pyhap/camera_multi_tier.py b/pyhap/camera_multi_tier.py index 2cc21dde..3f29836a 100644 --- a/pyhap/camera_multi_tier.py +++ b/pyhap/camera_multi_tier.py @@ -13,27 +13,24 @@ from __future__ import annotations -import asyncio -from collections.abc import Mapping, Sequence -import contextlib +from collections.abc import Mapping from dataclasses import dataclass, field +import ipaddress import logging -import re import secrets -import socket from typing import TYPE_CHECKING, Any, Protocol, cast from pyhap import hksv3 from pyhap.camera import parse_setup_address from pyhap.camera_stream import ( DEFAULT_HOOK_TIMEOUT, + HookFailed, SessionState, StreamController, StreamSession, - _check_hook_timeout, - _config_from_option, - _failure_status, - _HookFailed, + check_hook_timeout, + config_from_option, + failure_status, ) from pyhap.util import base64_to_bytes, to_base64_str @@ -42,16 +39,11 @@ __all__ = [ "DEFAULT_IDLE_TIMEOUT", - "FFMPEG_AUDIO_OUTPUT", - "FFMPEG_VIDEO_OUTPUT", "MULTI_TIER_MIN_SESSIONS", - "FFmpegMultiTierDelegate", "MultiTierConfig", "MultiTierController", "MultiTierDelegate", "MultiTierSession", - "RtcpMonitor", - "open_rtcp_monitor", ] logger = logging.getLogger(__name__) @@ -61,6 +53,7 @@ # Seconds a started session lives without keep_alive(). DEFAULT_IDLE_TIMEOUT = 30.0 _SRTP_SUITE = hksv3.SRTPCryptoSuite.AES_CM_128_HMAC_SHA1_80 +_STATUS = hksv3.RTPStreamingStatus @dataclass(eq=False, kw_only=True) @@ -70,7 +63,7 @@ class MultiTierSession(StreamSession): ``address`` and the ports are where the controller receives, with the SRTP keys it sent. The accessory sends from ``accessory_address`` and the ``accessory_*_port`` values, by default its address on the controller's - connection and the controller's port numbers; ``handle_prepare`` may + connection and the controller's port numbers. ``handle_prepare`` may change them. ``video_ssrc`` and ``audio_ssrc`` are the accessory's SSRCs from the Setup Endpoints answer, the controller filters by them. The ``controller_*_ssrc`` values are what the controller sent with the last @@ -99,67 +92,11 @@ def keep_alive(self) -> None: """Restart the idle timeout of the started session. Call it when the controller is still there, e.g. for each of its RTCP - packets as :class:`RtcpMonitor` does. Call it in the event loop. + packets. Call it in the event loop. """ controller = self._controller if isinstance(controller, MultiTierController): - controller.keep_alive(self) - - -class RtcpMonitor(asyncio.DatagramProtocol): - """Keeps a session alive while its controller sends RTCP. - - Listens on the accessory's video port of the session, where the - controller sends its RTCP. Each datagram from the controller's address - calls :meth:`MultiTierSession.keep_alive`, without decrypting it. Others - are ignored. Open it with :func:`open_rtcp_monitor`. - """ - - def __init__(self, session: MultiTierSession) -> None: - self.session = session - self.port = 0 - self._transport: asyncio.DatagramTransport | None = None - - def connection_made(self, transport: asyncio.BaseTransport) -> None: - self._transport = cast(asyncio.DatagramTransport, transport) - self.port = int(transport.get_extra_info("sockname")[1]) - - def datagram_received(self, data: bytes, addr: tuple[Any, ...]) -> None: - # The controller's address has no scope ID, a received one may. - if str(addr[0]).split("%", 1)[0] == self.session.address: - self.session.keep_alive() - - @property - def closed(self) -> bool: - return self._transport is None or self._transport.is_closing() - - def close(self) -> None: - """Close the socket, e.g. in ``handle_end``.""" - if self._transport is not None: - self._transport.close() - - -async def open_rtcp_monitor(session: MultiTierSession) -> RtcpMonitor: - """Open an :class:`RtcpMonitor` for ``session``, e.g. in ``handle_prepare``. - - Binds a free UDP port of the controller's IP version and announces it as - the accessory's video port, ``session.accessory_video_port``. The stream - is then sent from another port, with its RTCP to the controller's video - port; for ffmpeg, an ``srtp://`` output without ``localrtpport`` and with - ``rtcpport`` set to ``session.video_port``. Close the monitor when the - session ends. - - :raises OSError: If no port can be bound. - """ - loop = asyncio.get_running_loop() - # Any interface, as the stream's own ports. Only the controller's - # datagrams count. - any_address = "::" if session.is_ipv6 else "0.0.0.0" # noqa: S104 - _, monitor = await loop.create_datagram_endpoint( - lambda: RtcpMonitor(session), local_addr=(any_address, 0) - ) - session.accessory_video_port = monitor.port - return monitor + controller._keep_alive(self) class MultiTierDelegate(Protocol): @@ -171,8 +108,7 @@ class MultiTierDelegate(Protocol): With the default ``idle_timeout`` a started session ends after 30 s unless the delegate calls :meth:`MultiTierSession.keep_alive`, e.g. for each - RTCP packet of the controller; :class:`RtcpMonitor` does that for a - delegate that cannot see the controller's RTCP itself. + RTCP packet of the controller. """ async def handle_prepare(self, session: MultiTierSession) -> None: @@ -225,12 +161,17 @@ class MultiTierConfig: ``delegate`` sends the media. ``audio`` is the one audio tier the guide allows. ``max_sessions`` per sensor is at least five, as the guide - requires, and one connection has at most ``max_sessions_per_connection`` - of them. A delegate hook gets ``hook_timeout`` seconds. A session must - start within ``setup_timeout`` seconds of its Setup Endpoints answer. + requires. One connection has at most ``max_sessions_per_connection`` of + them, by default three. That is below the guide's "at least five + simultaneous sessions" per service on purpose: sessions one connection + leaks cannot take the room of other controllers, and remote viewers, + who share the home hub's connection, stream over WebRTC. Raise it, at + most to ``max_sessions``, to let one connection have more. + A delegate hook gets ``hook_timeout`` seconds. A session must start + within ``setup_timeout`` seconds of its Setup Endpoints answer. A started session ends unless :meth:`MultiTierSession.keep_alive` is called at least every ``idle_timeout`` seconds, e.g. for each RTCP packet - of the controller; None turns that off. + of the controller. None turns that off. """ delegate: MultiTierDelegate | None = None @@ -247,10 +188,10 @@ def from_option(cls, value: MultiTierConfig | Mapping[str, Any]) -> MultiTierCon :raises ValueError: For an unknown key. """ - return _config_from_option(cls, value, "multi_tier") + return config_from_option(cls, value, "multi_tier") def __post_init__(self) -> None: - _check_hook_timeout(self.hook_timeout) + check_hook_timeout(self.hook_timeout) if self.max_sessions < MULTI_TIER_MIN_SESSIONS: raise ValueError( f"multi_tier max_sessions must be at least {MULTI_TIER_MIN_SESSIONS}" @@ -268,14 +209,14 @@ def __post_init__(self) -> None: raise ValueError("multi_tier timeouts must be positive") -def supported_audio_tiers(audio: hksv3.AudioStreamTier) -> bytes: +def _supported_audio_tiers(audio: hksv3.AudioStreamTier) -> bytes: """The Supported Audio Stream Tiers value, exactly one Opus tier.""" return hksv3.SupportedAudioStreamTiers( hksv3.AudioCodec.OPUS, hksv3.AUDIO_PAYLOAD_TYPE, [audio] ).encode() -def supported_video_tiers(tiers: list[hksv3.VideoStreamTier]) -> bytes: +def _supported_video_tiers(tiers: list[hksv3.VideoStreamTier]) -> bytes: """The Supported Video Stream Tiers value, H.265 only.""" return hksv3.SupportedVideoStreamTiers( hksv3.VideoCodec.H265, hksv3.VIDEO_PAYLOAD_TYPE, tiers @@ -308,8 +249,8 @@ def __init__( ) for name, value in ( ("SensorUUID", sensor.sensor_uuid.bytes), - ("SupportedVideoStreamTiers", supported_video_tiers(self.tiers)), - ("SupportedAudioStreamTiers", supported_audio_tiers(config.audio)), + ("SupportedVideoStreamTiers", _supported_video_tiers(self.tiers)), + ("SupportedAudioStreamTiers", _supported_audio_tiers(config.audio)), ): service.configure_char(name, value=to_base64_str(value)) # The guide allows AES_CM_128_HMAC_SHA1_80 only. @@ -323,9 +264,10 @@ def __init__( ) # The answer has the session's SRTP keys, only its controller reads it. setup_endpoints.read_back_response = True + # Neither the request nor the status is shared with other connections. service.configure_char( "RTPStreamingControl", setter_callback=self._set_streaming_control - ) + ).read_back_response = True @property def active(self) -> bool: @@ -339,7 +281,7 @@ def update_status(self) -> None: if not active: self._end_sessions("streaming was turned off") - def keep_alive(self, session: MultiTierSession) -> None: + def _keep_alive(self, session: MultiTierSession) -> None: """Restart the ``idle_timeout`` of a started session. A former session with the same ID keeps nothing alive. See @@ -366,7 +308,7 @@ async def _release_hook(self, session: MultiTierSession) -> None: async def _set_setup_endpoints( self, value: str, sender_client_addr: tuple[str, int] | None = None - ) -> Any: + ) -> str: """Set up a session, only the writing connection reads the answer.""" deadline = self._deadline() try: @@ -414,11 +356,12 @@ async def _setup( return error if existing is None and (no_room := self._no_room(client_addr)): logger.warning("[%s] %s", session_id, no_room) - return hksv3.SetupEndpointsStatus.BUSY + # Busy asks to try again, a closed service never takes one. + return error if self._closed else hksv3.SetupEndpointsStatus.BUSY if self.delegate is None: logger.warning("[%s] No multi-tier delegate, refusing", session_id) return error - accessory_address = self.camera.stream_address_for( + accessory_address = self.camera._stream_address_for( client_addr, endpoint.is_ipv6 ) if accessory_address is None: @@ -443,7 +386,7 @@ async def _setup( return error if no_room := self._no_room(client_addr): logger.warning("[%s] %s", session_id, no_room) - return hksv3.SetupEndpointsStatus.BUSY + return error if self._closed else hksv3.SetupEndpointsStatus.BUSY session = MultiTierSession( id=session_id, sensor=self.sensor, @@ -475,8 +418,23 @@ async def _setup( end_on_failure=True, first=True, ) - except _HookFailed as err: - return _failure_status(hksv3.SetupEndpointsStatus, err.status, error) + except HookFailed as err: + return failure_status(hksv3.SetupEndpointsStatus, err.status, error) + ports = session.accessory_video_port, session.accessory_audio_port + if not all(map(_valid_port, ports)): + logger.error( + "[%s] The prepare hook set invalid ports %s", session_id, ports + ) + self._end(session, "the prepare hook set invalid ports") + return error + if not _valid_address(session.accessory_address, session.is_ipv6): + logger.error( + "[%s] The accessory address %r is invalid", + session_id, + session.accessory_address, + ) + self._end(session, "the accessory address is invalid") + return error session.state = SessionState.PREPARED self._set_timer( session, @@ -495,7 +453,7 @@ async def _set_streaming_control( request = hksv3.RTPStreamingControl.decode(base64_to_bytes(value)) except ValueError: logger.warning("Ignoring a malformed RTP Streaming Control write") - status = hksv3.encode_streaming_status(None, hksv3.RTPStreamingStatus.ERROR) + status = hksv3.encode_streaming_status(None, _STATUS.ERROR) return to_base64_str(status) logger.debug("RTP Streaming Control: %s", request) result = await self._streaming_control(request, sender_client_addr, deadline) @@ -508,26 +466,25 @@ async def _streaming_control( client_addr: tuple[str, int] | None, deadline: float, ) -> hksv3.RTPStreamingStatus: - status = hksv3.RTPStreamingStatus session = self._owned(request.session_id, client_addr) if session is None: - return status.UNKNOWN_SESSION + return _STATUS.UNKNOWN_SESSION if request.command == hksv3.RTPStreamingCommand.END: if session.state is not SessionState.STARTED: self._end(session, "ended before it started") - return status.NO_SUCH_STREAM + return _STATUS.NO_SUCH_STREAM ended = await self._end_by_request(session, deadline) - return status.SUCCESS if ended else status.ERROR + return _STATUS.SUCCESS if ended else _STATUS.ERROR if request.command != hksv3.RTPStreamingCommand.START: - return status.ERROR + return _STATUS.ERROR if session.state is SessionState.NEW or session.busy: - return status.BUSY + return _STATUS.BUSY if not self.active: logger.warning("[%s] Streaming is turned off, refusing it", session.id) - return status.ERROR + return _STATUS.ERROR tiers = self._tiers(request) if tiers is None: - return status.ERROR + return _STATUS.ERROR previous_ssrcs = session.controller_video_ssrc, session.controller_audio_ssrc session.controller_video_ssrc = request.video_ssrc session.controller_audio_ssrc = request.audio_ssrc @@ -542,7 +499,7 @@ async def _streaming_control( await self._run_hook( session, "reconfigure" if started else "start", hook, deadline ) - except _HookFailed as err: + except HookFailed as err: session.controller_video_ssrc, session.controller_audio_ssrc = ( previous_ssrcs ) @@ -553,11 +510,11 @@ async def _streaming_control( "Not started in time", SessionState.PREPARED, ) - return _failure_status(status, err.status, status.ERROR) + return failure_status(_STATUS, err.status, _STATUS.ERROR) session.video_tier, session.audio_tier = tiers session.state = SessionState.STARTED - self.keep_alive(session) - return status.SUCCESS + self._keep_alive(session) + return _STATUS.SUCCESS def _tiers( self, request: hksv3.RTPStreamingControl @@ -577,6 +534,25 @@ def _tiers( return video, audio +def _valid_port(port: object) -> bool: + return type(port) is int and 0 < port <= 0xFFFF + + +def _valid_address(address: object, is_ipv6: bool) -> bool: + """Whether ``address`` is a plain IP address string of the session's family. + + Like the controller's address, it must not carry an IPv6 scope ID. An + IPv4-mapped IPv6 address does not count as IPv6. + """ + if not isinstance(address, str): + return False + try: + parsed = parse_setup_address(address.encode("utf-8"), is_ipv6) + except ValueError: + return False + return not (isinstance(parsed, ipaddress.IPv6Address) and parsed.ipv4_mapped) + + def _ssrc() -> int: # Up to 2^31 - 1, ffmpeg takes the SSRC as a signed int. return secrets.randbelow(2**31 - 1) + 1 @@ -612,243 +588,3 @@ def _answer(session: MultiTierSession) -> hksv3.SetupEndpointsResponse: session.video_ssrc, session.audio_ssrc, ) - - -# Global options of FFmpegMultiTierDelegate, before the input: no banner, no -# status line, only warnings on stderr. -FFMPEG_GLOBAL_OPTIONS = ("-hide_banner", "-nostats", "-loglevel", "warning") - -# Output arguments of FFmpegMultiTierDelegate, each formatted with the -# session values. Video is copied, audio encoded to 16 kHz mono Opus in -# 20 ms frames, which ffmpeg sends with the 48 kHz RTP clock. Video goes out -# from a free port, its announced port is the session's RtcpMonitor. -FFMPEG_VIDEO_OUTPUT = ( - *("-map", "0:v:0", "-an", "-c:v", "copy", "-bsf:v", "dump_extra"), - *("-payload_type", "{v_payload_type}", "-ssrc", "{v_ssrc}", "-f", "rtp"), - *("-srtp_out_suite", "AES_CM_128_HMAC_SHA1_80"), - *("-srtp_out_params", "{v_srtp_key}"), - "srtp://{host}:{v_port}?rtcpport={v_port}&pkt_size=1190", -) -FFMPEG_AUDIO_OUTPUT = ( - *("-map", "0:a:0", "-c:a", "libopus", "-application", "lowdelay"), - *("-frame_duration", "20", "-ar", "16000", "-ac", "1", "-b:a", "{a_bitrate}k"), - *("-payload_type", "{a_payload_type}", "-ssrc", "{a_ssrc}", "-f", "rtp"), - *("-srtp_out_suite", "AES_CM_128_HMAC_SHA1_80"), - *("-srtp_out_params", "{a_srtp_key}"), - "srtp://{host}:{a_port}?rtcpport={a_port}&localrtpport={a_local_port}&pkt_size=188", -) - - -class FFmpegMultiTierDelegate: - """Streams each session with its own ffmpeg process. - - Pass it as the ``delegate`` of the ``multi_tier`` option. The command is - ``ffmpeg``, :data:`FFMPEG_GLOBAL_OPTIONS`, the ``ffmpeg_input`` arguments, - :data:`FFMPEG_VIDEO_OUTPUT` - and, if the controller wants audio, :data:`FFMPEG_AUDIO_OUTPUT`. It is an - argument list that no shell sees. - - The video is copied whatever tier the controller picks. A tier change - keeps the running stream as it is, the delegate does not adapt it to the - tier. - - The audio output maps the first audio stream of the input, so ffmpeg - fails on an input without audio; pass ``audio=False`` for one, and - sessions stream video only. - - ffmpeg does not report the controller's RTCP, so the delegate listens - for it with an :class:`RtcpMonitor` on the announced video port, which - keeps the session alive. When ffmpeg exits on its own, the session ends. - """ - - def __init__( - self, ffmpeg_input: Sequence[str], ffmpeg: str = "ffmpeg", audio: bool = True - ) -> None: - self.argv = (ffmpeg, *FFMPEG_GLOBAL_OPTIONS, *ffmpeg_input) - self.audio = audio - - async def handle_prepare(self, session: MultiTierSession) -> None: - """Listen for the controller's RTCP on the announced video port. - - ffmpeg sends the audio from a free port, which the answer announces. - """ - session.data["rtcp_monitor"] = await open_rtcp_monitor(session) - session.accessory_audio_port = _free_udp_port(session.is_ipv6) - - def command( - self, - session: MultiTierSession, - video_tier: hksv3.VideoStreamTier, - audio_tier: hksv3.AudioStreamTier | None, - ) -> list[str]: - """The argument list for a session.""" - values = { - "host": f"[{session.address}]" if session.is_ipv6 else session.address, - "v_port": session.video_port, - "a_port": session.audio_port, - "a_local_port": session.accessory_audio_port, - "v_srtp_key": to_base64_str( - session.video_srtp_key + session.video_srtp_salt - ), - "a_srtp_key": to_base64_str( - session.audio_srtp_key + session.audio_srtp_salt - ), - "v_ssrc": session.video_ssrc, - "a_ssrc": session.audio_ssrc, - "v_payload_type": hksv3.VIDEO_PAYLOAD_TYPE, - "a_payload_type": hksv3.AUDIO_PAYLOAD_TYPE, - "a_bitrate": audio_tier.target_bitrate // 1000 if audio_tier else 0, - } - with_audio = self.audio and audio_tier is not None - output = FFMPEG_VIDEO_OUTPUT + (FFMPEG_AUDIO_OUTPUT if with_audio else ()) - return [*self.argv, *(arg.format(**values) for arg in output)] - - async def handle_start( - self, - session: MultiTierSession, - video_tier: hksv3.VideoStreamTier, - audio_tier: hksv3.AudioStreamTier | None, - ) -> None: - """Start ffmpeg for the session.""" - cmd = self.command(session, video_tier, audio_tier) - logger.debug("[%s] Executing: %s", session.id, _redact_command(cmd)) - process = await asyncio.create_subprocess_exec( - *cmd, - stdin=asyncio.subprocess.DEVNULL, - stdout=asyncio.subprocess.DEVNULL, - stderr=asyncio.subprocess.PIPE, - ) - session.data["process"] = process - session.data["watcher"] = asyncio.ensure_future(_watch(session, process)) - logger.info("[%s] Started stream process %d", session.id, process.pid) - - async def handle_reconfigure( - self, - session: MultiTierSession, - video_tier: hksv3.VideoStreamTier, - audio_tier: hksv3.AudioStreamTier | None, - ) -> None: - """Keep the running stream. - - A copied stream cannot change its tier, and a new ffmpeg with the - session's SSRC and SRTP key would repeat the keystream. - """ - logger.debug("[%s] Keeping the stream for tier %s", session.id, video_tier) - - async def handle_end(self, session: MultiTierSession) -> None: - """Stop the session's ffmpeg, killing it after two seconds.""" - monitor: RtcpMonitor | None = session.data.pop("rtcp_monitor", None) - if monitor is not None: - monitor.close() - process = session.data.pop("process", None) - if process is None: - return - logger.info("[%s] Stopping the stream process", session.id) - with contextlib.suppress(ProcessLookupError): - process.terminate() - try: - await asyncio.wait_for(process.wait(), 2.0) - except TimeoutError: - logger.warning("[%s] Killing the stream process", session.id) - with contextlib.suppress(ProcessLookupError): - process.kill() - await process.wait() - watcher = session.data.pop("watcher") - with contextlib.suppress(Exception): - await watcher - - -def _free_udp_port(ipv6: bool) -> int: - """A UDP port that is free now, for ffmpeg to bind. - - Two sessions of controllers that announced the same port would otherwise - both bind it. - """ - family = socket.AF_INET6 if ipv6 else socket.AF_INET - with socket.socket(family, socket.SOCK_DGRAM) as sock: - sock.bind(("::" if ipv6 else "0.0.0.0", 0)) # noqa: S104 - return int(sock.getsockname()[1]) - - -def _redact_command(cmd: list[str]) -> str: - """Join the command for logging, without the SRTP keys and input secrets.""" - return " ".join( - "" if previous == "-srtp_out_params" else _redact_arg(arg) - for previous, arg in zip(["", *cmd], cmd, strict=False) - ) - - -# User info runs up to the last "@" of the token, it may hold any character. -_URL = re.compile( - r"(?P[a-z][a-z0-9+.-]*://)(?:\S*@)?(?P[^\s'\"]*)", re.IGNORECASE -) -# An argument that is a URL is one token, its user info may hold spaces. -_URL_ARG = re.compile( - r"(?P[a-z][a-z0-9+.-]*://)(?:.*@)?(?P.*)", re.IGNORECASE | re.DOTALL -) - - -def _redact_url(match: re.Match[str]) -> str: - # Keep the host and port, drop user info, path and query. - host, *path = re.split(r"[/?#]", match["rest"], maxsplit=1) - return f"{match['scheme']}{host}{'/' if path else ''}" - - -def _redact_arg(arg: str) -> str: - """Redact a command argument, a URL argument as a whole.""" - if match := _URL_ARG.fullmatch(arg): - return _redact_url(match) - return _redact_urls(arg) - - -def _redact_urls(text: str) -> str: - """Drop user info, path and query of each URL in ``text``, e.g. for logs. - - Camera URLs often carry credentials or access tokens there. - """ - return _URL.sub(_redact_url, text) - - -async def _watch( - session: MultiTierSession, process: asyncio.subprocess.Process -) -> None: - """Log ffmpeg's output and end the session with ffmpeg.""" - output = asyncio.ensure_future(_log_output(session, process)) - try: - code = await process.wait() - with contextlib.suppress(Exception): - await output - finally: - output.cancel() - if session.data.get("process") is process: - logger.warning("[%s] Stream process exited with %s", session.id, code) - session.end("the stream process exited") - - -# ffmpeg ends status lines with "\r", so the output is read in chunks, not -# lines; a line longer than this is logged in pieces. -_OUTPUT_CHUNK = 4096 - - -async def _log_output( - session: MultiTierSession, process: asyncio.subprocess.Process -) -> None: - # Reading the output also keeps the pipe from filling up and ffmpeg from - # blocking, so it keeps reading until ffmpeg closes it. - if process.stderr is None: - return - pending = b"" - while chunk := await process.stderr.read(_OUTPUT_CHUNK): - *lines, pending = re.split(rb"[\r\n]", pending + chunk) - if len(pending) >= _OUTPUT_CHUNK: - lines.append(pending) - pending = b"" - for line in lines: - _log_line(session, line) - _log_line(session, pending) - - -def _log_line(session: MultiTierSession, line: bytes) -> None: - # Token based, user info with a raw space is not covered. - if text := _redact_urls(line.decode(errors="replace").strip()): - logger.debug("[%s] %s", session.id, text) diff --git a/pyhap/camera_stream.py b/pyhap/camera_stream.py index 38099a2c..b3c56dc9 100644 --- a/pyhap/camera_stream.py +++ b/pyhap/camera_stream.py @@ -18,7 +18,9 @@ until one deadline, ``hook_timeout`` after the write, which leaves time to answer within the HAP response timeout. A hook that takes longer is cancelled and its session ends. Once a session ended, the delegate releases -it exactly once, in the background, if it saw the session at all. +it exactly once, in the background, if it saw the session at all. Until its +release finished or timed out, the session still counts towards the session +limits. A delegate keeps its state by session object, e.g. in ``session.data``, not by session ID: a controller may set up a session with the ID of one that is @@ -32,6 +34,7 @@ from collections.abc import Coroutine, Mapping from dataclasses import dataclass, field, fields from enum import Enum, IntEnum +import functools import logging from types import MappingProxyType from typing import TYPE_CHECKING, Any, Generic, TypeVar, cast @@ -58,6 +61,10 @@ HOOK_TIMEOUT_MARGIN = 1.0 MAX_HOOK_TIMEOUT = RESPONSE_TIMEOUT - HOOK_TIMEOUT_MARGIN DEFAULT_HOOK_TIMEOUT = MAX_HOOK_TIMEOUT +# A release waits up to hook_timeout for the cancelled hooks and then up to +# hook_timeout for the delegate. close() gives up on tasks after this many +# hook timeouts. +CLOSE_TIMEOUT_FACTOR = 3 _S = TypeVar("_S", bound="StreamSession") _T = TypeVar("_T") @@ -68,9 +75,11 @@ class StreamError(Exception): """Raised by a delegate hook to answer the request with ``status``. - ``status`` is one of the request's answer, e.g. - :attr:`pyhap.hksv3.RTPStreamingStatus.BUSY`. Other exceptions, and hooks - that take too long, answer with the request's error status. + ``status`` must be a value of the status enum of the request the hook + answers, e.g. :attr:`pyhap.hksv3.RTPStreamingStatus.BUSY` for RTP + Streaming Control or :attr:`pyhap.hksv3.SetupEndpointsStatus.BUSY` for + Setup Endpoints. Another value, other exceptions and hooks that take too + long answer with the request's error status. """ def __init__(self, status: int, message: str = "") -> None: @@ -125,7 +134,7 @@ def busy(self) -> bool: def end(self, reason: str = "the camera ended it") -> None: """End the session on the camera's side, e.g. when its media stopped. - Schedules the end on the event loop and returns at once; the + Schedules the end on the event loop and returns at once. The delegate then releases the session as for any other end. Does nothing for a session that ended. Call it in the event loop. """ @@ -134,7 +143,7 @@ def end(self, reason: str = "the camera ended it") -> None: controller._loop.call_soon(controller._end, self, reason) -class _HookFailed(Exception): +class HookFailed(Exception): """A hook failed, ``status`` is the one the delegate asked for, if any. Raised by :meth:`StreamController._run_hook` to the stream services. @@ -145,7 +154,7 @@ def __init__(self, status: int | None = None) -> None: self.status = status -def _failure_status(status_type: type[_E], status: int | None, error: _E) -> _E: +def failure_status(status_type: type[_E], status: int | None, error: _E) -> _E: """The status a failed request answers, ``error`` unless the delegate chose.""" try: chosen = status_type(status) @@ -154,13 +163,13 @@ def _failure_status(status_type: type[_E], status: int | None, error: _E) -> _E: return chosen if chosen else error -def _check_hook_timeout(hook_timeout: float) -> None: +def check_hook_timeout(hook_timeout: float) -> None: """Make sure a hook leaves time to answer within the HAP response timeout.""" if not 0 < hook_timeout <= MAX_HOOK_TIMEOUT: raise ValueError(f"hook_timeout must be above 0 and at most {MAX_HOOK_TIMEOUT}") -def _config_from_option(config_type: type[_C], value: Any, name: str) -> _C: +def config_from_option(config_type: type[_C], value: Any, name: str) -> _C: """Build a config dataclass from an option dict, or take it as it is. :raises ValueError: For an unknown key or a value of another type. @@ -188,14 +197,18 @@ def __init__( max_sessions: int, max_sessions_per_connection: int, ) -> None: - _check_hook_timeout(hook_timeout) + check_hook_timeout(hook_timeout) self.camera = camera self.sensor = sensor self.hook_timeout = hook_timeout self.max_sessions = max_sessions self.max_sessions_per_connection = max_sessions_per_connection self._sessions: dict[UUID, _S] = {} + # Ended sessions whose release still runs, they keep their room. + self._releasing: set[_S] = set() self._tasks: set[asyncio.Task[Any]] = set() + # Background tasks nobody waits for any more, their errors are warned. + self._abandoned: set[asyncio.Task[Any]] = set() self._closed = False @property @@ -225,11 +238,10 @@ def _no_room(self, client_addr: tuple[str, int] | None) -> str | None: """Why a new session of ``client_addr`` does not fit, None if it does.""" if self._closed: return f"The {self.kind} service is closed" - if len(self._sessions) >= self.max_sessions: + sessions = [*self._sessions.values(), *self._releasing] + if len(sessions) >= self.max_sessions: return f"Too many {self.kind} sessions" - owned = sum( - session.client_addr == client_addr for session in self._sessions.values() - ) + owned = sum(session.client_addr == client_addr for session in sessions) if owned >= self.max_sessions_per_connection: return f"Too many {self.kind} sessions of the connection" return None @@ -262,17 +274,20 @@ async def _run_hook( ``first`` marks the hook that introduces the session to the delegate, which then releases it when it ends. - :raises _HookFailed: If the hook failed or took too long, or the + :raises HookFailed: If the hook failed or took too long, or the session ended meanwhile. """ if session.ended: hook.close() - raise _HookFailed + raise HookFailed if first: session._delegated = True task = self._loop.create_task(hook) session._tasks.add(task) task.add_done_callback(session._tasks.discard) + task.add_done_callback( + functools.partial(self._retrieve, session, f"The {name} hook") + ) try: done, _ = await asyncio.wait( {task}, timeout=max(deadline - self._loop.time(), 0) @@ -284,9 +299,15 @@ async def _run_hook( if not done: logger.warning("[%s] The %s hook timed out", session.id, name) self._end(session, f"the {name} hook timed out") - raise _HookFailed - if task.cancelled() or session.ended: - raise _HookFailed + raise HookFailed + if session.ended: + raise HookFailed + if task.cancelled(): + # The hook cancelled itself, not the end of its session. + logger.warning("[%s] The %s hook was cancelled", session.id, name) + if end_on_failure: + self._end(session, f"the {name} hook was cancelled") + raise HookFailed error = task.exception() if error is None: return task.result() @@ -298,14 +319,36 @@ async def _run_hook( logger.error("[%s] The %s hook failed", session.id, name, exc_info=error) if end_on_failure: self._end(session, f"the {name} hook failed") - raise _HookFailed(error.status if isinstance(error, StreamError) else None) + raise HookFailed(error.status if isinstance(error, StreamError) else None) - def _background(self, coro: Coroutine[Any, Any, _T]) -> asyncio.Task[_T]: + def _background( + self, session: _S, label: str, coro: Coroutine[Any, Any, _T] + ) -> asyncio.Task[_T]: + """Run ``coro`` as a task of the controller, ``label`` names it in logs.""" task = self._loop.create_task(coro) self._tasks.add(task) task.add_done_callback(self._tasks.discard) + # A release that timed out is not awaited any more. + task.add_done_callback(functools.partial(self._retrieve, session, label)) return task + def _retrieve(self, session: _S, label: str, task: asyncio.Task[Any]) -> None: + """Retrieve the exception of a hook or background task, also unawaited. + + Nobody reports the error of an abandoned task, e.g. a release still + running at its timeout, so it is warned about. Other errors are only + logged at debug, their waiter reports them or their session ended. + """ + abandoned = task in self._abandoned + self._abandoned.discard(task) + if not task.cancelled() and (error := task.exception()) is not None: + if abandoned: + logger.warning( + "[%s] %s failed after its timeout: %r", session.id, label, error + ) + else: + logger.debug("[%s] %s ended with %r", session.id, label, error) + def _end(self, session: _S, reason: str) -> asyncio.Task[bool] | None: """End ``session`` now and release it in the background. @@ -324,10 +367,20 @@ def _end(self, session: _S, reason: str) -> asyncio.Task[bool] | None: logger.debug("[%s] %s session ended: %s", session.id, self.kind, reason) self._session_ended(session, previous) if session._delegated: - session._release = self._background(self._release(session, hooks)) + self._releasing.add(session) + session._release = self._background( + session, "The release task", self._release(session, hooks) + ) return session._release async def _release(self, session: _S, hooks: list[asyncio.Task[Any]]) -> bool: + """Release ``session``, False if the delegate's release timed out.""" + try: + return await self._release_after(session, hooks) + finally: + self._releasing.discard(session) + + async def _release_after(self, session: _S, hooks: list[asyncio.Task[Any]]) -> bool: if hooks: # Cancelled hooks finish before the release starts, unless one # ignores its cancellation. @@ -338,19 +391,30 @@ async def _release(self, session: _S, hooks: list[asyncio.Task[Any]]) -> bool: session.id, len(pending), ) - try: - await asyncio.wait_for(self._release_hook(session), self.hook_timeout) - except TimeoutError: + release = self._background(session, "The end hook", self._release_hook(session)) + done, _ = await asyncio.wait({release}, timeout=self.hook_timeout) + if not done: + # Not awaited: a release that ignores its cancellation must not + # keep the room or the camera's stop. + release.cancel() + self._abandoned.add(release) logger.warning("[%s] Releasing the session timed out", session.id) return False - # The delegate may fail in many ways, the session is gone anyway. - except Exception: - logger.exception("[%s] Releasing the session failed", session.id) - return False + # The release finished, so the session is gone even if it failed. + if release.cancelled(): + logger.warning("[%s] Releasing the session was cancelled", session.id) + elif (error := release.exception()) is not None: + logger.error( + "[%s] Releasing the session failed", session.id, exc_info=error + ) return True async def _end_by_request(self, session: _S, deadline: float) -> bool: - """End a session for a request, waiting for its release until ``deadline``.""" + """End a session for a request, waiting for its release until ``deadline``. + + Returns False if the release did not finish by ``deadline``. The + session then keeps its room until its release finished or timed out. + """ release = self._end(session, "the controller ended it") if release is None: return True @@ -395,30 +459,6 @@ def _end_sessions(self, reason: str) -> None: for session in list(self._sessions.values()): self._end(session, reason) - async def end_session(self, session_id: UUID) -> bool: - """End a session, e.g. when its stream ended on the camera's side. - - Waits until the delegate released it. Returns False if there is no - such session. - """ - session = self._sessions.get(session_id) - if session is None: - return False - release = self._end(session, "the camera ended it") - if release is not None: - await asyncio.shield(release) - return True - - async def end_all_sessions(self) -> None: - """End all sessions and wait until the delegate released them.""" - releases = [ - self._end(session, "the camera ended it") - for session in list(self._sessions.values()) - ] - await asyncio.gather( - *(asyncio.shield(release) for release in releases if release is not None) - ) - def client_disconnected(self, client_addr: tuple[str, int]) -> None: """End the sessions of a connection that closed.""" for session in list(self._sessions.values()): @@ -426,11 +466,22 @@ def client_disconnected(self, client_addr: tuple[str, int]) -> None: self._end(session, "the controller disconnected") async def close(self) -> None: - """End all sessions and wait for every task, e.g. when the camera stops. + """End all sessions and wait for their releases, e.g. when the camera stops. - No new session fits afterwards. + Gives up on tasks that still run after three times ``hook_timeout``, + e.g. a release that ignores its cancellation. No new session fits + afterwards. """ self._closed = True self._end_sessions("the camera stopped") + deadline = self._loop.time() + CLOSE_TIMEOUT_FACTOR * self.hook_timeout while pending := [task for task in self._tasks if not task.done()]: - await asyncio.wait(pending) + remaining = deadline - self._loop.time() + if remaining <= 0: + logger.warning( + "Closing the %s service while %d tasks still run", + self.kind, + len(pending), + ) + return + await asyncio.wait(pending, timeout=remaining) diff --git a/pyhap/camera_webrtc.py b/pyhap/camera_webrtc.py index 774ad4a4..a2fb64a3 100644 --- a/pyhap/camera_webrtc.py +++ b/pyhap/camera_webrtc.py @@ -7,7 +7,7 @@ its ``webrtc`` option, next to the multi-tier RTP service. Controllers view remotely through it. The accessory offers, the controller answers, may renegotiate and update its SFrame keys, and ends the session. A -:class:`WebRTCDelegate` runs the WebRTC media; it shares +:class:`WebRTCDelegate` runs the WebRTC media. It shares :class:`pyhap.hksv3.WebRTCOffer`, :class:`pyhap.hksv3.WebRTCReofferAnswer`, :class:`pyhap.hksv3.IceCandidate` and :class:`pyhap.hksv3.SFrameKey` with the HAP side. @@ -24,13 +24,13 @@ from pyhap import hksv3 from pyhap.camera_stream import ( DEFAULT_HOOK_TIMEOUT, + HookFailed, SessionState, StreamController, StreamSession, - _check_hook_timeout, - _config_from_option, - _failure_status, - _HookFailed, + check_hook_timeout, + config_from_option, + failure_status, ) from pyhap.util import base64_to_bytes, to_base64_str @@ -77,10 +77,12 @@ class WebRTCDelegate(Protocol): with a status, any other exception answers with an error. A hook that takes longer than ``hook_timeout`` is cancelled and its session ends. - An answered session has no timeout of its own: remote viewers share the - home hub's connection, so its close ends nothing. The delegate must call - ``session.end()`` when the media connection fails, e.g. on ICE or consent - failure, or the session keeps one of the sensor's slots. + An answered session has no timeout of its own. When the home hub's + connection closes its sessions end, but remote viewers share that + connection, so a viewer that leaves closes nothing on the HAP side. The + delegate must call ``session.end()`` when the media connection fails or + the viewer leaves, e.g. on ICE or consent failure or a DTLS close, or the + session keeps one of the sensor's slots. """ async def handle_solicit_offer( @@ -141,7 +143,7 @@ class WebRTCConfig: delegate hook gets ``hook_timeout`` seconds. A session the controller did not answer within ``answer_timeout`` seconds ends. The offer advertises the sensor's video tier of ``remote_quality``, the lowest tier - if it has none. Medium keeps 4K off the remote path; the Scrypted HomeKit + if it has none. Medium keeps 4K off the remote path. The Scrypted HomeKit plugin offers the first tier. """ @@ -158,10 +160,10 @@ def from_option(cls, value: WebRTCConfig | Mapping[str, Any]) -> WebRTCConfig: :raises ValueError: For an unknown key. """ - return _config_from_option(cls, value, "webrtc") + return config_from_option(cls, value, "webrtc") def __post_init__(self) -> None: - _check_hook_timeout(self.hook_timeout) + check_hook_timeout(self.hook_timeout) if self.max_sessions < WEBRTC_MIN_SESSIONS: raise ValueError( f"webrtc max_sessions must be at least {WEBRTC_MIN_SESSIONS}" @@ -177,7 +179,7 @@ def __post_init__(self) -> None: raise ValueError(f"Unknown webrtc remote_quality {self.remote_quality}") -def remote_tier( +def _remote_tier( tiers: list[hksv3.VideoStreamTier], quality: hksv3.VideoQuality | int ) -> hksv3.VideoStreamTier: """The smallest tier of ``quality``, else the smallest tier.""" @@ -222,7 +224,7 @@ def __init__( self.config = config self.delegate = config.delegate self.audio_tier = multi_tier.config.audio - self.video_tier = remote_tier(multi_tier.tiers, config.remote_quality) + self.video_tier = _remote_tier(multi_tier.tiers, config.remote_quality) service = self.service = camera.add_preload_service( "CameraWebRTCStreamManagement", unique_id=str(self.sensor.sensor_uuid) ) @@ -246,9 +248,13 @@ def __init__( ("WebRTCReoffer", self._set_reoffer), ("WebRTCUpdateSession", self._set_update_session), ): - service.configure_char(name, setter_callback=setter) + # The requests carry SDPs with ICE passwords and SFrame keys, so + # neither they nor the answers are shared with other connections. + service.configure_char( + name, setter_callback=setter + ).read_back_response = True - def refusal(self) -> hksv3.WebRTCSolicitOfferStatus | None: + def _refusal(self) -> hksv3.WebRTCSolicitOfferStatus | None: """Why the sensor may not stream now, None if it may.""" if not self.camera.homekit_camera_active: return hksv3.WebRTCSolicitOfferStatus.PRIVACY_MODE_ACTIVE @@ -258,7 +264,7 @@ def refusal(self) -> hksv3.WebRTCSolicitOfferStatus | None: def update_status(self) -> None: """End the sessions if the sensor may no longer stream.""" - if self.refusal() is not None: + if self._refusal() is not None: self._end_sessions("streaming was turned off") def _session_ended(self, session: WebRTCSession, previous: SessionState) -> None: @@ -305,7 +311,7 @@ async def _solicit_offer( deadline: float, ) -> hksv3.WebRTCSolicitOfferResponse: error = hksv3.WebRTCSolicitOfferStatus.ERROR - refusal = self.refusal() + refusal = self._refusal() if refusal is not None: logger.info("Refusing a WebRTC offer, streaming is off: %s", refusal) return hksv3.WebRTCSolicitOfferResponse(refusal) @@ -345,10 +351,14 @@ async def _solicit_offer( end_on_failure=True, first=True, ) - except _HookFailed as err: + except HookFailed as err: return hksv3.WebRTCSolicitOfferResponse( - _failure_status(hksv3.WebRTCSolicitOfferStatus, err.status, error) + failure_status(hksv3.WebRTCSolicitOfferStatus, err.status, error) ) + if not isinstance(offer, hksv3.WebRTCOffer): + logger.error("[%s] The offer is not a WebRTCOffer: %r", session.id, offer) + self._end(session, "the offer is not a WebRTCOffer") + return hksv3.WebRTCSolicitOfferResponse(error) if (offer.sframe is not None) != session.sframe: logger.error("[%s] The offer's SFrame key does not fit", session.id) self._end(session, "the offer's SFrame key does not fit") @@ -384,9 +394,9 @@ async def _set_provide_answer( ) try: await self._run_hook(session, "answer", hook, deadline, end_on_failure=True) - except _HookFailed as err: + except HookFailed as err: return _status( - session.id, _failure_status(_STATUS, err.status, _STATUS.ERROR) + session.id, failure_status(_STATUS, err.status, _STATUS.ERROR) ) # Only now, the expiry ends a session whose answer did not apply. self._cancel_timer(session) @@ -449,10 +459,18 @@ async def _reoffer( ) try: answer = await self._run_hook(session, "reoffer", hook, deadline) - except _HookFailed as err: + except HookFailed as err: return hksv3.WebRTCReofferResponse( - session.id, _failure_status(_STATUS, err.status, _STATUS.ERROR) + session.id, failure_status(_STATUS, err.status, _STATUS.ERROR) + ) + # Like a failed hook, a bad answer leaves the session as it was. + if not isinstance(answer, hksv3.WebRTCReofferAnswer): + logger.error( + "[%s] The reoffer answer is not a WebRTCReofferAnswer: %r", + session.id, + answer, ) + return hksv3.WebRTCReofferResponse(session.id, _STATUS.ERROR) if answer.sframe is not None and not session.sframe: logger.error( "[%s] The reoffer answer's SFrame key does not fit", session.id @@ -477,8 +495,8 @@ async def _set_update_session( ) try: await self._run_hook(session, "update session", hook, deadline) - except _HookFailed as err: + except HookFailed as err: return _status( - session.id, _failure_status(_STATUS, err.status, _STATUS.ERROR) + session.id, failure_status(_STATUS, err.status, _STATUS.ERROR) ) return _status(session.id, _STATUS.SUCCESS) diff --git a/pyhap/hksv3.py b/pyhap/hksv3.py index 713eb715..f07ece02 100644 --- a/pyhap/hksv3.py +++ b/pyhap/hksv3.py @@ -5,11 +5,11 @@ Follows the HAP TLV8 list encoding: each item of a list is a TLV of its own, and items of the same type are separated by an empty TLV of type 0. Values -longer than 255 bytes are sent as 255-byte fragments of the same type; a TLV +longer than 255 bytes are sent as 255-byte fragments of the same type. A TLV continues the value before it only after such a full fragment. :mod:`pyhap.tlv` maps each type to one value, so it cannot express repeated -items of a type or the separators these lists need; hence the small encoder +items of a type or the separators these lists need. Hence the small encoder and decoder here. """ @@ -447,16 +447,25 @@ class RTPStreamingControl: @classmethod def decode(cls, data: bytes) -> RTPStreamingControl: fields = _fields(data) + video_ssrc = _optional_ssrc(fields, 4) + audio_ssrc = _optional_ssrc(fields, 6) return cls( UUID(bytes=_required(fields, 1)), _to_enum(RTPStreamingCommand, _required(fields, 2)), _optional_int(fields, 3), - _optional_int(fields, 4), + video_ssrc, _optional_int(fields, 5), - _optional_int(fields, 6), + audio_ssrc, ) +def _optional_ssrc(fields: dict[int, list[bytes]], tag: int) -> int | None: + ssrc = _optional_int(fields, tag) + if ssrc is not None and ssrc > 0xFFFFFFFF: + raise ValueError(f"SSRC {ssrc} is out of range") + return ssrc + + def encode_streaming_status( session_id: UUID | None, status: RTPStreamingStatus | int ) -> bytes: @@ -650,16 +659,17 @@ def tiers_from_sensor(sensor: SensorConfiguration) -> list[VideoStreamTier]: DEFAULT_AUDIO_TIER = AudioStreamTier(1, 24000, AudioSampleRate.KHZ_48) -# Average and maximum kbps of the guide's bit rate table, by the short side -# of the frame; the aspect ratio does not matter. A frame between two rows -# takes the larger one. +# Average and maximum kbps of the guide's bit rate table, by pixel count: +# each row is keyed by the largest frame of its class in the guide's table of +# resolutions, whatever the aspect ratio. 2880x2880 counts as 4K, 2048x1536 +# as 2K and 1440x1080 as 720p, as in that table. A frame between two rows +# takes the larger one, and a frame above 4K is held to the 4K row. _GUIDE_BITRATES = { - 2160: (4500, 5000), - 1440: (2800, 3000), - 1080: (1700, 1800), - 720: (768, 800), - 360: (180, 190), - 240: (180, 190), + 3840 * 2160: (4500, 5000), + 2560 * 1440: (2800, 3000), + 1920 * 1080: (1700, 1800), + 1440 * 1080: (768, 800), + 640 * 480: (180, 190), } @@ -684,8 +694,9 @@ def check_video_tiers(sensor: SensorConfiguration) -> list[str]: """Check the video encodings of a sensor as tiers against the guide. Returns warnings for what the guide only recommends: a gap between peak - and average of more than 10 % of the peak, peaks above its table and - frames larger than 4K, which the table does not cover. + and average of more than 10 % of the peak and peaks above its table. A + frame larger than 4K, which the table does not cover, is held to its 4K + row. :raises ValueError: For a sensor without encodings, an unknown quality, Highest without High or a peak below the average. @@ -704,12 +715,12 @@ def check_video_tiers(sensor: SensorConfiguration) -> list[str]: raise ValueError(f"{name}: peak bit rate below the average") if (stream.peak_bitrate - stream.average_bitrate) * 10 > stream.peak_bitrate: warnings.append(f"{name}: average more than 10 % below the peak") - short_side = min(stream.width, stream.height) - rows = [row for row in sorted(_GUIDE_BITRATES) if row >= short_side] - if not rows: - warnings.append(f"{name}: larger than the guide's 4K bit rate table") - continue - guide = _GUIDE_BITRATES[rows[0]] + pixels = stream.width * stream.height + row = min( + (row for row in _GUIDE_BITRATES if row >= pixels), + default=max(_GUIDE_BITRATES), + ) + guide = _GUIDE_BITRATES[row] if stream.peak_bitrate > guide[1]: warnings.append(f"{name}: peak above the guide's {guide[1]} kbps") return warnings @@ -871,10 +882,10 @@ class WebRTCOffer: This, :class:`WebRTCReofferAnswer`, :class:`IceCandidate` and :class:`SFrameKey` are the types WebRTC media and the HAP side share. - ``sdp`` carries all ICE candidates, HAP has no trickle ICE; the candidate - list of the Solicit Offer answer stays empty, as in the Scrypted HomeKit - plugin. ``sframe`` is the key the accessory sends with, see - :func:`generate_sframe_key`; None for a session without SFrame. + ``sdp`` carries all ICE candidates because HAP has no trickle ICE. The + candidate list of the Solicit Offer answer stays empty, as in the Scrypted + HomeKit plugin. ``sframe`` is the key the accessory sends with, see + :func:`generate_sframe_key`, or None for a session without SFrame. """ sdp: str diff --git a/tests/conftest.py b/tests/conftest.py index 35cc46e1..5f041320 100644 --- a/tests/conftest.py +++ b/tests/conftest.py @@ -3,7 +3,7 @@ import asyncio from collections.abc import AsyncIterator, Callable, Iterator import pathlib -from typing import Any, cast +from typing import Any from unittest.mock import AsyncMock, MagicMock, patch from uuid import UUID, uuid4 @@ -196,54 +196,3 @@ def _make(**options: Any) -> Camera: ) return _make - - -class _FFmpegStderr: - """The stderr pipe of a fake ffmpeg, it ends when the process exits.""" - - def __init__(self, data: bytes, exited: asyncio.Event) -> None: - self._data = data - self._exited = exited - - async def read(self, n: int = -1) -> bytes: - if not self._data: - await self._exited.wait() - return b"" - size = len(self._data) if n < 0 else n - chunk, self._data = self._data[:size], self._data[size:] - await asyncio.sleep(0) - return chunk - - -class _FFmpegProcess: - """An ffmpeg process that runs until it is terminated or exits.""" - - def __init__(self, *lines: bytes) -> None: - self.pid = 42 - self.returncode: int | None = None - self._exited = asyncio.Event() - self.stderr = _FFmpegStderr(b"".join(lines), self._exited) - self.terminate = MagicMock(side_effect=lambda: self.exit(-15)) - self.kill = MagicMock(side_effect=lambda: self.exit(-9)) - - def exit(self, code: int) -> None: - self.returncode = code - self._exited.set() - - async def wait(self) -> int: - await self._exited.wait() - return cast(int, self.returncode) - - -@pytest.fixture -def ffmpeg_process_factory() -> type[_FFmpegProcess]: - """Make fake ffmpeg processes with the given output.""" - return _FFmpegProcess - - -@pytest.fixture -def ffmpeg_process() -> _FFmpegProcess: - """A running ffmpeg process, its output has a frame and an input line.""" - return _FFmpegProcess( - b"frame=1\n", b"Input #0, rtsp, from 'rtsp://user:secret@cam/stream-token':\n" - ) diff --git a/tests/test_camera_multi_tier.py b/tests/test_camera_multi_tier.py index 098d1d2c..d3dd2001 100644 --- a/tests/test_camera_multi_tier.py +++ b/tests/test_camera_multi_tier.py @@ -2,26 +2,16 @@ import asyncio from collections.abc import Callable -import logging import struct from typing import Any, cast -from unittest.mock import ANY, AsyncMock, MagicMock, patch +from unittest.mock import AsyncMock, MagicMock, patch from uuid import UUID import pytest from pyhap import hksv3, tlv from pyhap.camera import SETUP_ADDR_INFO, SETUP_SRTP_PARAM, SETUP_TYPES, Camera -from pyhap.camera_multi_tier import ( - FFMPEG_GLOBAL_OPTIONS, - FFmpegMultiTierDelegate, - MultiTierController, - MultiTierSession, - RtcpMonitor, - _redact_command, - _redact_urls, - open_rtcp_monitor, -) +from pyhap.camera_multi_tier import MultiTierController, MultiTierSession from pyhap.camera_stream import SessionState, StreamError from pyhap.camera_webrtc import WebRTCController from pyhap.util import base64_to_bytes, to_base64_str @@ -179,54 +169,18 @@ async def test_streaming_enabled_persisted( async def test_camera_forwards_to_controllers( stream_camera: Callable[..., Camera], ) -> None: - """Test the camera's session actions reach each controller.""" + """Test the camera's stop and disconnect reach each controller.""" camera = stream_camera() - controllers = [ - *camera.multi_tier_controllers.values(), - *camera.webrtc_controllers.values(), - ] with ( patch.object(MultiTierController, "close") as close, + patch.object(WebRTCController, "close") as webrtc_close, patch.object(MultiTierController, "client_disconnected") as disconnected, - patch.object(MultiTierController, "end_session", return_value=True) as end, - patch.object(MultiTierController, "end_all_sessions") as end_all, ): await camera.stop() camera.client_disconnected(CLIENT) - assert await camera.end_session(SESSION) is True - await camera.end_all_sessions() close.assert_awaited_once() + webrtc_close.assert_awaited_once() disconnected.assert_called_once_with(CLIENT) - end.assert_awaited_once_with(SESSION) - end_all.assert_awaited_once() - assert len(controllers) == 2 - - -@pytest.mark.parametrize( - ("multi_tier_ended", "webrtc_ended", "expected"), - [(True, True, True), (False, True, True), (False, False, False)], -) -async def test_camera_ends_the_id_in_each_controller( - stream_camera: Callable[..., Camera], - multi_tier_ended: bool, - webrtc_ended: bool, - expected: bool, -) -> None: - """Test the camera ends a session ID in every controller, not the first only.""" - camera = stream_camera() - with ( - patch.object( - MultiTierController, "end_session", return_value=multi_tier_ended - ) as multi_tier_end, - patch.object( - WebRTCController, "end_session", return_value=webrtc_ended - ) as webrtc_end, - ): - assert await camera.end_session(SESSION) is expected - await camera.stop() - - multi_tier_end.assert_awaited_once_with(SESSION) - webrtc_end.assert_awaited_once_with(SESSION) # Sessions @@ -284,10 +238,14 @@ def _session(camera: Camera) -> MultiTierSession: def _control( - command: int, video_tier: int | None = 1, audio_tier: int | None = 1 + command: int, + video_tier: int | None = 1, + audio_tier: int | None = 1, + session_id: UUID = SESSION, ) -> str: - """An RTP Streaming Control write for SESSION.""" - data = hksv3.encode_item(1, SESSION.bytes) + hksv3.encode_item(2, bytes([command])) + """An RTP Streaming Control write, by default for SESSION.""" + data = hksv3.encode_item(1, session_id.bytes) + data += hksv3.encode_item(2, bytes([command])) if video_tier is not None: data += hksv3.encode_item(3, video_tier.to_bytes(4, "little")) data += hksv3.encode_item(4, (0x1234).to_bytes(4, "little")) @@ -318,6 +276,20 @@ def _setup_request_of(session_id: UUID) -> str: return to_base64_str(request[:2] + session_id.bytes + request[18:]) +async def _setup_and_end_many(camera: Camera, count: int) -> list[int]: + """Set up ``count`` sessions and end each before it starts. + + Returns the Setup Endpoints statuses. + """ + statuses: list[int] = [] + for index in range(count): + session_id = UUID(int=index + 1) + answer = await _setup(camera, _setup_request_of(session_id)) + statuses.append(answer[SETUP_TYPES["STATUS"]][0]) + await _streaming_control(camera, command=1, session_id=session_id) + return statuses + + async def _setup_many(camera: Camera, count: int) -> list[int]: """Set up ``count`` sessions with other IDs, return the answer statuses.""" statuses: list[int] = [] @@ -358,7 +330,7 @@ async def test_setup_answers_from_the_connection( ) -> None: """Test the answer has the accessory's address on the connection.""" camera = stream_camera() - with patch.object(camera.driver, "local_address", return_value="fd00::10"): + with patch.object(camera.driver, "_local_address", return_value="fd00::10"): answer = await _setup(camera, _setup_request(b"fd00::1", True)) address = tlv.decode(answer[SETUP_TYPES["ADDRESS"]]) assert address[SETUP_ADDR_INFO["ADDRESS_VER"]] == b"\x01" @@ -381,6 +353,132 @@ async def _prepare(session: MultiTierSession) -> None: assert address[SETUP_ADDR_INFO["VIDEO_RTP_PORT"]] == struct.pack(" Callable[..., Any]: + """A prepare hook that sets the accessory's ports.""" + + async def _prepare(session: MultiTierSession) -> None: + session.accessory_video_port = video + session.accessory_audio_port = audio + + return _prepare + + +@pytest.mark.parametrize( + ("video", "audio"), + [(0, 50002), (50000, 65536), ("50000", 50002), (50000, 50002.0), (True, 1)], + ids=["zero", "too-large", "str", "float", "bool"], +) +async def test_prepare_sets_invalid_ports( + stream_camera: Callable[..., Camera], + multi_tier_delegate: AsyncMock, + caplog: pytest.LogCaptureFixture, + video: Any, + audio: Any, +) -> None: + """Test ports a prepare hook sets outside 1 to 65535 fail the setup at once.""" + multi_tier_delegate.handle_prepare.side_effect = _prepare_ports(video, audio) + camera = stream_camera() + + answer = await _setup(camera) + sessions = dict(_controller(camera).sessions) + await camera.stop() + + assert answer[SETUP_TYPES["STATUS"]] == b"\x02" + assert sessions == {} + assert "The prepare hook set invalid ports" in caplog.text + multi_tier_delegate.handle_end.assert_awaited_once() + + +def _prepare_address(address: Any) -> Callable[..., Any]: + """A prepare hook that sets the accessory's address.""" + + async def _prepare(session: MultiTierSession) -> None: + session.accessory_address = address + + return _prepare + + +@pytest.mark.parametrize( + "address", + [ + None, + "fd00::10", + "not an address", + "192.168.1.256", + 3232235796, + b"\xc0\xa8\x01\x14", + ], + ids=["none", "wrong-family", "garbage", "out-of-range", "int", "packed"], +) +async def test_prepare_sets_invalid_address( + stream_camera: Callable[..., Camera], + multi_tier_delegate: AsyncMock, + caplog: pytest.LogCaptureFixture, + address: Any, +) -> None: + """Test an address a prepare hook sets that is no IPv4 address fails the setup.""" + multi_tier_delegate.handle_prepare.side_effect = _prepare_address(address) + camera = stream_camera() + + answer = await _setup(camera) + sessions = dict(_controller(camera).sessions) + await camera.stop() + + assert answer[SETUP_TYPES["STATUS"]] == b"\x02" + assert sessions == {} + assert "The accessory address" in caplog.text + multi_tier_delegate.handle_end.assert_awaited_once() + + +async def test_prepare_sets_ipv6_address( + stream_camera: Callable[..., Camera], multi_tier_delegate: AsyncMock +) -> None: + """Test an IPv6 session takes an IPv6 address from the prepare hook.""" + multi_tier_delegate.handle_prepare.side_effect = _prepare_address("fd00::20") + camera = stream_camera() + with patch.object(camera.driver, "_local_address", return_value="fd00::10"): + answer = await _setup(camera, _setup_request(b"fd00::1", True)) + await camera.stop() + address = tlv.decode(answer[SETUP_TYPES["ADDRESS"]]) + assert address[SETUP_ADDR_INFO["ADDRESS"]] == b"fd00::20" + + +@pytest.mark.parametrize( + "address", + ["192.168.1.20", "fe80::20%eth0", "::ffff:192.168.1.20"], + ids=["ipv4", "scope-id", "ipv4-mapped"], +) +async def test_prepare_sets_invalid_address_on_ipv6( + stream_camera: Callable[..., Camera], + multi_tier_delegate: AsyncMock, + caplog: pytest.LogCaptureFixture, + address: str, +) -> None: + """Test an address from the prepare hook that is no plain IPv6 address fails.""" + multi_tier_delegate.handle_prepare.side_effect = _prepare_address(address) + camera = stream_camera() + with patch.object(camera.driver, "_local_address", return_value="fd00::10"): + answer = await _setup(camera, _setup_request(b"fd00::1", True)) + sessions = dict(_controller(camera).sessions) + await camera.stop() + assert answer[SETUP_TYPES["STATUS"]] == b"\x02" + assert sessions == {} + assert "The accessory address" in caplog.text + + +async def test_prepare_sets_edge_ports( + stream_camera: Callable[..., Camera], multi_tier_delegate: AsyncMock +) -> None: + """Test ports 1 and 65535 are valid.""" + multi_tier_delegate.handle_prepare.side_effect = _prepare_ports(1, 65535) + + answer = await _setup(stream_camera()) + + address = tlv.decode(answer[SETUP_TYPES["ADDRESS"]]) + assert address[SETUP_ADDR_INFO["VIDEO_RTP_PORT"]] == struct.pack(" None: - """Test a set up again is busy when a new session took its room meanwhile.""" + """Test a new session cannot take the room of a session set up again.""" camera = stream_camera() await _setup(camera) await _setup_many(camera, 2) @@ -550,9 +648,9 @@ async def test_setup_again_finds_no_room( sessions = dict(_controller(camera).sessions) await camera.stop() - assert again[SETUP_TYPES["STATUS"]] == b"\x01" - assert new[SETUP_TYPES["STATUS"]] == b"\x00" - assert sorted(sessions) == [UUID(int=1), UUID(int=2), UUID(int=3)] + assert again[SETUP_TYPES["STATUS"]] == b"\x00" + assert new[SETUP_TYPES["STATUS"]] == b"\x01" + assert sorted(sessions) == [UUID(int=1), UUID(int=2), SESSION] assert multi_tier_delegate.handle_end.await_count == 4 @@ -615,6 +713,23 @@ async def test_setup_again_release_too_slow( multi_tier_delegate.handle_prepare.assert_awaited_once() +async def test_setup_again_release_fails( + stream_camera: Callable[..., Camera], multi_tier_delegate: AsyncMock +) -> None: + """Test a set up again proceeds when the former release failed but finished.""" + multi_tier_delegate.handle_end.side_effect = RuntimeError + camera = stream_camera() + await _setup(camera) + first = _session(camera) + + answer = await _setup(camera) + + assert answer[SETUP_TYPES["STATUS"]] == b"\x00" + assert _session(camera) is not first + assert multi_tier_delegate.handle_prepare.await_count == 2 + multi_tier_delegate.handle_end.assert_awaited_once_with(first) + + async def test_setup_of_another_controller( stream_camera: Callable[..., Camera], multi_tier_delegate: AsyncMock ) -> None: @@ -796,6 +911,39 @@ async def test_end( assert await _streaming_control(camera, command=1) == 1 +async def test_end_release_fails( + stream_camera: Callable[..., Camera], multi_tier_delegate: AsyncMock +) -> None: + """Test an end answers success when the release failed but finished.""" + multi_tier_delegate.handle_end.side_effect = RuntimeError + camera = stream_camera() + await _setup(camera) + await _streaming_control(camera) + + assert await _streaming_control(camera, command=1) == 0 + assert _controller(camera).sessions == {} + assert _controller(camera)._releasing == set() + + +async def test_end_release_too_slow( + stream_camera: Callable[..., Camera], + multi_tier_delegate: AsyncMock, + end_gate: asyncio.Event, +) -> None: + """Test an end answers an error when the release overruns.""" + camera = stream_camera( + multi_tier={"delegate": multi_tier_delegate, "hook_timeout": 0.05} + ) + await _setup(camera) + await _streaming_control(camera) + + status = await _streaming_control(camera, command=1) + end_gate.set() + await camera.stop() + + assert status == 4 + + async def test_end_before_start( stream_camera: Callable[..., Camera], multi_tier_delegate: AsyncMock ) -> None: @@ -807,6 +955,49 @@ async def test_end_before_start( multi_tier_delegate.handle_end.assert_awaited_once() +async def test_releasing_sessions_keep_their_room( + stream_camera: Callable[..., Camera], + multi_tier_delegate: AsyncMock, + end_gate: asyncio.Event, +) -> None: + """Test sessions ended before they started count until they are released.""" + camera = stream_camera() + + statuses = await _setup_and_end_many(camera, 4) + end_gate.set() + await camera.stop() + + assert statuses == [0, 0, 0, 1] + assert multi_tier_delegate.handle_end.await_count == 3 + + +async def test_releasing_sessions_count_towards_max_sessions( + stream_camera: Callable[..., Camera], + multi_tier_delegate: AsyncMock, + end_gate: asyncio.Event, +) -> None: + """Test sessions being released count towards max_sessions of the sensor.""" + camera = stream_camera( + multi_tier={"delegate": multi_tier_delegate, "max_sessions_per_connection": 5} + ) + + statuses = await _setup_and_end_many(camera, 6) + end_gate.set() + await camera.stop() + + assert statuses == [0, 0, 0, 0, 0, 1] + + +async def test_setup_after_stop(stream_camera: Callable[..., Camera]) -> None: + """Test a stopped camera answers Setup Endpoints with an error, not busy.""" + camera = stream_camera() + await camera.stop() + + answer = await _setup(camera) + + assert answer[SETUP_TYPES["STATUS"]] == b"\x02" + + async def test_malformed_control(stream_camera: Callable[..., Camera]) -> None: """Test a write that does not decode answers an error.""" answer = await _controller(stream_camera())._set_streaming_control("AQ==", CLIENT) @@ -877,81 +1068,6 @@ async def test_former_session_keeps_nothing_alive( await camera.stop() -async def _send_rtcp(monitor: RtcpMonitor, host: str) -> None: - """Send a datagram from ``host`` to ``monitor`` and wait until it handled it.""" - handled = asyncio.Event() - received = monitor.datagram_received - - def _received(data: bytes, addr: tuple[Any, ...]) -> None: - received(data, addr) - handled.set() - - transport, _ = await asyncio.get_running_loop().create_datagram_endpoint( - asyncio.DatagramProtocol, remote_addr=(host, monitor.port) - ) - with patch.object(monitor, "datagram_received", _received): - transport.sendto(b"\x81\xc9\x00\x01") - await asyncio.wait_for(handled.wait(), 2) - transport.close() - - -@pytest.mark.parametrize( - ("address", "kept_alive"), - [(b"127.0.0.1", True), (b"192.168.1.114", False)], - ids=["controller", "other-host"], -) -async def test_rtcp_monitor( - stream_camera: Callable[..., Camera], address: bytes, kept_alive: bool -) -> None: - """Test a datagram from the controller keeps the session alive, others not.""" - camera = stream_camera() - await _setup(camera, _setup_request(address)) - session = _session(camera) - monitor = await open_rtcp_monitor(session) - await _streaming_control(camera) - timer = session._timer - - await _send_rtcp(monitor, "127.0.0.1") - - assert timer is not None - assert timer.cancelled() is kept_alive - assert session.accessory_video_port == monitor.port - assert monitor.port != 0 - monitor.close() - assert monitor.closed - await camera.stop() - - -async def test_rtcp_monitor_ipv6(stream_sensor: hksv3.SensorConfiguration) -> None: - """Test the monitor of an IPv6 controller listens on IPv6.""" - session = MultiTierSession( - id=SESSION, - sensor=stream_sensor, - client_addr=CLIENT, - address="::1", - is_ipv6=True, - video_port=50000, - audio_port=50002, - video_srtp_key=KEY, - video_srtp_salt=SALT, - audio_srtp_key=KEY, - audio_srtp_salt=SALT, - video_ssrc=1, - audio_ssrc=2, - accessory_address="::1", - accessory_video_port=50000, - accessory_audio_port=50002, - ) - monitor = await open_rtcp_monitor(session) - - with patch.object(session, "keep_alive") as keep_alive: - await _send_rtcp(monitor, "::1") - monitor.close() - - keep_alive.assert_called_once_with() - assert session.accessory_video_port == monitor.port - - @pytest.mark.parametrize( "end", [ @@ -994,269 +1110,5 @@ async def test_driver_local_address( connection = MagicMock() connection.transport.get_extra_info.return_value = sockname stream_driver.http_server.connections[CLIENT] = connection - assert stream_driver.local_address(CLIENT, ipv6) == expected - assert stream_driver.local_address(OTHER, ipv6) is None - - -# The ffmpeg delegate - -FFMPEG_INPUT = ["-i", "rtsp://user:secret@cam/stream-token"] - - -async def _ffmpeg_session( - stream_camera: Callable[..., Camera], request: str | None = None, **options: Any -) -> tuple[Camera, MultiTierSession, FFmpegMultiTierDelegate]: - delegate = FFmpegMultiTierDelegate(FFMPEG_INPUT) - camera = stream_camera(multi_tier={"delegate": delegate, **options}) - await _setup(camera, request) - return camera, _session(camera), delegate - - -async def test_ffmpeg_command(stream_camera: Callable[..., Camera]) -> None: - """Test the command is an argument list with video and audio outputs.""" - camera, session, delegate = await _ffmpeg_session(stream_camera) - - cmd = delegate.command(session, HIGH, hksv3.DEFAULT_AUDIO_TIER) - await camera.stop() - - assert cmd[:7] == ["ffmpeg", *FFMPEG_GLOBAL_OPTIONS, *FFMPEG_INPUT] - assert "-nostats" in cmd - assert cmd[cmd.index("-ssrc") + 1] == str(session.video_ssrc) - assert cmd[cmd.index("-srtp_out_params") + 1] == to_base64_str(KEY + SALT) - assert cmd[cmd.index("-payload_type") + 1] == "99" - assert cmd[cmd.index("-b:a") + 1] == "24k" - assert cmd[cmd.index("libopus") - 3 : cmd.index("libopus")] == [ - "-map", - "0:a:0", - "-c:a", - ] - assert cmd[-1] == ( - "srtp://192.168.1.114:50002?rtcpport=50002" - f"&localrtpport={session.accessory_audio_port}&pkt_size=188" - ) - assert "srtp://192.168.1.114:50000?rtcpport=50000&pkt_size=1190" in cmd - - -async def test_ffmpeg_command_video_only( - stream_camera: Callable[..., Camera], -) -> None: - """Test a stream without audio has no audio output.""" - camera, session, delegate = await _ffmpeg_session(stream_camera) - cmd = delegate.command(session, HIGH, None) - await camera.stop() - assert "libopus" not in cmd - assert cmd[-1].startswith("srtp://192.168.1.114:50000?") - - -async def test_ffmpeg_command_input_without_audio( - stream_camera: Callable[..., Camera], -) -> None: - """Test audio=False streams video only even when the controller wants audio.""" - camera, session, delegate = await _ffmpeg_session(stream_camera) - video_only = FFmpegMultiTierDelegate(FFMPEG_INPUT, audio=False) - - cmd = video_only.command(session, HIGH, hksv3.DEFAULT_AUDIO_TIER) - await camera.stop() - - assert cmd == delegate.command(session, HIGH, None) - assert "0:a:0" not in cmd - - -async def test_ffmpeg_command_ipv6(stream_camera: Callable[..., Camera]) -> None: - """Test an IPv6 controller is bracketed in the URLs.""" - with patch( - "pyhap.accessory_driver.AccessoryDriver.local_address", return_value="fd00::10" - ): - camera, session, delegate = await _ffmpeg_session( - stream_camera, _setup_request(b"fd00::1", True) - ) - await camera.stop() - assert delegate.command(session, HIGH, None)[-1].startswith("srtp://[fd00::1]:") - - -async def test_ffmpeg_start_and_stop( - stream_camera: Callable[..., Camera], - ffmpeg_process: Any, - caplog: pytest.LogCaptureFixture, -) -> None: - """Test ffmpeg starts once, keeps running on a tier change and stops.""" - camera, session, delegate = await _ffmpeg_session(stream_camera) - expected = delegate.command(session, HIGH, hksv3.DEFAULT_AUDIO_TIER) - with ( - patch("asyncio.create_subprocess_exec", return_value=ffmpeg_process) as start, - caplog.at_level(logging.DEBUG, logger="pyhap.camera_multi_tier"), - ): - assert await _streaming_control(camera) == 0 - assert await _streaming_control(camera, video_tier=2) == 0 - assert await _streaming_control(camera, command=1) == 0 - - start.assert_called_once_with(*expected, stdin=ANY, stdout=ANY, stderr=ANY) - ffmpeg_process.terminate.assert_called_once() - assert "frame=1" in caplog.text - assert "from 'rtsp://cam/'" in caplog.text - assert to_base64_str(KEY + SALT) not in caplog.text - assert "secret" not in caplog.text - assert "stream-token" not in caplog.text - assert "exited" not in caplog.text - - -async def test_ffmpeg_exit_ends_the_session( - stream_camera: Callable[..., Camera], - ffmpeg_process: Any, - caplog: pytest.LogCaptureFixture, -) -> None: - """Test the session ends when ffmpeg exits on its own.""" - camera, _, _ = await _ffmpeg_session(stream_camera) - with patch("asyncio.create_subprocess_exec", return_value=ffmpeg_process): - assert await _streaming_control(camera) == 0 - - ffmpeg_process.exit(1) - await asyncio.sleep(0.01) - - assert _controller(camera).sessions == {} - assert "Stream process exited with 1" in caplog.text - await camera.stop() - - -async def test_ffmpeg_status_lines_keep_the_pipe_drained( - stream_camera: Callable[..., Camera], - ffmpeg_process_factory: Any, - caplog: pytest.LogCaptureFixture, -) -> None: - """Test more than 64 KiB of carriage return status lines are read.""" - status = b"".join(b"frame=%d fps=30\r" % i for i in range(5000)) - process = ffmpeg_process_factory(status, b"last\n") - assert len(status) > 65536 - camera, _, _ = await _ffmpeg_session(stream_camera) - with ( - patch("asyncio.create_subprocess_exec", return_value=process), - caplog.at_level(logging.DEBUG, logger="pyhap.camera_multi_tier"), - ): - assert await _streaming_control(camera) == 0 - while process.stderr._data: - await asyncio.sleep(0) - await asyncio.sleep(0) - - assert "frame=4999 fps=30" in caplog.text - assert "last" in caplog.text - process.exit(1) - for _ in range(100): - if not _controller(camera).sessions: - break - await asyncio.sleep(0) - assert _controller(camera).sessions == {} - await camera.stop() - - -async def test_ffmpeg_announces_the_rtcp_monitor( - stream_camera: Callable[..., Camera], -) -> None: - """Test the answer announces the port the RTCP monitor bound.""" - camera = stream_camera( - multi_tier={"delegate": FFmpegMultiTierDelegate(FFMPEG_INPUT)} - ) - - answer = await _setup(camera) - - monitor = _session(camera).data["rtcp_monitor"] - assert tlv.decode(answer[SETUP_TYPES["ADDRESS"]])[ - SETUP_ADDR_INFO["VIDEO_RTP_PORT"] - ] == struct.pack(" None: - """Test a running ffmpeg does not keep a session alive without RTCP.""" - camera, session, _ = await _ffmpeg_session(stream_camera, idle_timeout=0.05) - monitor = session.data["rtcp_monitor"] - with patch("asyncio.create_subprocess_exec", return_value=ffmpeg_process): - assert await _streaming_control(camera) == 0 - await asyncio.sleep(0.2) - - assert session.ended - assert _controller(camera).sessions == {} - ffmpeg_process.terminate.assert_called_once() - assert monitor.closed - await camera.stop() - - -async def test_ffmpeg_rtcp_keeps_the_session_alive( - stream_camera: Callable[..., Camera], ffmpeg_process: Any -) -> None: - """Test the controller's RTCP keeps a running ffmpeg session alive.""" - camera, session, _ = await _ffmpeg_session( - stream_camera, _setup_request(b"127.0.0.1") - ) - with patch("asyncio.create_subprocess_exec", return_value=ffmpeg_process): - assert await _streaming_control(camera) == 0 - timer = session._timer - - await _send_rtcp(session.data["rtcp_monitor"], "127.0.0.1") - - assert timer is not None - assert timer.cancelled() - assert session.state is SessionState.STARTED - await camera.stop() - - -async def test_ffmpeg_start_fails(stream_camera: Callable[..., Camera]) -> None: - """Test a missing ffmpeg answers an error.""" - camera, _, _ = await _ffmpeg_session(stream_camera) - with patch("asyncio.create_subprocess_exec", side_effect=FileNotFoundError): - assert await _streaming_control(camera) == 4 - await camera.stop() - - -@pytest.mark.parametrize( - ("text", "expected"), - [ - ("rtsps://cam:7441/token?enableSrtp", "rtsps://cam:7441/"), - ( - "from 'rtsp://u:p@[fd00::1]:554/s':", - "from 'rtsp://[fd00::1]:554/':", - ), - ("rtsp://u:p@cam", "rtsp://cam"), - ("rtsp://admin:abc/def@cam/s", "rtsp://cam/"), - ("rtsp://admin:p#w@cam/s", "rtsp://cam/"), - ("from 'rtsp://admin:pa'ss@cam/s'", "from 'rtsp://cam/'"), - ("srtp://192.168.1.114:50000", "srtp://192.168.1.114:50000"), - ("frame=1 fps=30", "frame=1 fps=30"), - ], -) -def test_redact_urls(text: str, expected: str) -> None: - """Test URLs keep only their scheme, host and port.""" - assert _redact_urls(text) == expected - - -@pytest.mark.parametrize( - ("cmd", "expected"), - [ - ( - ["ffmpeg", "-i", "rtsp://admin:my pass@cam/s"], - "ffmpeg -i rtsp://cam/", - ), - ( - ["ffmpeg", "-i", "rtsp://admin:my\npass@cam/s"], - "ffmpeg -i rtsp://cam/", - ), - ( - ["ffmpeg", "-i", "rtsps://u:p@w@cam:7441/s?token"], - "ffmpeg -i rtsps://cam:7441/", - ), - ( - ["ffmpeg", "-srtp_out_params", "secret", "srtp://192.168.1.114:50000"], - "ffmpeg -srtp_out_params srtp://192.168.1.114:50000", - ), - ( - ["ffmpeg", "-f", "tee", "[f=rtsp]rtsp://u:p@cam/s"], - "ffmpeg -f tee [f=rtsp]rtsp://cam/", - ), - ], -) -def test_redact_command(cmd: list[str], expected: str) -> None: - """Test each URL argument of a command is redacted as a whole.""" - assert _redact_command(cmd) == expected + assert stream_driver._local_address(CLIENT, ipv6) == expected + assert stream_driver._local_address(OTHER, ipv6) is None diff --git a/tests/test_camera_stream.py b/tests/test_camera_stream.py index 9fe9bb31..14ca3257 100644 --- a/tests/test_camera_stream.py +++ b/tests/test_camera_stream.py @@ -2,7 +2,9 @@ import asyncio from collections.abc import Callable +import contextlib from dataclasses import dataclass +import gc import logging from typing import Any from unittest.mock import AsyncMock @@ -13,14 +15,14 @@ from pyhap import hksv3 from pyhap.camera_stream import ( MAX_HOOK_TIMEOUT, + HookFailed, SessionState, StreamController, StreamError, StreamSession, - _check_hook_timeout, - _config_from_option, - _failure_status, - _HookFailed, + check_hook_timeout, + config_from_option, + failure_status, ) CLIENT = ("192.168.1.114", 50000) @@ -37,7 +39,7 @@ async def _hook(result: Any = 1) -> Any: return result -async def _never() -> None: +async def _never(*_args: object) -> None: await asyncio.Event().wait() @@ -80,7 +82,7 @@ async def test_hook_timeout_ends_the_session( session = add_session(CLIENT) deadline = session_controller._deadline() - with pytest.raises(_HookFailed) as failed: + with pytest.raises(HookFailed) as failed: await session_controller._run_hook( session, "test", _never(), deadline, first=True ) @@ -110,7 +112,7 @@ async def test_hook_errors( """Test a delegate's status is kept, other errors have none.""" session = add_session(CLIENT) - with pytest.raises(_HookFailed) as failed: + with pytest.raises(HookFailed) as failed: await session_controller._run_hook( session, "test", _raise(error), session_controller._deadline() ) @@ -127,7 +129,7 @@ async def test_failed_hook_may_end_the_session( """Test a failed first hook ends the session and still releases it.""" session = add_session(CLIENT) - with pytest.raises(_HookFailed): + with pytest.raises(HookFailed): await session_controller._run_hook( session, "test", @@ -171,7 +173,7 @@ async def test_end_without_delegate( session = add_session(CLIENT) assert session_controller._end(session, "test") is None - assert await session_controller.end_session(session.id) is False + assert session_controller._end(session, "again") is None session_release.assert_not_called() @@ -190,29 +192,44 @@ async def test_end_waits_for_cancelled_hooks( ) await asyncio.sleep(0) - assert await session_controller.end_session(session.id) is True + release = session_controller._end(session, "test") + assert release is not None + assert await release is True session_release.assert_awaited_once_with(session) - with pytest.raises(_HookFailed): + with pytest.raises(HookFailed): await hook +@pytest.mark.parametrize( + ("error", "message"), + [ + (RuntimeError, "Releasing the session failed"), + (asyncio.CancelledError, "Releasing the session was cancelled"), + ], +) async def test_release_failure_is_logged( session_controller: StreamController[StreamSession], session_release: AsyncMock, caplog: pytest.LogCaptureFixture, add_session: Callable[..., StreamSession], + error: type[BaseException], + message: str, ) -> None: - """Test a failing release still ends the session.""" - session_release.side_effect = RuntimeError + """Test a failing release still finishes it and frees the session's room.""" + session_release.side_effect = error + add_session(CLIENT) + add_session(CLIENT) session = add_session(CLIENT) session._delegated = True release_task = session_controller._end(session, "test") assert release_task is not None - assert await release_task is False - assert "Releasing the session failed" in caplog.text + assert await release_task is True + assert message in caplog.text + assert "after its timeout" not in caplog.text + assert session_controller._no_room(CLIENT) is None async def test_cancelled_request_ends_the_session( @@ -244,7 +261,7 @@ async def test_hook_of_ended_session( session_controller._end(session, "test") hook = _hook() - with pytest.raises(_HookFailed): + with pytest.raises(HookFailed): await session_controller._run_hook( session, "test", hook, session_controller._deadline() ) @@ -328,22 +345,6 @@ async def test_client_disconnected( session_release.assert_awaited_once_with(mine) -async def test_end_all_sessions( - session_controller: StreamController[StreamSession], - session_release: AsyncMock, - add_session: Callable[..., StreamSession], -) -> None: - """Test ending all sessions waits for each release.""" - first = add_session(CLIENT) - second = add_session(OTHER) - first._delegated = second._delegated = True - - await session_controller.end_all_sessions() - - assert session_controller.sessions == {} - assert session_release.await_count == 2 - - async def test_end_by_request_waits_until_the_deadline( session_controller: StreamController[StreamSession], session_release: AsyncMock, @@ -353,13 +354,10 @@ async def test_end_by_request_waits_until_the_deadline( session_release.side_effect = _never session = add_session(CLIENT) session._delegated = True + deadline = asyncio.get_running_loop().time() + 0.01 - assert ( - await session_controller._end_by_request( - session, session_controller._deadline() - ) - is False - ) + assert await session_controller._end_by_request(session, deadline) is False + session_release.assert_awaited_once_with(session) assert session.state is SessionState.ENDED @@ -367,19 +365,19 @@ async def test_end_by_request_waits_until_the_deadline( def test_hook_timeout_leaves_time_to_answer(hook_timeout: float) -> None: """Test hooks must end before the HAP response timeout.""" with pytest.raises(ValueError, match="hook_timeout"): - _check_hook_timeout(hook_timeout) + check_hook_timeout(hook_timeout) def test_config_from_option() -> None: """Test options build a config, a config is taken as it is.""" config = _Config(max_sessions=6) - assert _config_from_option(_Config, config, "multi_tier") is config - assert _config_from_option(_Config, {"max_sessions": 6}, "x") == config + assert config_from_option(_Config, config, "multi_tier") is config + assert config_from_option(_Config, {"max_sessions": 6}, "x") == config with pytest.raises(ValueError, match="Unknown multi_tier option: start"): - _config_from_option(_Config, {"start": 1}, "multi_tier") + config_from_option(_Config, {"start": 1}, "multi_tier") with pytest.raises(ValueError, match="must be a dict"): - _config_from_option(_Config, [], "multi_tier") + config_from_option(_Config, [], "multi_tier") @pytest.mark.parametrize( @@ -389,7 +387,7 @@ def test_config_from_option() -> None: def test_failure_status(status: int | None, expected: int) -> None: """Test a delegate picks a known failure status, else the error.""" error = hksv3.RTPStreamingStatus.ERROR - assert _failure_status(hksv3.RTPStreamingStatus, status, error) == expected + assert failure_status(hksv3.RTPStreamingStatus, status, error) == expected async def test_session_end( @@ -449,7 +447,7 @@ async def test_release_after_a_stuck_hook( assert not finished.is_set() assert "while 1 cancelled hooks still run" in caplog.text session_release.assert_awaited_once_with(session) - with pytest.raises(_HookFailed): + with pytest.raises(HookFailed): await request await finished.wait() @@ -481,3 +479,190 @@ async def test_no_room_once_closed( await session_controller.close() assert session_controller._no_room(CLIENT) == "The stream service is closed" + + +async def test_releasing_session_keeps_its_room( + session_controller: StreamController[StreamSession], + session_release: AsyncMock, + add_session: Callable[..., StreamSession], +) -> None: + """Test an ended session counts towards the limits until it is released.""" + released = asyncio.Event() + + async def _release(_session: StreamSession) -> None: + await released.wait() + + session_release.side_effect = _release + add_session(CLIENT) + add_session(CLIENT) + session = add_session(CLIENT) + session._delegated = True + + release = session_controller._end(session, "test") + + assert session_controller._no_room(CLIENT) == ( + "Too many stream sessions of the connection" + ) + released.set() + assert release is not None + assert await release is True + assert session_controller._no_room(CLIENT) is None + + +async def test_cancelled_hook_ends_the_session( + session_controller: StreamController[StreamSession], + session_release: AsyncMock, + add_session: Callable[..., StreamSession], + caplog: pytest.LogCaptureFixture, +) -> None: + """Test a hook that cancels itself fails and may end its session.""" + session = add_session(CLIENT) + + with pytest.raises(HookFailed): + await session_controller._run_hook( + session, + "test", + _raise(asyncio.CancelledError()), + session_controller._deadline(), + end_on_failure=True, + first=True, + ) + + assert session.state is SessionState.ENDED + assert "The test hook was cancelled" in caplog.text + await session_controller.close() + session_release.assert_awaited_once_with(session) + + +async def _fails_when_cancelled(started: asyncio.Event) -> None: + """A hook that raises another error when it is cancelled.""" + started.set() + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + raise RuntimeError("cleanup failed") from None + + +async def test_hook_error_after_the_end_is_retrieved( + session_controller: StreamController[StreamSession], + add_session: Callable[..., StreamSession], +) -> None: + """Test the error of a hook whose session ended meanwhile is not lost.""" + loop = asyncio.get_running_loop() + unhandled: list[dict[str, Any]] = [] + loop.set_exception_handler(lambda _loop, context: unhandled.append(context)) + session = add_session(CLIENT) + started = asyncio.Event() + request = asyncio.ensure_future( + session_controller._run_hook( + session, "test", _fails_when_cancelled(started), loop.time() + 1 + ) + ) + await started.wait() + + session_controller._end(session, "test") + await asyncio.wait({request}) + assert isinstance(request.exception(), HookFailed) + # Drop the failure's traceback, it holds the hook's task. + del request + gc.collect() + loop.set_exception_handler(None) + + assert unhandled == [] + + +def _catch_unhandled(loop: asyncio.AbstractEventLoop) -> list[dict[str, Any]]: + """Collect what reaches the loop's exception handler.""" + unhandled: list[dict[str, Any]] = [] + loop.set_exception_handler(lambda _loop, context: unhandled.append(context)) + return unhandled + + +async def _fails_after_cancellation(_session: StreamSession) -> None: + """A release that catches its cancellation and then raises.""" + try: + await asyncio.Event().wait() + except asyncio.CancelledError: + pass + raise RuntimeError("late release failure") + + +async def test_timed_out_release_error_is_retrieved( + session_controller: StreamController[StreamSession], + session_release: AsyncMock, + add_session: Callable[..., StreamSession], + caplog: pytest.LogCaptureFixture, +) -> None: + """Test the error of a release raised after its timeout is warned about.""" + loop = asyncio.get_running_loop() + unhandled = _catch_unhandled(loop) + session_release.side_effect = _fails_after_cancellation + session = add_session(CLIENT) + session._delegated = True + + release = session_controller._end(session, "test") + assert release is not None + assert await release is False + await asyncio.sleep(0) + del release + gc.collect() + loop.set_exception_handler(None) + + assert unhandled == [] + assert ( + "pyhap.camera_stream", + logging.WARNING, + f"[{session.id}] The end hook failed after its timeout: " + "RuntimeError('late release failure')", + ) in caplog.record_tuples + assert session_controller._abandoned == set() + + +async def _ignores_cancellation(stop: asyncio.Event) -> None: + with contextlib.suppress(asyncio.CancelledError): + await stop.wait() + await stop.wait() + + +async def _ignores_cancellation_then_fails(stop: asyncio.Event) -> None: + await _ignores_cancellation(stop) + raise RuntimeError("late release failure") + + +@pytest.mark.parametrize( + "release", [_ignores_cancellation, _ignores_cancellation_then_fails] +) +async def test_close_gives_up_on_a_stubborn_release( + session_controller: StreamController[StreamSession], + session_release: AsyncMock, + add_session: Callable[..., StreamSession], + caplog: pytest.LogCaptureFixture, + release: Callable[[asyncio.Event], Any], +) -> None: + """Test close() returns although a release ignores its cancellation. + + Its error, if it fails once it ends, is retrieved. + """ + loop = asyncio.get_running_loop() + unhandled = _catch_unhandled(loop) + stop = asyncio.Event() + + async def _release(_session: StreamSession) -> None: + await release(stop) + + session_release.side_effect = _release + session = add_session(CLIENT) + session._delegated = True + + await asyncio.wait_for(session_controller.close(), 1) + stop.set() + await asyncio.sleep(0) + await asyncio.sleep(0) + gc.collect() + loop.set_exception_handler(None) + + assert unhandled == [] + assert "Releasing the session timed out" in caplog.text + assert "Closing the stream service while 1 tasks still run" in caplog.text + assert session_controller._no_room(OTHER) == "The stream service is closed" + assert session_controller._releasing == set() diff --git a/tests/test_camera_webrtc.py b/tests/test_camera_webrtc.py index 90fb3f49..d326b231 100644 --- a/tests/test_camera_webrtc.py +++ b/tests/test_camera_webrtc.py @@ -11,7 +11,7 @@ from pyhap import hksv3 from pyhap.camera import Camera from pyhap.camera_stream import SessionState, StreamError -from pyhap.camera_webrtc import WebRTCController, WebRTCSession, remote_tier +from pyhap.camera_webrtc import WebRTCController, WebRTCSession, _remote_tier from pyhap.util import base64_to_bytes, to_base64_str WEBRTC = "CameraWebRTCStreamManagement" @@ -102,8 +102,8 @@ async def _hang(*_args: object) -> None: def test_remote_tier(stream_sensor: hksv3.SensorConfiguration) -> None: """Test the remote tier is the smallest of a quality, else the smallest.""" tiers = hksv3.tiers_from_sensor(stream_sensor) - assert remote_tier(tiers, hksv3.VideoQuality.MEDIUM) == MEDIUM - assert remote_tier(tiers, hksv3.VideoQuality.HIGHEST) == LOW + assert _remote_tier(tiers, hksv3.VideoQuality.MEDIUM) == MEDIUM + assert _remote_tier(tiers, hksv3.VideoQuality.HIGHEST) == LOW async def test_solicit_offer( @@ -197,6 +197,44 @@ async def test_solicit_offer_refused( webrtc_delegate.handle_solicit_offer.assert_not_called() +async def test_solicit_offer_after_stop( + stream_camera: Callable[..., Camera], webrtc_delegate: AsyncMock +) -> None: + """Test a stopped camera answers an offer with an error.""" + camera = stream_camera() + await camera.stop() + + response = await _solicit(camera) + + assert response.status == hksv3.WebRTCSolicitOfferStatus.ERROR + assert response.session_id is None + webrtc_delegate.handle_solicit_offer.assert_not_called() + + +@pytest.mark.parametrize( + "offer", [None, "v=0\r\n", hksv3.WebRTCReofferAnswer("v=0\r\n")] +) +async def test_offer_of_another_type( + stream_camera: Callable[..., Camera], + webrtc_delegate: AsyncMock, + caplog: pytest.LogCaptureFixture, + offer: Any, +) -> None: + """Test an offer hook that returns no WebRTCOffer fails, the session ends.""" + webrtc_delegate.handle_solicit_offer.return_value = offer + camera = stream_camera() + + response = await _solicit(camera) + sessions = dict(_controller(camera).sessions) + await camera.stop() + + assert response.status == hksv3.WebRTCSolicitOfferStatus.ERROR + assert response.session_id is None + assert sessions == {} + assert "The offer is not a WebRTCOffer" in caplog.text + webrtc_delegate.handle_end.assert_awaited_once() + + async def test_solicit_offer_without_delegate( stream_camera: Callable[..., Camera], ) -> None: @@ -210,6 +248,57 @@ async def test_solicit_offer_limit(stream_camera: Callable[..., Camera]) -> None assert await _solicit_many(camera, 7) == [0, 0, 0, 0, 0, 0, 2] +@pytest.fixture +def end_gate(webrtc_delegate: AsyncMock) -> asyncio.Event: + """Make ``handle_end`` wait for the returned event.""" + gate = asyncio.Event() + + async def _end(_session: WebRTCSession) -> None: + await gate.wait() + + webrtc_delegate.handle_end.side_effect = _end + return gate + + +async def _solicit_ids(camera: Camera, count: int) -> list[UUID]: + """Solicit ``count`` offers, return the session IDs.""" + ids: list[UUID] = [] + for _ in range(count): + response = await _solicit(camera) + assert response.session_id is not None + ids.append(response.session_id) + return ids + + +async def _end_all(camera: Camera, ids: list[UUID]) -> list[int]: + """Write End for each session at once, return the statuses.""" + answers = await asyncio.gather(*(_end(camera, session_id) for session_id in ids)) + return [answer.status for answer in answers] + + +async def test_releasing_sessions_count_towards_max_sessions( + stream_camera: Callable[..., Camera], end_gate: asyncio.Event +) -> None: + """Test ended sessions being released count towards max_sessions.""" + camera = stream_camera() + ends = asyncio.ensure_future(_end_all(camera, await _solicit_ids(camera, 6))) + # One step starts the gather, the next the End writes. + await asyncio.sleep(0) + await asyncio.sleep(0) + sessions = dict(_controller(camera).sessions) + + refused = await _solicit(camera, client=OTHER) + end_gate.set() + statuses = await ends + accepted = await _solicit(camera, client=OTHER) + await camera.stop() + + assert sessions == {} + assert refused.status == hksv3.WebRTCSolicitOfferStatus.ERROR + assert statuses == [0, 0, 0, 0, 0, 0] + assert accepted.status == hksv3.WebRTCSolicitOfferStatus.SUCCESS + + async def test_solicit_offer_limit_per_connection( stream_camera: Callable[..., Camera], webrtc_delegate: AsyncMock ) -> None: @@ -399,12 +488,32 @@ async def test_end( async def test_end_fails( stream_camera: Callable[..., Camera], webrtc_delegate: AsyncMock ) -> None: - """Test a failed release answers an error, the session is gone anyway.""" + """Test a failed but finished release answers success and frees the room.""" webrtc_delegate.handle_end.side_effect = RuntimeError - camera = stream_camera() + camera = stream_camera( + webrtc={"delegate": webrtc_delegate, "max_sessions_per_connection": 1} + ) session = await _active(camera) - assert (await _end(camera, session.id)).status == 3 + + assert (await _end(camera, session.id)).status == 0 assert _controller(camera).sessions == {} + assert (await _solicit(camera)).status == 0 + + +async def test_end_release_too_slow( + stream_camera: Callable[..., Camera], + webrtc_delegate: AsyncMock, + end_gate: asyncio.Event, +) -> None: + """Test an end answers an error when the release overruns.""" + camera = stream_camera(webrtc={"delegate": webrtc_delegate, "hook_timeout": 0.05}) + session = await _active(camera) + + status = (await _end(camera, session.id)).status + end_gate.set() + await camera.stop() + + assert status == 3 async def test_unknown_command(stream_camera: Callable[..., Camera]) -> None: @@ -440,19 +549,23 @@ async def test_sessions_end( webrtc_delegate.handle_end.assert_awaited_once_with(session) -async def test_camera_ends_a_session( +async def test_delegate_ends_a_session( stream_camera: Callable[..., Camera], webrtc_delegate: AsyncMock ) -> None: - """Test the integrator ends a session and all of them.""" + """Test the delegate ends sessions, each is released once.""" camera = stream_camera() first = await _active(camera) second = await _session(camera) - assert await camera.end_session(first.id) is True - assert await camera.end_session(first.id) is False - await camera.end_all_sessions() + first.end() + first.end() + second.end() + await asyncio.sleep(0) + sessions = dict(_controller(camera).sessions) + await camera.stop() - assert second.state is SessionState.ENDED + assert sessions == {} + assert _count(camera) == 0 assert webrtc_delegate.handle_end.await_count == 2 @@ -554,6 +667,23 @@ async def test_reoffer_answer_with_unexpected_sframe_key( assert session.state is SessionState.ACTIVE +async def test_reoffer_answer_of_wrong_type( + stream_camera: Callable[..., Camera], + webrtc_delegate: AsyncMock, + caplog: pytest.LogCaptureFixture, +) -> None: + """Test a reoffer hook returning no WebRTCReofferAnswer answers an error.""" + camera = stream_camera() + session = await _active(camera) + webrtc_delegate.handle_reoffer.return_value = None + + response = await _reoffer(camera, session.id) + + assert response == hksv3.WebRTCReofferResponse(session.id, 3) + assert session.state is SessionState.ACTIVE + assert "The reoffer answer is not a WebRTCReofferAnswer: None" in caplog.text + + async def test_update_session( stream_camera: Callable[..., Camera], webrtc_delegate: AsyncMock ) -> None: diff --git a/tests/test_camera_webrtc_hap.py b/tests/test_camera_webrtc_hap.py index 209f5b60..ded26340 100644 --- a/tests/test_camera_webrtc_hap.py +++ b/tests/test_camera_webrtc_hap.py @@ -1,10 +1,11 @@ -"""Tests for WebRTC writes of pyhap.camera through the accessory driver.""" +"""Tests for HKSV 17.99 stream writes of pyhap.camera through the driver.""" import asyncio from collections.abc import Callable import inspect from typing import Any from unittest.mock import AsyncMock +from uuid import UUID from pyhap import hksv3 from pyhap.accessory import Bridge @@ -13,6 +14,8 @@ from pyhap.util import base64_to_bytes, to_base64_str WEBRTC = "CameraWebRTCStreamManagement" +MULTI_TIER = "CameraMultiTierRTPStreamManagement" +SESSION = UUID("acccc6cc-1505-6345-158d-a974b4509006") CLIENT = ("192.168.1.114", 50000) OTHER = ("192.168.1.115", 51000) SOLICIT = to_base64_str( @@ -27,8 +30,8 @@ def _bridged(driver: AccessoryDriver, camera: Camera) -> Camera: return camera -def _char_id(camera: Camera, name: str) -> tuple[int, int]: - service = camera.get_service(WEBRTC) +def _char_id(camera: Camera, name: str, service_name: str = WEBRTC) -> tuple[int, int]: + service = camera.get_service(service_name) assert service is not None iid = camera.iid_manager.get_iid(service.get_characteristic(name)) assert camera.aid is not None @@ -42,21 +45,31 @@ async def _write( name: str, value: str, timeout: float = 9.0, + response: bool = True, + service_name: str = WEBRTC, ) -> dict[str, Any]: - """Write like a controller that asks for a write response.""" - aid, iid = _char_id(camera, name) - query = {"characteristics": [{"aid": aid, "iid": iid, "value": value, "r": True}]} + """Write like a controller, by default one that asks for a write response. + + Returns the result of the characteristic, empty if the answer has no body. + """ + aid, iid = _char_id(camera, name, service_name) + char = {"aid": aid, "iid": iid, "value": value, "r": response} + query = {"characteristics": [char]} result = driver._set_characteristics_for_controller(query, CLIENT, timeout) assert inspect.isawaitable(result) - response = await result - assert response is not None - return dict(response["characteristics"][0]) + body = await result + # Without r a write that succeeds has no body. + return {} if body is None else dict(body["characteristics"][0]) def _read( - driver: AccessoryDriver, camera: Camera, name: str, client: tuple[str, int] + driver: AccessoryDriver, + camera: Camera, + name: str, + client: tuple[str, int], + service_name: str = WEBRTC, ) -> Any: - aid, iid = _char_id(camera, name) + aid, iid = _char_id(camera, name, service_name) response = driver.get_characteristics([f"{aid}.{iid}"], client) return response["characteristics"][0]["value"] @@ -90,6 +103,57 @@ async def test_responses_read_back_per_connection( await camera.stop() +async def test_answer_without_write_response_stays_private( + stream_driver: AccessoryDriver, stream_camera: Callable[..., Camera] +) -> None: + """Test an answer written without r is not shared, its writer reads the status.""" + camera = _bridged(stream_driver, stream_camera()) + result = await _write(stream_driver, camera, "WebRTCSolicitOffer", SOLICIT) + offer = hksv3.WebRTCSolicitOfferResponse.decode(base64_to_bytes(result["value"])) + assert offer.session_id is not None + answer = hksv3.WebRTCProvideAnswer(offer.session_id, "v=0\r\na=ice-pwd:secret\r\n") + + result = await _write( + stream_driver, + camera, + "WebRTCProvideAnswer", + to_base64_str(answer.encode()), + response=False, + ) + + assert result == {} + assert _read(stream_driver, camera, "WebRTCProvideAnswer", OTHER) == "" + assert hksv3.WebRTCSessionStatus.decode( + base64_to_bytes(_read(stream_driver, camera, "WebRTCProvideAnswer", CLIENT)) + ) == hksv3.WebRTCSessionStatus(offer.session_id, hksv3.WebRTCStreamingStatus(0)) + await camera.stop() + + +async def test_streaming_control_without_write_response_stays_private( + stream_driver: AccessoryDriver, stream_camera: Callable[..., Camera] +) -> None: + """Test an RTP Streaming Control write without r is not shared.""" + camera = _bridged(stream_driver, stream_camera()) + request = hksv3.encode_item(1, SESSION.bytes) + hksv3.encode_item(2, b"\x02") + + result = await _write( + stream_driver, + camera, + "RTPStreamingControl", + to_base64_str(request), + response=False, + service_name=MULTI_TIER, + ) + + assert result == {} + assert _read(stream_driver, camera, "RTPStreamingControl", OTHER, MULTI_TIER) == "" + status = _read(stream_driver, camera, "RTPStreamingControl", CLIENT, MULTI_TIER) + assert base64_to_bytes(status) == hksv3.encode_streaming_status( + SESSION, hksv3.RTPStreamingStatus.UNKNOWN_SESSION + ) + await camera.stop() + + async def test_slow_hook_answers_error_not_timeout( stream_driver: AccessoryDriver, stream_camera: Callable[..., Camera], diff --git a/tests/test_hksv3.py b/tests/test_hksv3.py index 83e3a4f7..1a0eb8fa 100644 --- a/tests/test_hksv3.py +++ b/tests/test_hksv3.py @@ -480,6 +480,15 @@ def test_rtp_streaming_control_end() -> None: assert request == RTPStreamingControl(SESSION, RTPStreamingCommand.END, 3) +def test_rtp_streaming_control_largest_ssrc() -> None: + """Test the largest 32-bit SSRC still decodes for video and audio.""" + request = RTPStreamingControl.decode( + bytes.fromhex("0110" + SESSION.hex + "020102 0404ffffffff 0604ffffffff") + ) + assert request.video_ssrc == 0xFFFFFFFF + assert request.audio_ssrc == 0xFFFFFFFF + + def test_rtp_streaming_control_unknown_values() -> None: """Test unknown commands and types are kept and skipped.""" request = RTPStreamingControl.decode( @@ -497,10 +506,12 @@ def test_rtp_streaming_control_unknown_values() -> None: "0104aabbccdd 020102", "0110" + SESSION.hex + "020102 0300", "0110" + SESSION.hex + "0201", + "0110" + SESSION.hex + "020102 04050000000001", + "0110" + SESSION.hex + "020102 06050000000001", ], ) def test_rtp_streaming_control_invalid(data: str) -> None: - """Test missing fields, bad lengths and truncated data raise.""" + """Test missing fields, bad lengths, truncated data and wide SSRCs raise.""" with pytest.raises(ValueError): RTPStreamingControl.decode(bytes.fromhex(data)) @@ -580,6 +591,61 @@ def test_check_video_tiers_guide_table() -> None: assert hksv3.check_video_tiers(_sensor(GUIDE_4K, GUIDE_1080, GUIDE_360)) == [] +@pytest.mark.parametrize( + "streams", + [ + ((2, 2880, 2160, 4500, 5000), (3, 1600, 1200, 1700, 1800), GUIDE_360), + ((2, 2400, 3200, 4500, 5000), (3, 1200, 1600, 1700, 1800), GUIDE_360), + ((2, 2880, 2880, 4500, 5000), (3, 1440, 1440, 1700, 1800), GUIDE_360), + ((2, 2560, 1440, 2800, 3000), GUIDE_1080, (4, 640, 480, 180, 190)), + ((2, 2048, 1536, 2800, 3000), (3, 1600, 1200, 1700, 1800), GUIDE_360), + ((2, 1536, 2048, 2800, 3000), (3, 1200, 1600, 1700, 1800), GUIDE_360), + ((2, 1920, 1920, 2800, 3000), (3, 1440, 1440, 1700, 1800), GUIDE_360), + (GUIDE_1080, (3, 1280, 720, 768, 800), (4, 426, 240, 180, 190)), + (GUIDE_1080, (3, 1440, 1080, 768, 800), (4, 640, 480, 180, 190)), + ( + (2, 1200, 1600, 1700, 1800), + (3, 960, 1280, 768, 800), + (4, 480, 640, 180, 190), + ), + ( + (2, 1440, 1440, 1700, 1800), + (3, 1080, 1080, 768, 800), + (4, 480, 480, 180, 190), + ), + ], +) +def test_check_video_tiers_guide_resolutions( + streams: tuple[tuple[int, int, int, int, int], ...], +) -> None: + """Test every resolution of the guide at its class's bit rates gives no warning.""" + assert hksv3.check_video_tiers(_sensor(*streams)) == [] + + +def test_check_video_tiers_above_4k() -> None: + """Test a frame above 4K, here DCI 4K, is held to the guide's 4K row.""" + dci_4k = (2, 4096, 2160, 4500, 5000) + assert hksv3.check_video_tiers(_sensor(dci_4k, GUIDE_1080, GUIDE_360)) == [] + + +@pytest.mark.parametrize( + ("stream", "warning"), + [ + ((3, 2048, 1536, 2800, 3100), "peak above the guide's 3000 kbps"), + ((3, 1920, 1920, 2800, 3100), "peak above the guide's 3000 kbps"), + ((3, 1440, 1440, 1750, 1900), "peak above the guide's 1800 kbps"), + ((3, 1440, 1080, 768, 850), "peak above the guide's 800 kbps"), + ((3, 640, 480, 180, 200), "peak above the guide's 190 kbps"), + ], +) +def test_check_video_tiers_class_by_pixels( + stream: tuple[int, int, int, int, int], warning: str +) -> None: + """Test a frame takes the row of its class in the guide, not of its short side.""" + (found,) = hksv3.check_video_tiers(_sensor(GUIDE_4K, stream, GUIDE_360)) + assert warning in found + + @pytest.mark.parametrize( ("stream", "warning"), [ @@ -587,7 +653,7 @@ def test_check_video_tiers_guide_table() -> None: ((3, 1920, 1080, 1800, 1900), "peak above the guide's 1800 kbps"), ((3, 1080, 1920, 1800, 1900), "1080x1920@30: peak above the guide's 1800"), ((3, 1280, 600, 900, 950), "1280x600@30: peak above the guide's 800 kbps"), - ((3, 7680, 4320, 9000, 9500), "larger than the guide's 4K bit rate table"), + ((3, 7680, 4320, 9000, 9500), "7680x4320@30: peak above the guide's 5000"), ], ) def test_check_video_tiers_warnings(