GitHub/gamdl
1
0
Fork 0
mirror of https://github.com/glomatico/gamdl.git synced 2026-08-03 13:27:13 +03:00
Code Issues Packages Projects Releases Wiki Activity

Route wrapper decrypt through native TCP engine

Browse Source
This commit is contained in:
glomatico
2026-07-08 11:51:33 -03:00
parent 2e55b841ff
commit e9ab35c9d6
5 changed files with 69 additions and 130 deletions
Download Patch File Download Diff File Expand all files Collapse all files
gamdl/api/wrapper.py
+7 -117
View File
@@ -1,7 +1,6 @@
from __future__ import annotations
import inspect
import struct
from collections.abc import Awaitable, Callable
from typing import TypeVar
@@ -31,82 +30,23 @@ class WrapperApi:
def __init__(
self,
base_url: str,
decrypt_host: str,
decrypt_port: int,
client: httpx.AsyncClient,
me: dict,
):
self.base_url = base_url
self.decrypt_host = decrypt_host
self.decrypt_port = decrypt_port
self.client = client
self.me = me
@staticmethod
def build_decrypt_sample_frame(
adam_id: str,
skd_uri: str,
ciphertexts: list[bytes],
) -> bytes:
"""Build wrapper-v2 /decrypt binary request frame."""
adam_id_bytes = adam_id.encode("utf-8")
skd_uri_bytes = skd_uri.encode("utf-8")
if not adam_id_bytes:
raise ValueError("wrapper-v2: adam_id must not be empty")
if not skd_uri_bytes:
raise ValueError("wrapper-v2: skd_uri must not be empty")
if not ciphertexts:
raise ValueError("wrapper-v2: ciphertext batch must not be empty")
frame = bytearray()
frame += struct.pack(
">III",
len(adam_id_bytes),
len(skd_uri_bytes),
len(ciphertexts),
)
for ciphertext in ciphertexts:
frame += struct.pack(">I", len(ciphertext))
frame += adam_id_bytes
frame += skd_uri_bytes
for ciphertext in ciphertexts:
frame += ciphertext
return bytes(frame)
@staticmethod
def parse_decrypt_sample_frame(data: bytes, expected_count: int) -> list[bytes]:
"""Parse wrapper-v2 /decrypt binary response frame."""
if len(data) < 4:
raise IOError("wrapper-v2: POST /decrypt returned a truncated response")
(sample_count,) = struct.unpack_from(">I", data, 0)
if sample_count != expected_count:
raise IOError(
f"wrapper-v2: expected {expected_count} samples in response, "
f"got {sample_count}"
)
table_end = 4 + sample_count * 4
if len(data) < table_end:
raise IOError("wrapper-v2: POST /decrypt returned a truncated length table")
lengths = [
struct.unpack_from(">I", data, 4 + i * 4)[0] for i in range(sample_count)
]
offset = table_end
out: list[bytes] = []
for i, length in enumerate(lengths):
end = offset + length
if end > len(data):
raise IOError(
f"wrapper-v2: POST /decrypt returned truncated sample {i}"
)
out.append(data[offset:end])
offset = end
if offset != len(data):
raise IOError("wrapper-v2: POST /decrypt returned trailing bytes")
return out
@classmethod
async def create(
cls,
base_url: str = "http://127.0.0.1",
decrypt_host: str = "127.0.0.1",
decrypt_port: int = 10020,
get_credentials_func: CredentialsFunc | None = None,
get_2fa_code: TwoFactorCodeFunc | None = None,
) -> WrapperApi:
@@ -134,7 +74,7 @@ class WrapperApi:
"Provide get_credentials_func or log in via the wrapper.",
)
return cls(base_url, client, me)
return cls(base_url, decrypt_host, decrypt_port, client, me)
@staticmethod
async def login(
@@ -220,53 +160,3 @@ class WrapperApi:
log.debug("success", playback=playback)
return playback
async def decrypt(
self,
adam_id: str,
skd_uri: str,
ciphertexts: list[bytes],
) -> list[bytes]:
"""Decrypt one POST /decrypt batch; plaintexts match ciphertext order."""
log = logger.bind(
action="wrapper_decrypt",
adam_id=adam_id,
sample_count=len(ciphertexts),
)
frame = self.build_decrypt_sample_frame(adam_id, skd_uri, ciphertexts)
response = await self.client.post(
f"{self.base_url}/decrypt",
content=frame,
headers={
"content-type": "application/octet-stream",
"accept": "application/octet-stream",
},
)
if response.status_code == 401:
raise IOError(
"wrapper-v2: POST /decrypt returned 401 — log in with POST /login "
"or restore a session on the daemon first"
)
if response.status_code == 503:
raise IOError(
"wrapper-v2: decrypt unavailable (503) — check daemon logs /health "
"for playback_ready and Apple lib init"
)
if response.status_code != 200:
detail = ""
try:
j = response.json()
detail = (j.get("detail") or j.get("error") or str(j)) or ""
except Exception:
detail = (response.text or "")[:500]
raise IOError(
f"wrapper-v2: POST /decrypt failed HTTP {response.status_code}: {detail}"
)
plaintexts = self.parse_decrypt_sample_frame(
response.content,
len(ciphertexts),
)
log.debug("success")
return plaintexts
gamdl/cli/cli.py
+2
View File
@@ -87,6 +87,8 @@ async def main(config: CliConfig):
try:
wrapper_api = await WrapperApi.create(
base_url=config.wrapper_url,
decrypt_host=config.wrapper_decrypt_host,
decrypt_port=config.wrapper_decrypt_port,
get_credentials_func=InteractivePrompts.get_wrapper_credentials,
get_2fa_code=InteractivePrompts.get_wrapper_2fa_code,
)
gamdl/cli/cli_config.py
+16
View File
@@ -154,6 +154,22 @@ class CliConfig:
default=wrapper_api_create_sig.parameters["base_url"].default,
),
]
wrapper_decrypt_host: Annotated[
str,
option(
"--wrapper-decrypt-host",
help="Wrapper TCP decrypt host",
default=wrapper_api_create_sig.parameters["decrypt_host"].default,
),
]
wrapper_decrypt_port: Annotated[
int,
option(
"--wrapper-decrypt-port",
help="Wrapper TCP decrypt port",
default=wrapper_api_create_sig.parameters["decrypt_port"].default,
),
]
# API specific options
cookies_path: Annotated[
str,
gamdl/downloader/amdecrypt.py
+22 -13
View File
@@ -2,8 +2,8 @@
This is a modified version of https://github.com/sn0wst0rm/st0rmMusicPlayer/blob/main/scripts/amdecrypt.py
All the modifications made here were AI generated
FairPlay sample decryption talks to wrapper-v2 over HTTP POST /decrypt
(binary frame), not the legacy raw TCP port used by the original wrapper (e.g. 10020).
FairPlay sample decryption talks to wrapper-v2 over the raw TCP decrypt port
while HTTP remains reserved for account/playback control calls.
"""
from __future__ import annotations
@@ -22,6 +22,7 @@ import structlog
from Crypto.Cipher import AES
from .. import _amdecrypt
from ..api.wrapper import WrapperApi
logger = structlog.get_logger(__name__)
@@ -32,8 +33,8 @@ DEFAULT_SONG_DECRYPTION_KEY = b"2\xb8\xad\xe1v\x9e&\xb1\xff\xb8\x98cRy?\xc6"
# Pre-fetch key used for first sample description
PREFETCH_KEY = "skd://itunes.apple.com/P000000000/s1/e1"
# Max ciphertext blobs per POST /decrypt (same adam_id + uri). Increase for fewer
# round-trips; set to 1 if a given wrapper build mis-handles CBC between chunks.
# Max ciphertext blobs per TCP decrypt batch (same adam_id + uri). Increase for
# fewer round-trips; set to 1 if a given wrapper build mis-handles CBC between chunks.
WRAPPER_DECRYPT_BATCH_SIZE = 128
# wrapper-v2: use one SKD segment per ``adam_id``+``uri`` (do not interleave prefetch
@@ -995,8 +996,8 @@ async def decrypt_samples(
decrypted_data_path: Optional[str] = None,
) -> bytes:
"""
Send track-key samples to wrapper-v2 (HTTP POST /decrypt) for CBCS
decryption and decrypt default prefetch-key samples locally.
Send track-key samples to wrapper-v2 over raw TCP for CBCS decryption and
decrypt default prefetch-key samples locally.
Ciphertext is sent in batches of up to :data:`WRAPPER_DECRYPT_BATCH_SIZE` MP4 samples
per request (same ``adam_id`` and ``uri``). Literal or tail-only samples are applied
@@ -1042,14 +1043,22 @@ async def decrypt_samples(
return
if segment_adam is None or segment_uri is None:
raise IOError("wrapper-v2: internal error (segment without adam/uri)")
chunks = [t[1] for t in crypto_batch]
tails = [t[2] for t in crypto_batch]
sources = [t[0] for t in crypto_batch]
plains = await wrapper_api.decrypt(segment_adam, segment_uri, chunks)
if len(plains) != len(chunks):
native_items = [
(sample.data, aligned, tail, sample.subsamples)
for sample, aligned, tail in crypto_batch
]
reassembled = await asyncio.to_thread(
_amdecrypt.wrapper_decrypt_reassemble,
wrapper_api.decrypt_host,
wrapper_api.decrypt_port,
segment_adam,
segment_uri,
native_items,
)
if len(reassembled) != len(crypto_batch):
raise IOError("wrapper-v2: plaintext batch count mismatch")
for s, plain, tail in zip(sources, plains, tails):
emit(_reassemble_cbcs_sample(s, plain, tail))
for sample_data in reassembled:
emit(sample_data)
crypto_batch.clear()
try:
tests/test_wrapper_api.py
+22
View File
@@ -0,0 +1,22 @@
import httpx
from gamdl.api.wrapper import WrapperApi
def test_wrapper_api_is_http_control_plane_only():
api = WrapperApi(
"http://127.0.0.1",
"127.0.0.1",
10020,
httpx.AsyncClient(),
{"auth": {"state": "authenticated"}},
)
try:
assert api.base_url == "http://127.0.0.1"
assert api.decrypt_host == "127.0.0.1"
assert api.decrypt_port == 10020
assert not hasattr(api, "decrypt")
finally:
import anyio
anyio.run(api.client.aclose)