/
/
/
1"""Streaming and Widevine decryption for Apple Music."""
2
3from __future__ import annotations
4
5import base64
6import json
7import os
8from typing import TYPE_CHECKING, Any, cast
9
10import aiofiles
11from aiohttp.client_exceptions import ClientError
12from music_assistant_models.enums import ContentType, StreamType
13from music_assistant_models.errors import MediaNotFoundError
14from music_assistant_models.media_items import AudioFormat
15from music_assistant_models.streamdetails import StreamDetails
16from pywidevine import PSSH, Cdm, Device, DeviceTypes
17from pywidevine.license_protocol_pb2 import WidevinePsshData
18from shortuuid import uuid
19
20from music_assistant.helpers.json import json_loads
21from music_assistant.helpers.playlists import fetch_playlist
22
23from .constants import (
24 CACHE_CATEGORY_DECRYPT_KEY,
25 DECRYPT_CLIENT_ID_FILENAME,
26 DECRYPT_PRIVATE_KEY_FILENAME,
27 WIDEVINE_BASE_PATH,
28)
29from .helpers.utils import is_library_id
30
31if TYPE_CHECKING:
32 from .provider import AppleMusicProvider
33
34
35class AppleMusicStreamingManager:
36 """Handles stream URL resolution and Widevine decryption."""
37
38 def __init__(self, provider: AppleMusicProvider) -> None:
39 """Initialize streaming manager."""
40 self.provider = provider
41 self.api = provider.api_client
42 self.logger = provider.logger
43 self._decrypt_client_id: bytes | None = None
44 self._decrypt_private_key: bytes | None = None
45 self._session_id: str | None = None
46
47 async def initialize(self) -> None:
48 """Load Widevine CDM files and create a session ID."""
49 self._session_id = str(uuid())
50 try:
51 async with aiofiles.open(
52 os.path.join(WIDEVINE_BASE_PATH, DECRYPT_CLIENT_ID_FILENAME), "rb"
53 ) as _file:
54 self._decrypt_client_id = await _file.read()
55 async with aiofiles.open(
56 os.path.join(WIDEVINE_BASE_PATH, DECRYPT_PRIVATE_KEY_FILENAME), "rb"
57 ) as _file:
58 self._decrypt_private_key = await _file.read()
59 except FileNotFoundError:
60 self.logger.warning(
61 "Widevine CDM files not found at %s. "
62 "Streaming of encrypted catalog tracks will not work. "
63 "Radio stations and unencrypted library tracks are still available.",
64 WIDEVINE_BASE_PATH,
65 )
66
67 async def get_stream_details(self, item_id: str) -> StreamDetails:
68 """Return StreamDetails for a single catalog or library track."""
69 stream_metadata = await self._fetch_song_stream_metadata(item_id)
70 if is_library_id(item_id):
71 try:
72 stream_url = stream_metadata["assets"][0]["URL"]
73 except (KeyError, IndexError, TypeError) as exc:
74 raise MediaNotFoundError(
75 f"Failed to extract stream URL for library track {item_id}: {exc}"
76 ) from exc
77 return StreamDetails(
78 item_id=item_id,
79 provider=self.provider.instance_id,
80 path=stream_url,
81 stream_type=StreamType.HTTP,
82 audio_format=AudioFormat(content_type=ContentType.UNKNOWN),
83 can_seek=True,
84 allow_seek=True,
85 )
86 if not self._decrypt_client_id or not self._decrypt_private_key:
87 raise MediaNotFoundError(
88 "Widevine CDM files are not available. "
89 "Cannot stream encrypted catalog tracks without them."
90 )
91 license_url = stream_metadata["hls-key-server-url"]
92 stream_url, uri = await self._parse_stream_url_and_uri(stream_metadata["assets"])
93 if not stream_url or not uri:
94 raise MediaNotFoundError("No stream URL found for song.")
95 key_id = base64.b64decode(uri.split(",")[1])
96 return StreamDetails(
97 item_id=item_id,
98 provider=self.provider.instance_id,
99 audio_format=AudioFormat(content_type=ContentType.MP4, codec_type=ContentType.AAC),
100 stream_type=StreamType.ENCRYPTED_HTTP,
101 decryption_key=await self._get_decryption_key(license_url, key_id, uri, item_id),
102 path=stream_url,
103 can_seek=True,
104 allow_seek=True,
105 )
106
107 async def _fetch_song_stream_metadata(self, song_id: str) -> dict[str, Any]:
108 """Get the stream metadata for a song from Apple Music."""
109 playback_url = "https://play.music.apple.com/WebObjects/MZPlay.woa/wa/webPlayback"
110 data: dict[str, Any] = {}
111 self.logger.debug("_fetch_song_stream_metadata: Check if Library ID: %s", song_id)
112 if is_library_id(song_id):
113 data["universalLibraryId"] = song_id
114 data["isLibrary"] = True
115 else:
116 data["salableAdamId"] = song_id
117 for retry in (True, False):
118 try:
119 async with self.provider.mass.http_session.post(
120 playback_url,
121 headers=self._decryption_headers,
122 json=data,
123 ssl=True,
124 ) as response:
125 response.raise_for_status()
126 content = await response.json(loads=json_loads)
127 if content.get("failureType"):
128 message = content.get("failureMessage")
129 raise MediaNotFoundError(f"Failed to get song stream metadata: {message}")
130 return cast("dict[str, Any]", content["songList"][0])
131 except (MediaNotFoundError, ClientError) as exc:
132 if retry:
133 self.logger.warning("Failed to get song stream metadata: %s", exc)
134 continue
135 raise
136 raise MediaNotFoundError(f"Failed to get song stream metadata for {song_id}")
137
138 async def _parse_stream_url_and_uri(
139 self, stream_assets: list[dict[str, Any]]
140 ) -> tuple[str | None, str | None]:
141 """Parse the stream URL and key URI from the song assets."""
142 ctrp256_urls = [asset["URL"] for asset in stream_assets if asset["flavor"] == "28:ctrp256"]
143 if not ctrp256_urls:
144 raise MediaNotFoundError("No ctrp256 URL found for song.")
145 playlist_url = ctrp256_urls[0]
146 playlist_items = await fetch_playlist(
147 self.provider.mass, ctrp256_urls[0], raise_on_hls=False
148 )
149 playlist_item = playlist_items[0]
150 base_path = playlist_url.rsplit("/", 1)[0]
151 track_url = base_path + "/" + playlist_items[0].path
152 key = playlist_item.key
153 return (track_url, key)
154
155 @property
156 def _decryption_headers(self) -> dict[str, str]:
157 """Return headers required for decryption requests."""
158 return {
159 "authorization": f"Bearer {self.provider._music_app_token}",
160 "media-user-token": cast("str", self.provider._music_user_token),
161 "connection": "keep-alive",
162 "accept": "application/json",
163 "origin": "https://music.apple.com",
164 "referer": "https://music.apple.com/",
165 "accept-encoding": "gzip, deflate, br",
166 "content-type": "application/json;charset=utf-8",
167 "user-agent": (
168 "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36"
169 " (KHTML, like Gecko) Chrome/110.0.0.0 Safari/537.36"
170 ),
171 }
172
173 async def _get_decryption_key(
174 self, license_url: str, key_id: bytes, uri: str, item_id: str
175 ) -> str:
176 """Get (or retrieve from cache) the decryption key for a song."""
177 if decryption_key := await self.provider.mass.cache.get(
178 key=item_id,
179 provider=self.provider.instance_id,
180 category=CACHE_CATEGORY_DECRYPT_KEY,
181 checksum=self._session_id,
182 ):
183 self.logger.debug("Decryption key for %s found in cache.", item_id)
184 return cast("str", decryption_key)
185 pssh = self._build_pssh(key_id)
186 device = Device(
187 client_id=self._decrypt_client_id,
188 private_key=self._decrypt_private_key,
189 type_=DeviceTypes.ANDROID,
190 security_level=3,
191 flags={},
192 )
193 cdm = Cdm.from_device(device)
194 session_id = cdm.open()
195 try:
196 challenge = cdm.get_license_challenge(session_id, pssh)
197 track_license = await self._get_license(challenge, license_url, uri, item_id)
198 cdm.parse_license(session_id, track_license)
199 key = next((key for key in cdm.get_keys(session_id) if key.type == "CONTENT"), None)
200 if not key:
201 raise MediaNotFoundError(f"Unable to get decryption key for song {item_id}.")
202 decryption_key = key.key.hex()
203 finally:
204 cdm.close(session_id)
205 self.provider.mass.create_task(
206 self.provider.mass.cache.set(
207 key=item_id,
208 data=decryption_key,
209 expiration=3600,
210 provider=self.provider.instance_id,
211 category=CACHE_CATEGORY_DECRYPT_KEY,
212 checksum=self._session_id,
213 )
214 )
215 return decryption_key
216
217 def _build_pssh(self, key_id: bytes) -> PSSH:
218 """Build a Widevine PSSH object for the given key ID."""
219 pssh_data = WidevinePsshData()
220 pssh_data.algorithm = WidevinePsshData.AESCTR
221 pssh_data.key_ids.append(key_id)
222 init_data = base64.b64encode(pssh_data.SerializeToString()).decode("utf-8")
223 return PSSH.new(system_id=PSSH.SystemId.Widevine, init_data=init_data)
224
225 async def _get_license(self, challenge: bytes, license_url: str, uri: str, item_id: str) -> str:
226 """Request a Widevine license from Apple Music."""
227 challenge_b64 = base64.b64encode(challenge).decode("utf-8")
228 data = {
229 "challenge": challenge_b64,
230 "key-system": "com.widevine.alpha",
231 "uri": uri,
232 "adamId": item_id,
233 "isLibrary": False,
234 "user-initiated": True,
235 }
236 async with self.provider.mass.http_session.post(
237 license_url,
238 data=json.dumps(data),
239 headers=self._decryption_headers,
240 ssl=False,
241 ) as response:
242 response.raise_for_status()
243 content = await response.json(loads=json_loads)
244 track_license = content.get("license")
245 if not track_license:
246 raise MediaNotFoundError(f"No license found for song {item_id}.")
247 return cast("str", track_license)
248