fix(import): read an HLS export that names no title language or master key

Probe each media playlist for its KIDs and refuse a manifest unshackle cannot read.

docs/guide/downloading.md

@@ -843,6 +843,10 @@ The [mediaexport](https://github.com/unshackle-dl/mediaexport) package specifies
     service. If a service never sets it, importing that export fails with a message naming
     the service. Neither end guesses a language for you, so the fix belongs in the service.
 
+    An HLS master playlist gives no language for a video track. When the export gives no
+    title language, the video track takes the language of the default audio track. When
+    the master playlist gives no audio language either, the import stops with a message.
+
 ## Metadata and tagging
 
 unshackle looks up metadata automatically, but you can override the identifiers used for

tests/core/test_import_foreign_hls.py

@@ -0,0 +1,144 @@
+"""An HLS export from another tool: only a master playlist URL, title keys, and no track rows.
+
+The importer parses the master playlist for the tracks. Such a file can have no title
+language, and its master playlist can name no key, so each track's own media playlist is
+the only place that names its KIDs.
+"""
+
+from __future__ import annotations
+
+import base64
+import json
+import struct
+from pathlib import Path
+from types import SimpleNamespace
+from typing import Any, Optional, cast
+from unittest.mock import patch
+from uuid import UUID
+
+import click
+import pytest
+
+from unshackle.core.import_service import ImportService
+from unshackle.core.manifests import HLS
+from unshackle.core.tracks import Video
+
+MASTER_URL = "https://example.invalid/hls/playlist.m3u8?t=1"
+VIDEO_KID = UUID("00000000-499d-c8c7-6332-202020202020")
+AUDIO_KID = UUID("00000000-499d-c8c7-6336-202020202020")
+OTHER_KID = UUID("11111111-2222-3333-4444-555555555555")
+KEYS = {VIDEO_KID: "0000000000000000000000000000000a", AUDIO_KID: "0000000000000000000000000000000b"}
+KEYS[OTHER_KID] = "0000000000000000000000000000000c"
+
+
+def playready_object(kid: UUID) -> str:
+    """A base64 PlayReady object whose one WRM header names ``kid``."""
+    value = base64.b64encode(kid.bytes_le).decode()
+    xml = (
+        '<WRMHEADER xmlns="http://schemas.microsoft.com/DRM/2007/03/PlayReadyHeader" version="4.3.0.0"><DATA>'
+        f'<PROTECTINFO><KIDS><KID ALGID="AESCBC" VALUE="{value}"></KID></KIDS></PROTECTINFO></DATA></WRMHEADER>'
+    ).encode("utf-16-le")
+    record = struct.pack("<HH", 1, len(xml)) + xml
+    return base64.b64encode(struct.pack("<IH", len(record) + 6, 1) + record).decode()
+
+
+def media_playlist(kid: Optional[UUID]) -> str:
+    key = ""
+    if kid:
+        key = (
+            '#EXT-X-KEY:METHOD=SAMPLE-AES,KEYFORMAT="com.microsoft.playready",KEYFORMATVERSIONS="1",'
+            f'URI="data:text/plain;charset=UTF-16;base64,{playready_object(kid)}"\n'
+        )
+    return (
+        "#EXTM3U\n#EXT-X-TARGETDURATION:6\n#EXT-X-PLAYLIST-TYPE:VOD\n"
+        f'#EXT-X-MAP:URI="init.mp4"\n{key}#EXTINF:6.0,\nseg0.m4s\n#EXT-X-ENDLIST\n'
+    )
+
+
+def master(audio_language: str = "en") -> str:
+    lang = f'LANGUAGE="{audio_language}",' if audio_language else ""
+    return (
+        "#EXTM3U\n"
+        f'#EXT-X-MEDIA:TYPE=AUDIO,GROUP-ID="aud",{lang}NAME="Main",DEFAULT=YES,URI="audio.m3u8"\n'
+        '#EXT-X-MEDIA:TYPE=SUBTITLES,GROUP-ID="sub",LANGUAGE="en",NAME="English",URI="sub.m3u8"\n'
+        '#EXT-X-STREAM-INF:BANDWIDTH=6000000,RESOLUTION=1920x1080,CODECS="avc1.640028,ec-3",'
+        'AUDIO="aud",SUBTITLES="sub"\nvideo.m3u8\n'
+    )
+
+
+def write_export(tmp_path: Path, language: str = "", manifest_url: str = MASTER_URL) -> Path:
+    """An export in the other tool's own format, as that tool writes it."""
+    title: dict[str, Any] = {"id": "ep-1", "kind": "episode", "name": "Example Show", "season": 1, "episode": 1}
+    if language:
+        title["language"] = language
+    doc = {
+        "kind": "unidl-export",
+        "version": 1,
+        "service": "example",
+        "titles": [
+            {
+                "save_name": "Example.Show.S01E01",
+                "title": title,
+                "keys": [f"{kid.hex}:{key}" for kid, key in KEYS.items()],
+                "manifest_url": manifest_url,
+                "headers": {"User-Agent": "Example/1.0"},
+                "drm": {"system": "playready", "wrm_header": "<WRMHEADER/>"},
+            }
+        ],
+    }
+    path = tmp_path / "export.json"
+    path.write_text(json.dumps(doc), encoding="utf8")
+    return path
+
+
+def import_title(export: Path, master_text: str, playlists: dict[str, Optional[UUID]]) -> tuple[ImportService, Any]:
+    ctx = cast(click.Context, SimpleNamespace(parent=None, params={}))
+    svc = ImportService(ctx, "EXAMPLE", "ep-1", str(export))
+
+    def get(url: str, **_: Any) -> SimpleNamespace:
+        name = url.split("?")[0].rsplit("/", 1)[-1]
+        return SimpleNamespace(text=media_playlist(playlists.get(name)), raise_for_status=lambda: None)
+
+    setattr(svc.session, "get", get)
+    episode = next(iter(svc.get_titles()))
+    with patch.object(HLS, "from_url", side_effect=lambda url, session=None, **kw: HLS.from_text(master_text, url)):
+        episode.tracks = svc.get_tracks(episode)
+    return svc, episode
+
+
+def test_no_title_language_takes_the_default_audio_language(tmp_path: Path) -> None:
+    _, episode = import_title(write_export(tmp_path), master(), {})
+
+    assert [str(t.language) for t in episode.tracks.videos] == ["en"]
+
+
+def test_no_language_anywhere_names_the_missing_language(tmp_path: Path) -> None:
+    with pytest.raises(click.ClickException, match="No language"):
+        import_title(write_export(tmp_path), master(audio_language=""), {})
+
+
+def test_each_hls_track_gets_the_keys_its_media_playlist_names(tmp_path: Path) -> None:
+    """The master playlist names no key, so only the media playlist tells the tracks apart."""
+    svc, episode = import_title(
+        write_export(tmp_path, "en"), master(), {"video.m3u8": VIDEO_KID, "audio.m3u8": AUDIO_KID}
+    )
+    svc.resolve_server_keys(episode)
+
+    video, audio = episode.tracks.videos[0], episode.tracks.audio[0]
+    assert video.drm and video.drm[0].content_keys == {VIDEO_KID: KEYS[VIDEO_KID]}
+    assert audio.drm and audio.drm[0].content_keys == {AUDIO_KID: KEYS[AUDIO_KID]}
+    assert all(not t.drm for t in episode.tracks.subtitles)
+
+
+def test_an_hls_track_whose_playlist_names_no_key_stays_clear(tmp_path: Path) -> None:
+    svc, episode = import_title(write_export(tmp_path, "en"), master(), {"video.m3u8": VIDEO_KID})
+    svc.resolve_server_keys(episode)
+
+    assert not episode.tracks.audio[0].drm
+    assert isinstance(episode.tracks.videos[0], Video) and episode.tracks.videos[0].drm
+
+
+def test_a_manifest_that_unshackle_cannot_read_stops_the_import(tmp_path: Path) -> None:
+    """A manifest held only in the other tool's own form must not import as a title with no tracks."""
+    with pytest.raises(click.ClickException, match="cannot read"):
+        import_title(write_export(tmp_path, "en", manifest_url="x-unidl:json_manifest"), master(), {})

unshackle/core/import_service.py

@@ -8,6 +8,7 @@ from typing import Any, Callable, Optional, TypeVar, Union
 from uuid import UUID
 
 import click
+import m3u8
 import mediaexport
 import requests
 from langcodes import tag_is_valid
@@ -42,6 +43,43 @@ def fallback_language(language: Any) -> Optional[Any]:
     return language
 
 
+def hls_audio_language(manifest: HLS) -> Optional[str]:
+    """The language of the default audio rendition, else of the first audio rendition, else None.
+
+    An HLS master playlist gives no language for a video variant, so the parser needs a
+    fallback. An export from another tool can give no title language.
+    """
+    audio = [m for m in manifest.manifest.media if m.type == "AUDIO" and fallback_language(m.language)]
+    audio.sort(key=lambda m: m.default != "YES")
+    return str(audio[0].language) if audio else None
+
+
+def hls_playlist_kids(track: Any, session: Optional[requests.Session] = None) -> Optional[set[UUID]]:
+    """Return the KIDs that the Widevine and PlayReady keys of the track's media playlist name.
+
+    Return None when the playlist has no key or only AES-128 keys: the download gets an AES-128
+    key from its URI, so such a track takes no exported content key. An empty set means that
+    the track is encrypted but names no KID that unshackle can read.
+    """
+    try:
+        text = (session or requests.Session()).get(str(track.url)).text
+        keys = [k for k in m3u8.loads(text, uri=str(track.url)).keys if k is not None and k.method != "NONE"]
+    except Exception as e:
+        log.warning(f"Cannot read the media playlist of {track.id} for its KIDs: {e!r}")
+        return None
+    kids: set[UUID] = set()
+    for key in keys:
+        if key.method == "AES-128":
+            continue
+        try:
+            kids.update(real_kids(HLS.get_drm(key)))
+        except Exception as e:
+            log.debug(f"Skipping HLS key {key.keyformat or key.method} of {track.id}: {e!r}")
+    if kids or any(k.method != "AES-128" for k in keys):
+        return kids
+    return None
+
+
 def resolve_import_manifest_data(
     tracks: Tracks,
     manifest_type: Optional[str],
@@ -316,8 +354,16 @@ class ImportService:
                     f"Failed to fetch the {manifest_type} manifest for '{title}'. "
                     f"The manifest URL may have expired since export. ({e})"
                 )
+            language = fallback_language(title.language)
+            if language is None and isinstance(manifest, HLS):
+                language = hls_audio_language(manifest)
+                if language is None:
+                    raise click.ClickException(
+                        f"No language for '{title}': the export gives no title language and the HLS "
+                        "manifest gives no audio language."
+                    )
             try:
-                parsed = manifest.to_tracks(language=fallback_language(title.language))
+                parsed = manifest.to_tracks(language=language)
             except ValueError as e:
                 if "Language information could not be derived" in str(e):
                     raise click.ClickException(
@@ -365,6 +411,11 @@ class ImportService:
                 track.drm = drm
             tracks.add(track, warn_only=True)
 
+        if manifest_url and parser is None and not [*tracks.videos, *tracks.audio, *tracks.subtitles]:
+            raise click.ClickException(
+                f"Cannot import '{title}': unshackle cannot read its manifest {str(manifest_url)[:80]!r}."
+            )
+
         resolve_import_manifest_data(
             tracks,
             manifest_type,
@@ -608,7 +659,8 @@ class ImportService:
         reads only ``track.drm``. A rung that misses content key injection here downloads
         encrypted and muxes without error. A DASH manifest with no ContentProtection can still
         describe an encrypted track, with the PSSH only in the init segment, so that case
-        probes the init segment with `session`.
+        probes the init segment with `session`. An HLS track with no DRM gets its KIDs from the
+        EXT-X-KEY tags of its media playlist, because a master playlist can name no KID.
         """
         if track.drm:
             return {kid for drm in track.drm for kid in real_kids(drm)}
@@ -642,6 +694,8 @@ class ImportService:
             manifest = ism.get("manifest")
             if manifest is not None and manifest.findall(".//ProtectionHeader"):
                 return set()
+        if track.descriptor == Track.Descriptor.HLS and track.url:
+            return hls_playlist_kids(track, session)
         return None
 
     def exported_drm_system(self) -> str: