From ee24ebd50ff2b1c9f99d04875e62ee0a622c7d19 Mon Sep 17 00:00:00 2001 From: AP2 Contributor Date: Tue, 6 Oct 2026 16:54:02 +0200 Subject: [PATCH] airplay: Add opt-in controlType-1 playback relay --- docs/api/pyatv.settings.html | 25 +- docs/development/control.md | 76 +++- examples/airplay_relay.py | 52 +++ pyatv/protocols/airplay/__init__.py | 3 + pyatv/protocols/airplay/relay.py | 495 +++++++++++++++++++++++++ pyatv/protocols/mrp/__init__.py | 22 +- pyatv/protocols/mrp/protocol.py | 7 +- pyatv/settings.py | 3 + tests/protocols/airplay/test_relay.py | 504 ++++++++++++++++++++++++++ 9 files changed, 1170 insertions(+), 17 deletions(-) create mode 100644 examples/airplay_relay.py create mode 100644 pyatv/protocols/airplay/relay.py create mode 100644 tests/protocols/airplay/test_relay.py diff --git a/docs/api/pyatv.settings.html b/docs/api/pyatv.settings.html index c4a987788..e5b33c1d7 100644 --- a/docs/api/pyatv.settings.html +++ b/docs/api/pyatv.settings.html @@ -78,6 +78,7 @@

