feat(drm): add blowfish stripe cipher and mp3 audio codec

BlowfishStripe decrypts on the client after download, so a --remote client gets clear audio.

tests/core/test_blowfish_stripe.py

@@ -0,0 +1,112 @@
+"""Tests for the Blowfish CBC stripe cipher (``unshackle.core.drm.blowfish_stripe``)."""
+
+from __future__ import annotations
+
+import os
+from collections.abc import Iterator
+from functools import partial
+from pathlib import Path
+from typing import Any
+
+import pytest
+import requests
+from Cryptodome.Cipher import Blowfish
+
+from unshackle.core.api.handlers import serialize_audio_track, serialize_drm
+from unshackle.core.config import config
+from unshackle.core.constants import DOWNLOAD_CANCELLED
+from unshackle.core.drm import BlowfishStripe
+from unshackle.core.remote_service import build_tracks, reconstruct_drm
+from unshackle.core.tracks import Audio, Track
+from unshackle.core.tracks import track as track_module
+
+KEY = bytes.fromhex("00112233445566778899aabbccddeeff")
+CHUNK = BlowfishStripe.CHUNK
+
+
+def stripe_encrypt(data: bytes, key: bytes = KEY) -> bytes:
+    """Reference encryption: every third full chunk, each with a fresh CBC run."""
+    out = bytearray(data)
+    for offset in range(0, len(data), CHUNK * 3):
+        chunk = data[offset : offset + CHUNK]
+        if len(chunk) == CHUNK:
+            out[offset : offset + CHUNK] = Blowfish.new(key, Blowfish.MODE_CBC, iv=BlowfishStripe.IV).encrypt(chunk)
+    return bytes(out)
+
+
+def test_decrypt_round_trip(tmp_path: Path) -> None:
+    plain = os.urandom(CHUNK * 10 + 123)
+    path = tmp_path / "track.flac"
+    path.write_bytes(stripe_encrypt(plain))
+    assert path.read_bytes() != plain
+
+    BlowfishStripe(KEY).decrypt(path)
+
+    assert path.read_bytes() == plain
+    assert list(tmp_path.iterdir()) == [path]
+
+
+def test_short_trailing_chunk_untouched(tmp_path: Path) -> None:
+    # the last stripe starts at CHUNK * 3 and holds only 100 bytes, so it stays clear
+    plain = os.urandom(CHUNK * 3 + 100)
+    encrypted = stripe_encrypt(plain)
+    assert encrypted[CHUNK * 3 :] == plain[CHUNK * 3 :]
+    path = tmp_path / "track.mp3"
+    path.write_bytes(encrypted)
+
+    BlowfishStripe(KEY).decrypt(path)
+
+    assert path.read_bytes() == plain
+
+
+def test_serialize_round_trip_keeps_key() -> None:
+    serialized = serialize_drm([BlowfishStripe(KEY)])
+    assert serialized == [{"type": "blowfishstripe", "key": KEY.hex()}]
+
+    rebuilt = reconstruct_drm(serialized)
+
+    assert len(rebuilt) == 1
+    assert isinstance(rebuilt[0], BlowfishStripe)
+    assert rebuilt[0].key == KEY
+
+
+def test_remote_tracks_keep_the_stripe_drm() -> None:
+    track = Audio(
+        id_="a1", url="https://example.test/a1", language="en", codec=Audio.Codec.FLAC, drm=[BlowfishStripe(KEY)]
+    )
+
+    rebuilt = build_tracks({"audio": [serialize_audio_track(track, include_url=True)]})
+
+    assert isinstance(rebuilt.audio[0].drm[0], BlowfishStripe)
+    assert rebuilt.audio[0].drm[0].key == KEY
+
+
+def test_url_download_decrypts_once_without_a_kid_probe(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
+    monkeypatch.setattr(config.directories, "temp", tmp_path)
+    monkeypatch.setattr(Track, "get_key_id", lambda *a, **k: pytest.fail("a stripe-cipher file has no KID to probe"))
+    monkeypatch.setattr(
+        track_module, "assert_fragments_decrypted", lambda path: pytest.fail("a stripe-cipher file has no fragments")
+    )
+    plain = os.urandom(CHUNK * 7 + 5)
+
+    def downloader(**kwargs: Any) -> Iterator[dict[str, Any]]:
+        (Path(kwargs["output_dir"]) / kwargs["filename"]).write_bytes(stripe_encrypt(plain))
+        yield {"file_downloaded": Path(kwargs["output_dir"]) / kwargs["filename"]}
+
+    track = Audio(
+        id_="a1",
+        url="https://example.test/a1.flac",
+        language="en",
+        codec=Audio.Codec.FLAC,
+        drm=[BlowfishStripe(KEY)],
+        downloader=downloader,
+    )
+    try:
+        track.download(
+            session=requests.Session(), prepare_drm=partial(lambda *a, **kw: None), progress=partial(lambda **kw: None)
+        )
+    finally:
+        DOWNLOAD_CANCELLED.clear()
+
+    assert track.path is not None and track.path.read_bytes() == plain
+    assert track.drm is None

tests/core/test_music_container.py

@@ -67,6 +67,7 @@ def header_length(path: Path) -> float:
         (Audio.Codec.ALAC, "alac", ".m4a"),
         (Audio.Codec.FLAC, "flac", ".flac"),
         (Audio.Codec.OPUS, "libopus", ".opus"),
+        (Audio.Codec.MP3, "libmp3lame", ".mp3"),
     ],
 )
 def test_remux_states_the_full_length_in_the_codec_container(

tests/remote/unit/test_server_cdm_key_verification.py

@@ -15,6 +15,7 @@ import pytest
 from unshackle.core.api import handlers
 from unshackle.core.api.errors import APIError
 from unshackle.core.api.session_store import SessionEntry
+from unshackle.core.drm import BlowfishStripe
 from unshackle.core.remote_service import RemoteService, ServerVault
 
 pytestmark = pytest.mark.unit
@@ -245,3 +246,30 @@ def test_a_cdm_licence_unflags_a_pair_a_client_reported(monkeypatch, tmp_path):
     handlers.cache_to_vaults({KID.hex: "ab" * 16}, "SVC")
     assert not local.is_bad_key(KID, "ab" * 16)
     assert local.get_key(KID, "SVC") == "ab" * 16
+
+
+async def test_a_stripe_cipher_track_is_clear_and_never_licensed(monkeypatch):
+    """A track whose only DRM needs no licence goes to clear_tracks and keeps its DRM for the client."""
+    track = SimpleNamespace(id="aud", drm=[BlowfishStripe(bytes(16))])
+    session = SessionEntry(session_id="sess", service_tag="SVC", service_instance=SimpleNamespace())
+    session.tracks = {"aud": track}
+
+    async def validated(sid, req):
+        return session
+
+    monkeypatch.setattr(handlers, "get_validated_session", validated)
+    monkeypatch.setattr(handlers, "server_cdm_allowed", lambda request, tag: True)
+    monkeypatch.setattr(handlers, "ensure_track_drm", lambda track, session=None, init_data=None: None)
+    monkeypatch.setattr(handlers, "find_title_for_track", lambda tid, session: SimpleNamespace())
+    monkeypatch.setattr(handlers, "detect_cdm_type_for_service", lambda tag, cfg: "widevine")
+    monkeypatch.setattr(handlers, "resolve_device_name", lambda *a, **k: "dev")
+    monkeypatch.setattr(handlers, "fetch_init_segment", lambda track, sess: None)
+    monkeypatch.setattr(handlers, "handle_single_server_cdm", lambda *a, **k: pytest.fail("licensed a clear track"))
+    monkeypatch.setattr(handlers.config, "serve", {"users": {}}, raising=False)
+
+    resp = await handlers.session_license_handler(
+        {"mode": "server_cdm", "track_ids": ["aud"], "drm_type": "widevine"}, "sess", None
+    )
+
+    assert json.loads(resp.body)["clear_tracks"] == ["aud"]
+    assert isinstance(track.drm[0], BlowfishStripe)