"""Decryption of Tidal's encrypted (bento) audio streams.
Tidal delivers HIGH-quality AAC (and some lossless renditions) AES-CTR
encrypted. The bento manifest carries a base64 *security token*;
decrypted with a fixed public master key it yields the real 16-byte
AES key and an 8-byte nonce. The counter block is ``nonce || 8-byte
big-endian counter``, so a file offset maps directly to a counter
start — decryption can run in bounded chunks and resume from any
16-byte-aligned offset.
"""
from __future__ import annotations
import base64
import logging
from collections.abc import Callable
from pathlib import Path
from typing import Any
from Crypto.Cipher import AES
from Crypto.Util import Counter
logger = logging.getLogger("tidalpy.decrypt")
# Public constant from Tidal's web player; not a secret.
_MASTER_KEY_B64 = "UIlTTEMmmLfGowo/UC60x2H45W6MdGgTRfo/umg4754="
_MASTER_KEY = base64.b64decode(_MASTER_KEY_B64)
_CHUNK_SIZE = 1024 * 1024
_BLOCK = 16
[docs]
def decrypt_security_token(security_token: str) -> tuple[bytes, bytes]:
"""Decode a bento ``keyId`` into ``(aes_key, nonce)``.
Raises ``ValueError`` on malformed tokens.
"""
if not security_token:
raise ValueError("empty security token")
try:
raw = base64.b64decode(security_token)
except (ValueError, TypeError) as exc:
raise ValueError(f"security token is not base64: {exc}") from exc
if len(raw) < 32:
raise ValueError(f"security token too short ({len(raw)} bytes)")
iv = raw[:16]
encrypted = raw[16:]
# The ciphertext is a multiple of the block size; pad defensively.
if len(encrypted) % _BLOCK:
encrypted += b"\x00" * (_BLOCK - len(encrypted) % _BLOCK)
decryptor = AES.new(_MASTER_KEY, AES.MODE_CBC, iv)
decrypted = decryptor.decrypt(encrypted)
if len(decrypted) < 24:
raise ValueError("decrypted security token too short")
return decrypted[:16], decrypted[16:24]
def _cipher_for(key: bytes, nonce: bytes, counter_start: int) -> Any:
# The CTR cipher wrapper type is pycryptodome-internal; callers only
# use ``.decrypt(bytes) -> bytes``.
counter = Counter.new(64, prefix=nonce, initial_value=counter_start)
return AES.new(key, AES.MODE_CTR, counter=counter)
[docs]
class StreamDecryptor:
"""Incremental AES-CTR decryption for a running media stream.
The counter block is ``nonce || 8-byte big-endian counter``, so the
file offset maps directly to the counter start: chunks may be any
size (they are decrypted in place), and the state is just the byte
offset. This is what :mod:`tidalpy.streaming` uses to hand plaintext
bytes to a host without ever touching the disk.
"""
[docs]
def __init__(self, security_token: str) -> None:
key, nonce = decrypt_security_token(security_token)
if len(key) != _BLOCK:
raise ValueError(f"AES key must be {_BLOCK} bytes, got {len(key)}")
if len(nonce) != _BLOCK // 2:
raise ValueError(f"nonce must be {_BLOCK // 2} bytes, got {len(nonce)}")
self._key = key
self._nonce = nonce
self._offset = 0
[docs]
def decrypt(self, data: bytes) -> bytes:
if not data:
return data
cipher = _cipher_for(self._key, self._nonce, self._offset // _BLOCK)
out: bytes = cipher.decrypt(data)
self._offset += len(data)
return out
[docs]
def stream_decrypt(
src: str | Path,
dst: str | Path,
key: bytes,
nonce: bytes,
chunk_size: int = _CHUNK_SIZE,
should_stop: Callable[[], bool] | None = None,
) -> int:
"""Decrypt ``src`` into ``dst`` in bounded chunks; returns bytes written.
``should_stop`` is called before every chunk; when it returns True a
:class:`~tidalpy.errors.CancelledError` is raised (``dst`` is left
incomplete on purpose — callers manage cleanup).
"""
from .errors import CancelledError
written = 0
with open(src, "rb") as source, open(dst, "wb") as output:
while True:
if should_stop is not None and should_stop():
raise CancelledError("decryption cancelled")
chunk = source.read(chunk_size)
if not chunk:
break
cipher = _cipher_for(key, nonce, written // _BLOCK)
output.write(cipher.decrypt(chunk))
written += len(chunk)
return written
[docs]
def decrypt_in_place(path: str | Path, security_token: str) -> Path:
"""Decrypt ``path`` (encrypted) to ``path`` (plain), atomically.
Returns the (unchanged) path. The original bytes are replaced via a
temporary file so an interruption never leaves a half-decrypted
file at the final name.
"""
key, nonce = decrypt_security_token(security_token)
target = Path(path)
tmp = target.with_name(target.name + ".dec.tmp")
try:
stream_decrypt(target, tmp, key, nonce)
except BaseException:
# Never leave a half-decrypted orphan next to the original.
tmp.unlink(missing_ok=True)
raise
# Preserve the original file's permissions on the replacement.
try:
mode = target.stat().st_mode
except OSError:
mode = None
tmp.replace(target)
if mode is not None:
try:
target.chmod(mode)
except OSError:
pass
logger.debug("decrypted %s in place", target)
return target