M
  • Auto
  • Disable
  • Force
    • +
  • Relay
  • @@ -118,7 +119,7 @@

    Module pyatv.settings

    Settings for configuring pyatv.

    -
    Browse git
    +
    Browse git
    @@ -139,7 +140,7 @@

    Classes

    Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

    self is explicitly positional-only to allow self as a field name.

    -
    Browse git
    +
    Browse git

    Ancestors

    • pydantic.main.BaseModel
      • @@ -202,7 +203,7 @@

      Class variables

      Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

      self is explicitly positional-only to allow self as a field name.

      -
      Browse git
      +
      Browse git

      Ancestors

      • pydantic.main.BaseModel
        • @@ -229,7 +230,7 @@

        Class variables

        Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

        self is explicitly positional-only to allow self as a field name.

        -
        Browse git
        +
        Browse git

        Ancestors

        • pydantic.main.BaseModel
          • @@ -256,7 +257,7 @@

          Class variables

          Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

          self is explicitly positional-only to allow self as a field name.

          -
          Browse git
          +
          Browse git

          Ancestors

          • pydantic.main.BaseModel
            • @@ -326,7 +327,7 @@

            Static methods

            Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

            self is explicitly positional-only to allow self as a field name.

            -
            Browse git
            +
            Browse git

            Ancestors

            • pydantic.main.BaseModel
              • @@ -349,7 +350,7 @@

              Class variables

              How MRP tunneling over AirPlay is handled.

              -
              Browse git
              +
              Browse git

              Ancestors

              • builtins.str
                • @@ -369,6 +370,10 @@

                Class variables

                Force set up of MRP tunnel even if remote device does not supports it.

                +
                var Relay = relay
                +
                +

                Use the experimental controlType-1 relay for existing sender playback.

                +
              @@ -381,7 +386,7 @@

              Class variables

              Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

              self is explicitly positional-only to allow self as a field name.

              -
              Browse git
              +
              Browse git

              Ancestors

              • pydantic.main.BaseModel
                • @@ -420,7 +425,7 @@

                Class variables

                Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

                self is explicitly positional-only to allow self as a field name.

                -
                Browse git
                +
                Browse git

                Ancestors

                • pydantic.main.BaseModel
                  • @@ -467,7 +472,7 @@

                  Class variables

                  Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

                  self is explicitly positional-only to allow self as a field name.

                  -
                  Browse git
                  +
                  Browse git

                  Ancestors

                  • pydantic.main.BaseModel
                    • diff --git a/docs/development/control.md b/docs/development/control.md index 9e67d549c..83d3a81ea 100644 --- a/docs/development/control.md +++ b/docs/development/control.md @@ -6,6 +6,80 @@ link_group: development --- # Control +## Experimental AirPlay receiver relay + +AirPlay 2 receivers can relay an existing sender's playback information and +commands. Opt into this experimental mode with `MrpTunnel.Relay`; automatic +selection and the existing AirPlay tunnel are unchanged. Keep the audio sender +playing directly to the receiver. Connecting as a relay does not start audio. + +```python +import asyncio +import pyatv +from pyatv.const import Protocol +from pyatv.settings import MrpTunnel +from pyatv.storage.memory_storage import MemoryStorage + +async def main(): + loop = asyncio.get_running_loop() + storage = MemoryStorage() + devices = await pyatv.scan(loop, protocol=Protocol.AirPlay, storage=storage) + config = next(device for device in devices if device.name == "Receiver") + settings = await storage.get_settings(config) + settings.protocols.airplay.mrp_tunnel = MrpTunnel.Relay + atv = await pyatv.connect(config, loop, storage=storage) + try: + print(await atv.metadata.playing()) + artwork = await atv.metadata.artwork(width=512) + if artwork: + print(artwork.mimetype, len(artwork.bytes)) + await atv.remote_control.pause() + await atv.remote_control.play() + await atv.remote_control.next() + await atv.remote_control.previous() + await atv.remote_control.set_position(90) # seconds + await atv.audio.set_volume(50) # percent, selected receiver + finally: + await asyncio.gather(*atv.close()) + +asyncio.run(main()) +``` + +Each playback call above changes the active sender. Use only the calls you need. +The [relay example](https://github.com/postlund/pyatv/blob/master/examples/airplay_relay.py) +discovers a named receiver and performs one action (default: status): + +```shell +python -m examples.airplay_relay "Receiver" +python -m examples.airplay_relay "Receiver" pause +python -m examples.airplay_relay "Receiver" seek 90 +python -m examples.airplay_relay "Receiver" volume 50 +``` + +Normal metadata and push updater interfaces report the globally selected +client/player. `metadata.artwork()` reuses the existing artwork retrieval and +cache; it returns `None` when no artwork is available. Requested image dimensions +are a hint, and the sender may return a different size. Playback commands include +that player's complete path and check sender rejection responses. A group shares playback commands; volume targets the +selected receiver's reported output-device UID. To change all reported group +members explicitly, iterate over `atv.audio.output_devices` and pass each member +to `await atv.audio.set_volume(50, member)`. Volume calls wait for a matching +receiver update; a control socket acknowledgment alone does not confirm volume. + +This initial implementation supports transient AirPlay pairing and type-130, +controlType-1 streams without a dedicated dataPort. HomeKit authentication, +receiver compatibility beyond the reference, and modifying group membership +remain unverified. Available playback actions depend on the sender's capabilities. +TV navigation and power controls are not provided by this mode. + +The transport authenticates its own encrypted AirPlay session, requests an auth +ticket, negotiates the control-only and remote streams, and subscribes over +binary-plist `/command` requests. It uses the event channel for protobuf replies +and updates, sends periodic feedback, and tears down only its own control session. +Existing Media Remote message/state code is reused internally; no standalone MRP +service or audio data socket is opened. No packet interception or sender keys are +required. + Controlling a device is done with the remote control interface, {% include api i="interface.RemoteControl" %}. It allows you navigate the menus and change playback (play, pause, etc.). @@ -53,4 +127,4 @@ await rc.menu(action=InputAction.Hold) await rc.home(action=InputAction.DoubleTap) ``` -All input actions are specified in {% include api i="const.InputAction" %}. \ No newline at end of file +All input actions are specified in {% include api i="const.InputAction" %}. diff --git a/examples/airplay_relay.py b/examples/airplay_relay.py new file mode 100644 index 000000000..84e9c3467 --- /dev/null +++ b/examples/airplay_relay.py @@ -0,0 +1,52 @@ +"""Control existing AirPlay playback using pyatv's public interfaces.""" + +import argparse +import asyncio + +import pyatv +from pyatv.const import Protocol +from pyatv.settings import MrpTunnel +from pyatv.storage.memory_storage import MemoryStorage + + +async def main() -> None: + """Discover a receiver, opt into its relay, and perform one requested action.""" + parser = argparse.ArgumentParser(description=__doc__) + parser.add_argument("name", help="Exact name of the AirPlay receiver") + parser.add_argument( + "action", + nargs="?", + default="status", + choices=["status", "play", "pause", "next", "previous", "seek", "volume"], + ) + parser.add_argument( + "value", nargs="?", type=float, help="Seconds or volume (0–100)" + ) + args = parser.parse_args() + if args.action in ("seek", "volume") and args.value is None: + parser.error("seek and volume require a value") + + loop = asyncio.get_running_loop() + storage = MemoryStorage() + devices = await pyatv.scan(loop, protocol=Protocol.AirPlay, storage=storage) + config = next((device for device in devices if device.name == args.name), None) + if config is None: + print(f"Receiver not found: {args.name}") + return + settings = await storage.get_settings(config) + settings.protocols.airplay.mrp_tunnel = MrpTunnel.Relay + atv = await pyatv.connect(config, loop, storage=storage) + try: + print(await atv.metadata.playing()) + if args.action == "seek": + await atv.remote_control.set_position(args.value) + elif args.action == "volume": + await atv.audio.set_volume(args.value) + elif args.action != "status": + await getattr(atv.remote_control, args.action)() + finally: + await asyncio.gather(*atv.close()) + + +if __name__ == "__main__": + asyncio.run(main()) diff --git a/pyatv/protocols/airplay/__init__.py b/pyatv/protocols/airplay/__init__.py index 0bff2ea03..4649a16d0 100644 --- a/pyatv/protocols/airplay/__init__.py +++ b/pyatv/protocols/airplay/__init__.py @@ -26,6 +26,7 @@ Stream, ) from pyatv.protocols import mrp +from pyatv.protocols.airplay import relay from pyatv.protocols.airplay.ap2_session import AP2Session from pyatv.protocols.airplay.auth import extract_credentials from pyatv.protocols.airplay.mrp_connection import AirPlayMrpConnection @@ -375,6 +376,8 @@ def _device_info() -> Dict[str, Any]: if mrp_tunnel == MrpTunnel.Disable: _LOGGER.debug("Remote control tunnel disabled by setting") + elif mrp_tunnel == MrpTunnel.Relay: + yield relay.setup(core, credentials) elif mrp_tunnel == MrpTunnel.Force: _LOGGER.debug("Remote control channel is supported (forced)") yield _create_mrp_tunnel_data(core, credentials) diff --git a/pyatv/protocols/airplay/relay.py b/pyatv/protocols/airplay/relay.py new file mode 100644 index 000000000..06d681498 --- /dev/null +++ b/pyatv/protocols/airplay/relay.py @@ -0,0 +1,495 @@ +"""Experimental controlType-1 AirPlay relay for existing sender playback.""" + +import asyncio +from functools import partial +import hashlib +import logging +import math +import platform +from typing import Any, Dict, Optional, Set, Union, cast +from uuid import uuid4 + +from pyatv import exceptions +from pyatv.auth.hap_channel import setup_channel +from pyatv.auth.hap_pairing import AuthenticationType, HapCredentials +from pyatv.const import FeatureName, Protocol +from pyatv.core import Core, MutableService, SetupData +from pyatv.interface import Metadata, OutputDevice, Power +from pyatv.protocols.airplay.auth import verify_connection +from pyatv.protocols.airplay.channels import BaseDataStreamChannel, BaseEventChannel +from pyatv.protocols.mrp import ( + MrpAudio, + MrpRemoteControl, + create_with_connection, + messages, + protobuf, +) +from pyatv.protocols.mrp.connection import AbstractMrpConnection +from pyatv.support import async_timeout +from pyatv.support.http import ( + HttpConnection, + HttpResponse, + decode_bplist_from_body, + http_connect, +) +from pyatv.support.rtsp import RtspSession + +_LOGGER = logging.getLogger(__name__) + + +class RelayEvents(BaseEventChannel): + """Acknowledge receiver requests and enqueue their protobuf messages.""" + + def __init__(self, output_key: bytes, input_key: bytes, queue: asyncio.Queue): + """Initialize an encrypted event channel with a message queue.""" + super().__init__(output_key, input_key) + self.queue = queue + self.stream_id: Optional[int] = None + + def handle_received(self) -> None: + """Parse complete requests, preserving incomplete RTSP data.""" + while self.buffer: + request, _, rest = self.parse_request(self.buffer) + if request is None: + return + self.buffer = rest + headers = ( + {"CSeq": request.headers["CSeq"]} if "CSeq" in request.headers else {} + ) + self.send( + self.format_response( + HttpResponse( + request.protocol, request.version, 200, "OK", headers, b"" + ) + ) + ) + if (request.method, request.path) != ("POST", "/command"): + continue + if request.headers.get("X-Apple-StreamID") not in ( + None, + str(self.stream_id), + ): + continue + payload = BaseDataStreamChannel.decode_payload(cast(bytes, request.body)) + raw = ( + payload.get("params", {}).get("data") + if isinstance(payload, dict) + else None + ) + if isinstance(raw, bytes): + for message in BaseDataStreamChannel.decode_protobufs(raw): + self.queue.put_nowait(message) + + def connection_lost(self, exc: Optional[Exception]) -> None: + """Wake the consumer when the encrypted event connection ends.""" + self.queue.put_nowait(exc or EOFError("AirPlay event connection closed")) + + +class RelayConnection(AbstractMrpConnection): + """Carry protobuf messages over encrypted AirPlay control and event channels.""" + + def __init__(self, core: Core, credentials: HapCredentials) -> None: + """Initialize an independent controller for the advertised receiver.""" + super().__init__() + self.core = core + self.credentials = credentials + self.identifier = str(uuid4()).upper() + mac = bytearray(hashlib.sha256(self.identifier.encode()).digest()[:6]) + mac[0] = (mac[0] | 2) & 254 + self.mac = ":".join(f"{part:02X}" for part in mac) + self.connection: Optional[HttpConnection] = None + self.rtsp: Optional[RtspSession] = None + self.control_setup = False + self.events: Optional[RelayEvents] = None + self.stream_id: Optional[int] = None + self.tasks: Set[asyncio.Task] = set() + self.queue: asyncio.Queue[Union[protobuf.ProtocolMessage, Exception]] = ( + asyncio.Queue() + ) + self.outgoing: asyncio.Queue[protobuf.ProtocolMessage] = asyncio.Queue() + self.paths: Dict[tuple, protobuf.PlayerPath] = {} + self.lock = asyncio.Lock() + self.closing = False + self.close_task: Optional[asyncio.Task] = None + + @property + def connected(self) -> bool: + """Return whether both owned channels are ready.""" + return self.stream_id is not None and not self.closing + + def enable_encryption(self, output_key: bytes, input_key: bytes) -> None: + """Reject a second authentication layer inside the encrypted relay.""" + raise exceptions.NotSupportedError("Relay encryption is negotiated by AirPlay") + + async def exchange( + self, method: str, path: Optional[str] = None, **kwargs: Any + ) -> HttpResponse: + """Serialize control requests, including feedback and commands.""" + assert self.rtsp + headers = { + "User-Agent": "AirPlay/980.77.1", + "X-Apple-ProtocolVersion": "1", + **kwargs.pop("headers", {}), + } + async with self.lock, async_timeout(8): + return await self.rtsp.exchange(method, path, headers=headers, **kwargs) + + async def connect(self) -> None: + """Authenticate and subscribe without requesting an audio stream.""" + self.connection = await http_connect( + str(self.core.config.address), self.core.service.port + ) + self.rtsp = RtspSession(self.connection) + await self.exchange("GET", "/info", body={"qualifier": ["txtAirPlay"]}) + verifier = await verify_connection(self.credentials, self.connection) + await self.exchange( + "POST", + "/auth-setup", + headers={"X-Apple-AT": "16"}, + body={"ascm": 1, "tkrd": ["pair", "auth", "uuid"]}, + ) + response = await self.exchange( + "SETUP", + body={ + "isRemoteControlOnly": True, + "timingProtocol": "None", + "statsCollectionEnabled": False, + "updateSessionRequest": False, + "combinedGetInfoWithControlSetup": True, + "sessionUUID": str(uuid4()).upper(), + "sessionCorrelationUUID": self.identifier, + "deviceID": self.mac, + "macAddress": self.mac, + "name": "pyatv relay", + "model": "pyatv", + "osName": platform.system(), + "osVersion": platform.release(), + "sourceVersion": "980.77.1", + }, + ) + self.control_setup = True + event_port = decode_bplist_from_body(response)["eventPort"] + if ( + not isinstance(event_port, int) + or isinstance(event_port, bool) + or not 1 <= event_port <= 65535 + ): + raise exceptions.InvalidResponseError("Invalid eventPort") + await self.exchange("GET", "/info") + _, channel = await setup_channel( + lambda output, incoming: RelayEvents(output, incoming, self.queue), + verifier, + self.connection.remote_ip, + event_port, + "Events-Salt", + "Events-Read-Encryption-Key", + "Events-Write-Encryption-Key", + ) + self.events = cast(RelayEvents, channel) + self._background(self.consume()) + await self.exchange("RECORD") + response = await self.exchange( + "SETUP", + body={ + "streams": [ + { + "type": 130, + "controlType": 1, + "clientUUID": str(uuid4()).upper(), + "channelID": str(uuid4()).upper(), + "clientTypeUUID": "1910A70F-DBC0-4242-AF95-115DB30604E1", + } + ] + }, + ) + streams = decode_bplist_from_body(response).get("streams", []) + if ( + len(streams) != 1 + or streams[0].get("type") != 130 + or "dataPort" in streams[0] + ): + raise exceptions.InvalidResponseError( + "Expected a two-channel remote stream" + ) + stream_id = streams[0].get("streamID") + if ( + not isinstance(stream_id, int) + or isinstance(stream_id, bool) + or not 0 <= stream_id < 2**32 + ): + raise exceptions.InvalidResponseError("Invalid remote streamID") + self.stream_id = self.events.stream_id = stream_id + self._background(self._send_messages()) + self._background(self.feedback()) + + def _background(self, coroutine) -> None: + task = asyncio.create_task(coroutine) + self.tasks.add(task) + task.add_done_callback(self._completed) + + def _completed(self, task: asyncio.Task) -> None: + if task.cancelled() or self.closing: + return + error = task.exception() + if error: + _LOGGER.error("AirPlay relay connection failed: %s", error) + self.listener.stop() + self.core.device_listener.listener.connection_lost(error) + + def send(self, message: protobuf.ProtocolMessage) -> None: + """Queue a message, adjusting only the native relay subscription.""" + if not self.connected: + raise exceptions.InvalidStateError("Relay is not connected") + native = protobuf.ProtocolMessage() + native.CopyFrom(message) + if native.type == protobuf.DEVICE_INFO_MESSAGE: + inner = protobuf.extract_inner(native) + inner.Clear() + for field, value in { + "uniqueIdentifier": self.identifier, + "name": self.core.settings.info.name, + "localizedModelName": "Mac", + "applicationBundleIdentifier": "local.pyatv.relay", + "protocolVersion": 1, + "lastSupportedMessageType": 65, + "allowsPairing": False, + "supportsSharedQueue": True, + "sharedQueueVersion": 3, + "deviceUID": self.mac, + "deviceClass": protobuf.DeviceClass.Mac, + "logicalDeviceCount": 1, + "isProxyGroupPlayer": False, + "isAirplayActive": False, + "modelID": "pyatv", + }.items(): + setattr(inner, field, value) + elif native.type == protobuf.CLIENT_UPDATES_CONFIG_MESSAGE: + protobuf.extract_inner(native).CopyFrom( + protobuf.extract_inner(messages.client_updates_config(keyboard=False)) + ) + # Unnamed field 6 must be false for this native subscription. + native.MergeFromString(b"\xaa\x01\x02\x30\x00") + self.outgoing.put_nowait(native) + + async def _send_messages(self) -> None: + while True: + message = await self.outgoing.get() + await self.exchange( + "POST", + "/command", + headers={"X-Apple-StreamID": str(self.stream_id)}, + body={ + "params": { + "data": BaseDataStreamChannel.encode_protobufs([message]) + } + }, + ) + + async def consume(self) -> None: + """Deliver replies and cache full player paths without selecting a player.""" + while True: + message = await self.queue.get() + if isinstance(message, Exception): + raise message + if message.type == protobuf.SET_STATE_MESSAGE: + path = protobuf.extract_inner(message).playerPath + copy = protobuf.PlayerPath() + copy.CopyFrom(path) + self.paths[path.client.bundleIdentifier, path.player.identifier] = copy + self.listener.message_received(message, message.SerializeToString()) + + async def feedback(self) -> None: + """Keep the control-only session alive with two-second feedback.""" + while True: + await self.exchange("POST", "/feedback") + await asyncio.sleep(2) + + def close(self) -> None: + """Schedule teardown and expose its task to the owning facade.""" + if not self.closing: + self.closing = True + self.close_task = asyncio.create_task(self._close()) + + async def _close(self) -> None: + for task in self.tasks: + task.cancel() + await asyncio.gather(*self.tasks, return_exceptions=True) + try: + if self.rtsp and self.control_setup: + await asyncio.wait_for(self.exchange("TEARDOWN", body={}), 2) + except Exception: # pylint: disable=broad-except + _LOGGER.debug("Relay teardown failed", exc_info=True) + finally: + if self.events: + self.events.close() + if self.connection: + self.connection.close() + + +class RelayRemoteControl(MrpRemoteControl): + """Address playback commands to the globally selected sender player.""" + + def __init__(self, loop, psm, protocol) -> None: + """Reuse normal player selection and the native connection's full paths.""" + super().__init__(loop, psm, protocol) + self.connection = cast(RelayConnection, protocol.connection) + + async def _send_command(self, command, **kwargs): + if not self.psm.client: + raise exceptions.InvalidStateError("No active sender") + path = self.connection.paths.get( + (self.psm.client.bundle_identifier, self.psm.playing.identifier) + ) + if path is None: + raise exceptions.InvalidStateError("No active player path") + message = messages.command(command, **kwargs) + protobuf.extract_inner(message).playerPath.CopyFrom(path) + response = await self.protocol.send_and_receive(message) + if response.errorCode or response.type != protobuf.SEND_COMMAND_RESULT_MESSAGE: + raise exceptions.CommandError("Invalid or rejected relay command response") + result = protobuf.extract_inner(response) + if ( + result.sendError + or result.handlerReturnStatus + or result.commandResult.sendError + or any(status.statusCode for status in result.commandResult.statuses) + ): + raise exceptions.CommandError("Relay command rejected by sender") + + async def set_position(self, pos: int) -> None: + """Seek the selected sender to a finite, nonnegative position in seconds.""" + if not math.isfinite(pos) or pos < 0: + raise exceptions.CommandError("Position must be finite and nonnegative") + await self._send_command( + protobuf.CommandInfo_pb2.SeekToPlaybackPosition, + playbackPosition=pos, + sendOptions=0, + ) + + +class RelayAudio(MrpAudio): + """Control a receiver's volume using its reported group-member UID.""" + + def __init__(self, protocol, state_dispatcher, receiver_id: str) -> None: + """Bind volume to the receiver, independently of the source device.""" + self.receiver_id = receiver_id.replace(":", "").lower() + self.levels: Dict[str, float] = {} + self.changed = asyncio.Event() + super().__init__(protocol, state_dispatcher) + + @property + def device_uid(self) -> Optional[str]: + """Find the requested receiver among the sender's reported output devices.""" + for device in self.output_devices: + if device.identifier.replace(":", "").lower() == self.receiver_id: + return device.identifier + return None + + async def _volume_did_change(self, message) -> None: + inner = protobuf.extract_inner(message) + uid = inner.outputDeviceUID or inner.endpointUID + if not math.isfinite(inner.volume) or not 0 <= inner.volume <= 1: + return + self.levels[uid] = inner.volume * 100 + if uid == self.device_uid: + self._volume_controls_available = True + self._volume_controls_absolute = True + inner.outputDeviceUID = uid + await super()._volume_did_change(message) + self.changed.set() + + async def set_volume( + self, level: float, output_device: Optional[OutputDevice] = None + ) -> None: + """Set absolute volume and wait for a matching receiver update.""" + if not math.isfinite(level) or not 0 <= level <= 100: + raise exceptions.CommandError("Volume must be between 0 and 100") + uid = output_device.identifier if output_device else self.device_uid + if uid is None or uid not in { + device.identifier for device in self.output_devices + }: + raise exceptions.InvalidStateError( + "Receiver not found in reported output devices" + ) + self.changed.clear() + response = await self.protocol.send_and_receive( + messages.set_volume(uid, level / 100) + ) + if response.errorCode: + raise exceptions.CommandError("Relay volume command rejected") + async with async_timeout(5): + while abs(self.levels.get(uid, -1) - level) > 0.5: + self.changed.clear() + await self.changed.wait() + + +def setup(core: Core, credentials: HapCredentials) -> SetupData: + """Expose a native AirPlay relay through the existing pyatv interfaces.""" + if credentials.type != AuthenticationType.Transient: + raise exceptions.NotSupportedError( + "Experimental relay requires transient AirPlay pairing" + ) + connection = RelayConnection(core, credentials) + service = MutableService(None, Protocol.MRP, core.service.port, {}) + mrp_core = Core( + core.loop, + core.config, + service, + core.settings, + core.device_listener, + core.session_manager, + core.takeover, + core.state_dispatcher.create_copy(Protocol.MRP), + ) + data = create_with_connection( + mrp_core, + connection, + requires_heatbeat=False, + skip_keyboard=True, + remote_control_factory=RelayRemoteControl, + audio_factory=partial( + RelayAudio, receiver_id=core.service.properties.get("deviceid", "") + ), + ) + interfaces = dict(data.interfaces) + interfaces.pop(Power) + # Native receiver playback control does not provide a TV navigation interface. + supported = { + FeatureName.Artwork, + FeatureName.Play, + FeatureName.Pause, + FeatureName.PlayPause, + FeatureName.Next, + FeatureName.Previous, + FeatureName.SetPosition, + FeatureName.Title, + FeatureName.Artist, + FeatureName.Album, + FeatureName.Genre, + FeatureName.TotalTime, + FeatureName.Position, + FeatureName.Volume, + FeatureName.SetVolume, + FeatureName.VolumeUp, + FeatureName.VolumeDown, + FeatureName.App, + FeatureName.OutputDevices, + } + + async def connect() -> bool: + result = await data.connect() + metadata = cast(Metadata, interfaces[Metadata]) + # Initial player events can follow the subscription acknowledgment. + for _ in range(20): + if (await metadata.playing()).title: + break + await asyncio.sleep(0.1) + return result + + def close() -> Set[asyncio.Task]: + tasks = data.close() + if connection.close_task: + tasks.add(connection.close_task) + return tasks + + return SetupData(Protocol.MRP, connect, close, lambda: {}, interfaces, supported) diff --git a/pyatv/protocols/mrp/__init__.py b/pyatv/protocols/mrp/__init__.py index c83355cbd..b62e1803b 100644 --- a/pyatv/protocols/mrp/__init__.py +++ b/pyatv/protocols/mrp/__init__.py @@ -6,7 +6,18 @@ import logging import math import re -from typing import Any, Dict, Generator, List, Mapping, Optional, Set, Tuple, cast +from typing import ( + Any, + Callable, + Dict, + Generator, + List, + Mapping, + Optional, + Set, + Tuple, + cast, +) from aiohttp import ClientError, ClientSession @@ -1100,6 +1111,9 @@ def create_with_connection( # pylint: disable=too-many-locals core: Core, connection: AbstractMrpConnection, requires_heatbeat: bool = True, + skip_keyboard: bool = False, + remote_control_factory: Callable[..., MrpRemoteControl] = MrpRemoteControl, + audio_factory: Callable[..., MrpAudio] = MrpAudio, ) -> SetupData: """Set up a new MRP service from a connection.""" protocol = MrpProtocol( @@ -1107,13 +1121,13 @@ def create_with_connection( # pylint: disable=too-many-locals ) psm = PlayerStateManager(protocol) - remote_control = MrpRemoteControl(core.loop, psm, protocol) + remote_control = remote_control_factory(core.loop, psm, protocol) metadata = MrpMetadata( protocol, psm, core.config.identifier, core.session_manager.session ) power = MrpPower(core.loop, protocol, remote_control) push_updater = MrpPushUpdater(metadata, psm, core.state_dispatcher) - audio = MrpAudio(protocol, core.state_dispatcher) + audio = audio_factory(protocol, core.state_dispatcher) interfaces = { RemoteControl: remote_control, @@ -1125,7 +1139,7 @@ def create_with_connection( # pylint: disable=too-many-locals } async def _connect() -> bool: - await protocol.start() + await protocol.start(skip_keyboard=skip_keyboard) if requires_heatbeat: protocol.enable_heartbeat() return True diff --git a/pyatv/protocols/mrp/protocol.py b/pyatv/protocols/mrp/protocol.py index 21cbaa87d..ce27486ee 100644 --- a/pyatv/protocols/mrp/protocol.py +++ b/pyatv/protocols/mrp/protocol.py @@ -120,7 +120,9 @@ def __init__( self._outstanding: Dict[str, OutstandingMessage] = {} self._state: ProtocolState = ProtocolState.NOT_CONNECTED - async def start(self, skip_initial_messages: bool = False) -> None: + async def start( + self, skip_initial_messages: bool = False, skip_keyboard: bool = False + ) -> None: """Connect to device and listen to incoming messages.""" if self._state != ProtocolState.NOT_CONNECTED: raise exceptions.InvalidStateError(self._state.name) @@ -162,7 +164,8 @@ async def start(self, skip_initial_messages: bool = False) -> None: # Subscribe to updates at this stage await self.send_and_receive(messages.client_updates_config()) - await self.send_and_receive(messages.get_keyboard_session()) + if not skip_keyboard: + await self.send_and_receive(messages.get_keyboard_session()) except Exception: # Something went wrong, let's do cleanup self.stop() diff --git a/pyatv/settings.py b/pyatv/settings.py index a02d6e634..aba3aa608 100644 --- a/pyatv/settings.py +++ b/pyatv/settings.py @@ -68,6 +68,9 @@ class MrpTunnel(str, Enum): Force = "force" """Force set up of MRP tunnel even if remote device does not supports it.""" + Relay = "relay" + """Use the experimental controlType-1 relay for existing sender playback.""" + Disable = "disable" """Fully disable set up of MRP tunnel.""" diff --git a/tests/protocols/airplay/test_relay.py b/tests/protocols/airplay/test_relay.py new file mode 100644 index 000000000..b5f082047 --- /dev/null +++ b/tests/protocols/airplay/test_relay.py @@ -0,0 +1,504 @@ +"""Native AirPlay relay transport and public control API tests.""" + +import asyncio +from ipaddress import ip_address +import plistlib +from unittest.mock import AsyncMock, Mock + +import pytest +import pytest_asyncio + +import pyatv +from pyatv import exceptions +from pyatv.auth.hap_pairing import TRANSIENT_CREDENTIALS +from pyatv.conf import AppleTV +from pyatv.const import DeviceState, FeatureName, FeatureState, Protocol +from pyatv.core import MutableService +from pyatv.protocols.airplay import relay +from pyatv.protocols.airplay.channels import BaseDataStreamChannel +from pyatv.protocols.airplay.relay import RelayConnection, RelayEvents +from pyatv.protocols.mrp import messages, protobuf +from pyatv.settings import MrpTunnel, Settings +from pyatv.storage.memory_storage import MemoryStorage +from pyatv.support.http import HttpRequest, HttpResponse, format_request, parse_response + + +def player_path(bundle="selected"): + path = protobuf.PlayerPath() + path.origin.identifier = 1 + path.client.bundleIdentifier = bundle + path.client.processIdentifier = 42 + path.player.identifier = "MediaRemote-DefaultPlayer" + return path + + +def selected_state( + bundle="selected", state=protobuf.PlaybackState.Playing, artwork_available=False +): + msg = messages.create(protobuf.SET_STATE_MESSAGE) + inner = protobuf.extract_inner(msg) + inner.playerPath.CopyFrom(player_path(bundle)) + inner.playbackState = state + inner.playbackQueue.location = 0 + item = inner.playbackQueue.contentItems.add() + item.identifier = "item" + item.metadata.title = "Song" + item.metadata.trackArtistName = "Artist" + item.metadata.albumName = "Album" + item.metadata.duration = 300 + item.metadata.elapsedTime = 30 + item.metadata.elapsedTimeTimestamp = 1 + if artwork_available: + item.metadata.artworkAvailable = True + item.metadata.artworkMIMEType = "image/jpeg" + item.metadata.artworkIdentifier = "artwork-item" + for command in (1, 2, 3, 5, 6, 45): + info = inner.supportedCommands.supportedCommands.add() + info.command = command + info.enabled = True + return msg + + +@pytest.mark.asyncio +async def test_event_fragments_acknowledgments_batch_and_stream_filter(): + queue = asyncio.Queue() + channel = RelayEvents(bytes(32), bytes(32), queue) + channel.stream_id = 7 + channel.send = Mock() + message = selected_state() + data = BaseDataStreamChannel.encode_protobufs([message, message]) + body = plistlib.dumps({"params": {"data": data}}, fmt=plistlib.FMT_BINARY) + request = HttpRequest( + "POST", + "/command", + "RTSP", + "1.0", + { + "Content-Type": "application/x-apple-binary-plist", + "CSeq": "123", + "X-Apple-StreamID": "7", + }, + body, + ) + wire = format_request(request) + channel.buffer = wire[:20] + channel.handle_received() + channel.send.assert_not_called() + channel.buffer += wire[20:] + wire + channel.handle_received() + assert queue.qsize() == 4 + assert channel.send.call_count == 2 + response, rest = parse_response(channel.send.call_args.args[0]) + assert not rest and response.code == 200 and response.headers["CSeq"] == "123" + channel.buffer = wire.replace(b"StreamID: 7", b"StreamID: 8") + channel.handle_received() + assert channel.send.call_count == 3 and queue.qsize() == 4 + for _ in range(4): + queue.get_nowait() + channel.connection_lost(None) + assert isinstance(queue.get_nowait(), EOFError) + + +@pytest.fixture +def core(): + return Mock( + config=Mock(address="192.0.2.10"), service=Mock(port=7000), settings=Settings() + ) + + +@pytest.mark.asyncio +@pytest.mark.parametrize("dedicated", [False, True]) +async def test_native_sequence_negotiated_stream_and_cleanup( + monkeypatch, core, dedicated +): + conn = RelayConnection(core, TRANSIENT_CREDENTIALS) + http = Mock(remote_ip="192.0.2.10") + monkeypatch.setattr(relay, "http_connect", AsyncMock(return_value=http)) + monkeypatch.setattr(relay, "verify_connection", AsyncMock()) + events = RelayEvents(bytes(32), bytes(32), conn.queue) + events.close = Mock() + channel = AsyncMock(return_value=(Mock(), events)) + monkeypatch.setattr(relay, "setup_channel", channel) + requests = [] + stream = {"type": 130, "streamID": 7} + if dedicated: + stream["dataPort"] = 1234 + + async def exchange(method, path=None, **kwargs): + requests.append((method, path, kwargs)) + response = ( + {"streams": [stream]} + if "streams" in kwargs.get("body", {}) + else {"eventPort": 1234} + ) + return HttpResponse("RTSP", "1.0", 200, "OK", {}, plistlib.dumps(response)) + + conn.exchange = AsyncMock(side_effect=exchange) + try: + if dedicated: + with pytest.raises(exceptions.InvalidResponseError, match="two-channel"): + await conn.connect() + else: + await conn.connect() + assert conn.connected and conn.stream_id == events.stream_id == 7 + assert [(method, path) for method, path, _ in requests] == [ + ("GET", "/info"), + ("POST", "/auth-setup"), + ("SETUP", None), + ("GET", "/info"), + ("RECORD", None), + ("SETUP", None), + ] + assert requests[1][2]["headers"] == {"X-Apple-AT": "16"} + assert requests[1][2]["body"] == {"ascm": 1, "tkrd": ["pair", "auth", "uuid"]} + assert requests[2][2]["body"]["isRemoteControlOnly"] + assert requests[-1][2]["body"]["streams"][0]["controlType"] == 1 + assert "seed" not in requests[-1][2]["body"]["streams"][0] + assert channel.call_args.args[-3:] == ( + "Events-Salt", + "Events-Read-Encryption-Key", + "Events-Write-Encryption-Key", + ) + finally: + conn.close() + await conn.close_task + assert requests[-1][0] == "TEARDOWN" + assert not conn.connected + http.close.assert_called_once() + events.close.assert_called_once() + + +@pytest.mark.asyncio +async def test_authentication_failure_does_not_teardown_unestablished_session( + monkeypatch, core +): + conn = RelayConnection(core, TRANSIENT_CREDENTIALS) + http = Mock() + monkeypatch.setattr(relay, "http_connect", AsyncMock(return_value=http)) + monkeypatch.setattr( + relay, + "verify_connection", + AsyncMock(side_effect=exceptions.AuthenticationError("rejected")), + ) + conn.exchange = AsyncMock() + try: + with pytest.raises(exceptions.AuthenticationError): + await conn.connect() + finally: + conn.close() + await conn.close_task + assert [call.args[0] for call in conn.exchange.call_args_list] == ["GET"] + http.close.assert_called_once() + + +class Receiver: + """Mock network boundary while retaining the real public API and protobufs.""" + + def __init__(self): + self.messages = [] + self.connection = None + self.uid = "02:00:00:00:00:01" + self.reject = None + self.disconnected = False + self.confirm_volume = True + self.artwork_available = False + self.artwork_data = b"synthetic-artwork" + + def volume(self, level, uid=None): + message = messages.create(protobuf.VOLUME_DID_CHANGE_MESSAGE) + inner = protobuf.extract_inner(message) + inner.outputDeviceUID = uid or self.uid + inner.volume = level + return message + + async def exchange(self, connection, method, path=None, **kwargs): + self.connection = connection + body = kwargs.get("body", {}) + reply = {} + if method == "SETUP": + reply = ( + {"streams": [{"type": 130, "streamID": 7}]} + if "streams" in body + else {"eventPort": 1234} + ) + elif method == "TEARDOWN": + self.disconnected = True + elif path == "/command": + message = BaseDataStreamChannel.decode_protobufs(body["params"]["data"])[0] + self.messages.append(message) + response = messages.create( + protobuf.GENERIC_MESSAGE, identifier=message.identifier + ) + if message.type == protobuf.DEVICE_INFO_MESSAGE: + response = messages.create( + protobuf.DEVICE_INFO_MESSAGE, identifier=message.identifier + ) + info = protobuf.extract_inner(response) + info.deviceUID = "sender-uid" + info.name = "Sender" + for uid, name in ( + (self.uid, "Receiver"), + ("02:00:00:00:00:02", "Other"), + ): + member = info.groupedDevices.add() + member.deviceUID = uid + member.name = name + elif message.type == protobuf.CLIENT_UPDATES_CONFIG_MESSAGE: + select = messages.create(protobuf.SET_NOW_PLAYING_CLIENT_MESSAGE) + protobuf.extract_inner(select).client.CopyFrom(player_path().client) + for update in ( + select, + selected_state(artwork_available=self.artwork_available), + selected_state("inactive", 2), + self.volume(0.25), + ): + connection.queue.put_nowait(update) + elif message.type == protobuf.PLAYBACK_QUEUE_REQUEST_MESSAGE: + response = selected_state(artwork_available=self.artwork_available) + response.identifier = message.identifier + item = protobuf.extract_inner(response).playbackQueue.contentItems[0] + item.artworkData = self.artwork_data + item.artworkDataWidth = 600 + item.artworkDataHeight = 600 + elif message.type == protobuf.SEND_COMMAND_MESSAGE: + response = messages.command_result(message.identifier) + if self.reject == "envelope": + response.errorCode = 1 + elif self.reject: + result = protobuf.extract_inner(response) + if self.reject == "status": + result.commandResult.statuses.add().statusCode = 1 + elif self.reject == "nested": + result.commandResult.sendError = 1 + else: + setattr(result, self.reject, 1) + elif message.type == protobuf.SET_VOLUME_MESSAGE: + inner = protobuf.extract_inner(message) + # An unrelated group update must not confirm the requested change. + connection.queue.put_nowait(self.volume(0.1, "02:00:00:00:00:02")) + if self.confirm_volume: + connection.queue.put_nowait( + self.volume(inner.volume, inner.outputDeviceUID) + ) + if message.identifier: + connection.queue.put_nowait(response) + return HttpResponse("RTSP", "1.0", 200, "OK", {}, plistlib.dumps(reply)) + + +@pytest_asyncio.fixture +async def connected_receiver(monkeypatch, request): + receiver = Receiver() + receiver.artwork_available = getattr(request, "param", False) + monkeypatch.setattr( + relay, "http_connect", AsyncMock(return_value=Mock(remote_ip="192.0.2.10")) + ) + monkeypatch.setattr(relay, "verify_connection", AsyncMock()) + + async def channel(factory, *args): + event = factory(bytes(32), bytes(32)) + event.close = Mock() + return Mock(), event + + async def exchange(connection, *args, **kwargs): + return await receiver.exchange(connection, *args, **kwargs) + + monkeypatch.setattr(relay, "setup_channel", channel) + monkeypatch.setattr(RelayConnection, "exchange", exchange) + config = AppleTV(ip_address("192.0.2.10"), "Receiver") + config.add_service( + MutableService( + "receiver", + Protocol.AirPlay, + 7000, + {"features": "0xC05F8A00,0x1C340", "deviceid": receiver.uid}, + ) + ) + storage = MemoryStorage() + settings = await storage.get_settings(config) + settings.protocols.airplay.mrp_tunnel = MrpTunnel.Relay + atv = await pyatv.connect(config, asyncio.get_running_loop(), storage=storage) + yield atv, receiver + await asyncio.gather(*atv.close()) + assert receiver.disconnected + assert all(task.done() for task in receiver.connection.tasks) + + +@pytest.mark.asyncio +async def test_public_api_metadata_native_subscription_and_scoped_selection( + connected_receiver, +): + atv, receiver = connected_receiver + playing = await atv.metadata.playing() + assert ( + playing.title == "Song" + and playing.artist == "Artist" + and playing.album == "Album" + ) + assert playing.device_state == DeviceState.Playing + assert [message.type for message in receiver.messages] == [15, 38, 16] + info = protobuf.extract_inner(receiver.messages[0]) + assert not info.allowsPairing and info.supportsSharedQueue + assert info.deviceClass == protobuf.DeviceClass.Mac + updates = protobuf.extract_inner(receiver.messages[2]) + assert ( + updates.artworkUpdates and updates.volumeUpdates and updates.outputDeviceUpdates + ) + assert not updates.keyboardUpdates and not updates.nowPlayingUpdates + assert b"\x30\x00" in updates.SerializeToString() + assert atv.features.get_feature(FeatureName.Pause).state == FeatureState.Available + assert atv.features.get_feature(FeatureName.Home).state == FeatureState.Unsupported + assert atv.audio.volume == 25 + assert len(atv.audio.output_devices) == 2 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "action,command", + [ + ("play", 1), + ("pause", 2), + ("play_pause", 3), + ("next", 5), + ("previous", 6), + ("set_position", 45), + ], +) +async def test_public_playback_commands_target_selected_player( + connected_receiver, action, command +): + atv, receiver = connected_receiver + ( + await getattr(atv.remote_control, action)(90) + if action == "set_position" + else await getattr(atv.remote_control, action)() + ) + inner = protobuf.extract_inner(receiver.messages[-1]) + assert inner.command == command + assert inner.playerPath == player_path() + if action == "set_position": + assert inner.options.playbackPosition == 90 and inner.options.sendOptions == 0 + + +@pytest.mark.asyncio +@pytest.mark.parametrize( + "rejected", ["sendError", "handlerReturnStatus", "nested", "status", "envelope"] +) +async def test_public_control_rejection_is_not_reported_as_success( + connected_receiver, rejected +): + atv, receiver = connected_receiver + receiver.reject = rejected + with pytest.raises(exceptions.CommandError): + await atv.remote_control.pause() + + +@pytest.mark.asyncio +async def test_public_volume_targets_receiver_or_explicit_group_member( + connected_receiver, +): + atv, receiver = connected_receiver + await atv.audio.set_volume(50) + assert atv.audio.volume == 50 + inner = protobuf.extract_inner(receiver.messages[-1]) + assert inner.outputDeviceUID == receiver.uid and inner.volume == 0.5 + other = atv.audio.output_devices[1] + await atv.audio.set_volume(40, other) + inner = protobuf.extract_inner(receiver.messages[-1]) + assert inner.outputDeviceUID == other.identifier + assert inner.volume == pytest.approx(0.4) + assert atv.audio.volume == 50 + + +@pytest.mark.asyncio +@pytest.mark.parametrize("value", [-1, float("nan"), float("inf")]) +async def test_invalid_seek_is_rejected_before_sending(connected_receiver, value): + atv, receiver = connected_receiver + before = len(receiver.messages) + with pytest.raises(exceptions.CommandError): + await atv.remote_control.set_position(value) + assert len(receiver.messages) == before + + +@pytest.mark.asyncio +async def test_missing_player_path_is_rejected_before_sending(connected_receiver): + atv, receiver = connected_receiver + receiver.connection.paths.clear() + before = len(receiver.messages) + with pytest.raises(exceptions.InvalidStateError): + await atv.remote_control.pause() + assert len(receiver.messages) == before + + +@pytest.mark.asyncio +async def test_unrelated_group_volume_does_not_confirm_receiver_change( + connected_receiver, monkeypatch +): + atv, receiver = connected_receiver + receiver.confirm_volume = False + original_timeout = relay.async_timeout + monkeypatch.setattr(relay, "async_timeout", lambda _: original_timeout(0.05)) + with pytest.raises(asyncio.TimeoutError): + await atv.audio.set_volume(50) + assert atv.audio.volume == 25 + + +@pytest.mark.asyncio +async def test_missing_receiver_uid_never_falls_back_to_sender(connected_receiver): + atv, receiver = connected_receiver + for device in atv.audio.output_devices: + device.identifier = "unknown" + before = len(receiver.messages) + with pytest.raises(exceptions.InvalidStateError): + await atv.audio.set_volume(50) + assert len(receiver.messages) == before + + +@pytest.mark.asyncio +@pytest.mark.parametrize("connected_receiver", [False, True], indirect=True) +async def test_public_artwork_availability_retrieval_cache_and_track_change( + connected_receiver, +): + atv, receiver = connected_receiver + available = receiver.artwork_available + assert atv.features.get_feature(FeatureName.Artwork).state == ( + FeatureState.Available if available else FeatureState.Unavailable + ) + artwork = await atv.metadata.artwork(width=512) + requests = [ + message + for message in receiver.messages + if message.type == protobuf.PLAYBACK_QUEUE_REQUEST_MESSAGE + ] + if not available: + assert artwork is None and not requests + return + assert artwork.bytes == receiver.artwork_data + assert artwork.mimetype == "image/jpeg" + assert (artwork.width, artwork.height) == (600, 600) + assert len(requests) == 1 + request = protobuf.extract_inner(requests[0]) + assert request.location == 0 and request.length == 1 + assert request.artworkWidth == 512 + assert request.returnContentItemAssetsInUserCompletion + cached = await atv.metadata.artwork(width=512) + assert cached == artwork + assert len(receiver.messages) == 4 # Subscription plus one artwork request. + + update = selected_state(artwork_available=True) + item = protobuf.extract_inner(update).playbackQueue.contentItems[0] + item.identifier = "next-item" + item.metadata.title = "Next song" + item.metadata.artworkIdentifier = "artwork-next" + receiver.artwork_data = b"synthetic-next-artwork" + receiver.connection.queue.put_nowait(update) + + async def wait_for_track(): + while atv.metadata.artwork_id != "artwork-next": + await asyncio.sleep(0) + + await asyncio.wait_for(wait_for_track(), 1) + next_artwork = await atv.metadata.artwork(width=512) + assert next_artwork.bytes == receiver.artwork_data + assert next_artwork.bytes != artwork.bytes + assert receiver.messages[-1].type == protobuf.PLAYBACK_QUEUE_REQUEST_MESSAGE + assert len(receiver.messages) == 5