Compare commits

...
6 Commits
Author SHA1 Message Date
imSp4rky 7f91a1d36b feat(drm): license a track only when its own KID has no key
A vault key for the track's KID now stands in for the licence on the client and the server (PlayReady included). decrypt_verified licenses any other header KID the file uses.
2026-11-12 18:03:16 -07:00
imSp4rky 16fbf5d17a fix(remote): report each handled remote error once, from the code that catches it
A caught RemoteError now logs at debug under quiet(); a recovered licence prints one warning, only when the track's own key is held.
2026-11-12 17:15:40 -07:00
imSp4rky dc18b23b91 fix(remote): keep vault keys when a relayed licence is refused
proxy_license now raises ClickException instead of RemoteError (a SystemExit), so dl falls back to the track's vault key; licence messages name the track ("video 360p DV (592464997)").
2026-11-12 16:14:34 -07:00
imSp4rky f873ea1fb8 refactor(drm): share one CENC decrypt path across Widevine, PlayReady and ClearKey 2026-11-12 13:38:40 -07:00
imSp4rky b48da2a16a fix(mux): strip encoding dates from MP4Box and shaka-packager output
MP4Box gets -for-test (non-MP4 inputs keep their tkhd flags) and shaka decrypts get --use_fake_clock_for_muxer, so MP4 and --no-mux files carry no download time.
2026-11-12 13:05:14 -07:00
imSp4rky e29a523907 feat(tracks): download and mux stereoscopic 3D (MV-HEVC) video to MP4
Mark 3D from REQ-VIDEO-LAYOUT, DV profile 20 codecs or an lhvC box; mux with MP4Box, decrypt with mp4decrypt, skip FFmpeg repack, sample cut and hybrid; add {3d} and env MP4Box.
2026-11-12 04:47:18 -07:00
50 changed files with 1377 additions and 641 deletions
+1 -1
View File
@@ -339,7 +339,7 @@ When you give both `season` and `episode`, the server combines them into a `"{se
"bitrate": 8000, "bitrate_bps": 8000000, "width": 3840, "height": 2160,
"resolution": "2160p", "fps": "24", "range": "HDR10",
"range_display": "HDR", "scan_type": null, "closed_captions": [],
"dv_compatible_bitstream": false, "hdr10plus": false, "hdr10_base": false, "language": "en",
"dv_compatible_bitstream": false, "hdr10plus": false, "hdr10_base": false, "stereo": false, "language": "en",
"is_original_lang": true,
"name": null, "track_name": "English", "needs_repack": false, "edition": [], "note": null, "data": null,
"drm": [], "descriptor": "DASH"
+1
View File
@@ -454,6 +454,7 @@ Two DV mechanisms sit on top of the track model. **DV fixup** re-exposes DV sign
HLS and DASH streams that carry a valid DV RPU but advertise plain HEVC, with DV only in the
supplemental codecs (`dv_compatible_bitstream`). `--list` shows such a track as `HDR10+DV` or `HDR10P+DV`.
A service sets `hdr10_base` on a DV track that has an HDR10 base layer (DV profile 8.1), and `--list` shows it as `DV+HDR10`.
The HLS and DASH parsers set `stereo` on a stereoscopic 3D (MV-HEVC) track: an HLS `REQ-VIDEO-LAYOUT` that names `CH-STEREO`, or a DV profile 20 codec. `--list` shows it as `3D`.
**Hybrid** processing (`Video.Range.HYBRID`) selects an HDR10(+) base and a DV donor,
extracts the RPU with `dovi_tool`, and injects it onto the base layer, to make a single
track that carries both HDR10 metadata and the DV RPU. Both are automatic once the
+4 -4
View File
@@ -268,7 +268,7 @@ download. For example, `dovi_tool` only matters if you download Dolby Vision vid
| Tool | Binary looked up | Why it is needed |
| --- | --- | --- |
| Bento4 (`mp4decrypt`) | `mp4decrypt` | Alternative DRM decryptor to shaka-packager |
| MP4Box (GPAC) | `MP4Box` | Muxing DTS:X Profile 2 (DTS-UHD) and MPEG-H audio |
| MP4Box (GPAC) | `MP4Box` | Muxing DTS:X Profile 2 (DTS-UHD) and MPEG-H audio, and 3D video |
| dovi_tool | `dovi_tool` | Dolby Vision metadata handling |
| HDR10Plus_tool | `hdr10plus_tool` | HDR10+ metadata handling |
| SubtitleEdit | `seconv`, `SubtitleEdit` | Subtitle conversion (the `SeConv` CLI from 5.x) |
@@ -280,9 +280,9 @@ download. For example, `dovi_tool` only matters if you download Dolby Vision vid
| Docker | `docker` | Gluetun VPN proxy support |
| Git | `git` | Fetching and updating remote service repositories |
!!! info "MP4Box is only used for DTS:X Profile 2 and MPEG-H"
Titles with DTS:X Profile 2 (DTS-UHD) or MPEG-H audio mux to MP4; everything else muxes to MKV
with `mkvmerge`. Install with `apt install gpac`.
!!! info "MP4Box is only used for DTS:X Profile 2, MPEG-H and 3D video"
Titles with DTS:X Profile 2 (DTS-UHD) or MPEG-H audio, or a stereoscopic 3D video track, mux to
MP4; everything else muxes to MKV with `mkvmerge`. Install with `apt install gpac`.
!!! tip "SubtitleEdit specifics"
unshackle looks for `seconv` first, the batch CLI shipped with
+4
View File
@@ -164,6 +164,10 @@ unshackle dl -q 2160 -r HDR10,DV EXAMPLE '...'
The output file of a track with a Dolby Vision profile 8 RPU plays as Dolby Vision on a DV screen.
`--list` adds `3D` to a stereoscopic 3D (MV-HEVC) track, for example `[H.265, DV, 3D]`. When a title
has a 2D and a 3D track at the same resolution, unshackle selects the 2D track. A 3D title muxes to
MP4, see [3D output](output-and-naming.md#3d-output).
!!! warning "HYBRID requires dovi_tool"
`-r HYBRID` makes a single hybrid track by injecting the Dolby Vision RPU
metadata onto an HDR10/HDR10+ base layer with **dovi_tool**.
+22 -3
View File
@@ -15,7 +15,7 @@ When a download finishes, unshackle builds the final path from three pieces:
- **Output directory**: `directories.downloads` from your config, or whatever you pass to `-o/--output` on the command line for a single run.
- **Folder template**: an optional per-title subfolder. unshackle **always** puts TV episodes and songs in a folder. Movies only get one if you define a `movies` folder template, or a single-string `folder` template that applies to all title kinds (see [Folder templates](#folder-templates)).
- **Filename template**: the `output_template` for the title kind (movie, series, or song).
- **Extension**: chosen by the muxer for a movie or an episode: `.mkv` for video, `.mka` for audio-only, `.mks` for subtitle-only, `.mp4` for [DTS:X Profile 2 and MPEG-H audio](#dtsx-profile-2-and-mpeg-h-output). unshackle does not mux a song. It remuxes a song into the container of its codec, such as `.flac`, `.opus`, or `.m4a`. Read [Music output files](#music-output-files).
- **Extension**: chosen by the muxer for a movie or an episode: `.mkv` for video, `.mka` for audio-only, `.mks` for subtitle-only, `.mp4` for [DTS:X Profile 2 and MPEG-H audio](#dtsx-profile-2-and-mpeg-h-output) and [3D video](#3d-output). unshackle does not mux a song. It remuxes a song into the container of its codec, such as `.flac`, `.opus`, or `.m4a`. Read [Music output files](#music-output-files).
!!! example "A typical episode path"
```text
@@ -110,6 +110,7 @@ Every variable below is valid in both output and folder templates. unshackle tak
| `video` | Video codec | `H.264`, `H.265` |
| `hdr` | Dynamic-range label | `HDR`, `HDR10P`, `DV`, `DV.HDR`, `DV.HDR10P`, `HLG` |
| `hfr` | High frame rate marker (frame rate > 30) | `HFR` |
| `3d` | Stereoscopic 3D marker (MV-HEVC video) | `3D` |
| `edition` | Edition label from the track, if any | `Directors Cut` |
| `tag` | Release-group tag (from `config.tag` or `--tag`) | `TAG` |
| `repack` | `REPACK` when the `--repack` flag is used | `REPACK` |
@@ -429,6 +430,23 @@ An MP4 title loses:
- SubStation Alpha (`.ass`) subtitles
- container tags (`mkvpropedit`)
### 3D output
A stereoscopic 3D (MV-HEVC) video track also muxes to MP4 with MP4Box, because Matroska has no
mapping for its second view and `mkvmerge` drops it. There is no 3D Matroska output. unshackle
finds a 3D track from the manifest, or from a second video layer in the downloaded file. The MP4
title loses the same items as above. For a 3D title, unshackle also:
- decrypts the track with mp4decrypt, whatever [`decryption`](../reference/configuration/drm.md#decryption) selects
- writes QuickTime chapters
- keeps the whole downloaded segments of a `--sample` run, because the MP4Box cut removes the 3D signalling
- does not repack the track or rewrite its colour metadata with FFmpeg
- writes no `--include-sample` sample, because `mkvmerge` cuts only Matroska files
- never uses the track in a `HYBRID` merge
A Dolby Vision profile 20 track has no HDR10 base layer, so a player without Dolby Vision shows
wrong colours.
You configure muxing (the combination of video, audio, subtitle, chapter and attachment tracks into a single Matroska file with `mkvmerge`) under the `muxing` config key. unshackle muxes only movies and episodes. Read [Music output files](#music-output-files) for what a music download does instead.
```yaml title="unshackle.yaml"
@@ -464,17 +482,18 @@ Set `default_subtitle_type` to give the subtitle default to a different type. `d
A few muxing behaviors are automatic and not configurable:
- `--no-date` is always passed for privacy (no timestamps embedded).
- The output has no encoding date: mkvmerge gets `--no-date` and MP4Box gets `-for-test`.
- Descriptive audio gets the visually-impaired flag. SDH subtitles get the hearing-impaired flag. Forced subtitles get the forced flag.
- unshackle writes the chapters from the title's chapter list (see [Chapters](#chapters-naming) below).
The `--no-mux` flag skips muxing entirely and writes the individual track files, each with a track-type suffix, into the same output/folder structure.
The `--no-mux` flag skips muxing entirely and writes the individual track files, each with a track-type suffix, into the same output/folder structure. A track that shaka-packager decrypts has its date set to 1970-01-01; other tracks keep the date of the source file.
## Sample files
[`--sample`](cli-reference.md#output-muxing-files) and `--include-sample` write a sample into a
`Sample` folder inside the output folder of the title. The sample keeps the filename of the
title, with a `SAMPLE` tag at the end. The tag uses the separator of the output template.
`--include-sample` cuts the sample with `mkvmerge`, so a title that muxes to MP4 gets no sample.
```text title="A movie with --include-sample"
Some.Movie.2026.1080p.EXAMPLE.WEB-DL.H.264-TAG/
+2
View File
@@ -25,6 +25,8 @@ A vault stores every content key as a **KID → KEY** pair, grouped by **service
When unshackle needs a content key during a download it asks your vaults **in the order they appear in your config**. It returns the first non-null key it finds and stops searching. If a vault raises a permission error or is unreachable, unshackle skips it and continues the search with the next one.
A track needs no licence when a vault holds the content key of the KID that encrypts it. The DRM header of the track can name more KIDs, for example those of other tracks. Before the decrypt, unshackle reads the downloaded file, and licenses each of those KIDs that the file uses and no vault holds. When unshackle cannot read which KIDs the file uses, it licenses all of them.
When unshackle recovers new keys from a license, it pushes them to **all** configured vaults, so every vault gradually converges on the same set of keys. Two rules apply to pushing:
- unshackle skips keys that already exist in a vault (it never overwrites or deletes existing data).
+4
View File
@@ -187,6 +187,10 @@ service tag case-insensitively and uses the `default` entry as the fallback.
unshackle decrypts HLS AES-128 ClearKey in-process and ignores this setting.
unshackle decrypts a stereoscopic 3D (MV-HEVC) track with mp4decrypt whatever this setting says.
shaka-packager refuses its Dolby Vision configuration (version 3), and before v3.5.0 it also drops
the boxes of the second view.
mp4decrypt finds a content key by the KID in the first `sinf` box of the track. When that box
has no CENC `tenc` (for example, a PIFF `sinf` before the CENC one), unshackle also gives
mp4decrypt the content key of that KID by track ID. When mp4decrypt copies a track through
+2 -2
View File
@@ -40,7 +40,7 @@ folder-kind keys.
`absolute`, `date`, `quality`, `resolution`, `source`, `tag`, `track_number`, `artist`,
`album_artist`, `album`, `disc`, `track_total`, `disc_total`, `release_type`, `genre`,
`explicit`, `isrc`, `upc`, `label`, `audio`, `audio_channels`, `audio_full`, `atmos`,
`dual`, `multi`, `dubbed`, `video`, `hdr`, `hfr`, `edition`, `repack`, `lang_tag`,
`dual`, `multi`, `dubbed`, `video`, `hdr`, `hfr`, `3d`, `edition`, `repack`, `lang_tag`,
`title_type`.
!!! warning "`{part?}` is already inside `{season_episode}`"
@@ -129,7 +129,7 @@ Rule semantics:
- A value can be a scalar or a list. A list matches if **any** entry matches.
- Comparison is case-insensitive string equality.
- The keys of `when` are the filename template variables for the release itself:
`title_type`, `quality`, `resolution`, `source`, `video`, `hdr`, `hfr`, `audio`,
`title_type`, `quality`, `resolution`, `source`, `video`, `hdr`, `hfr`, `3d`, `audio`,
`audio_channels`, `audio_full`, `atmos`, `dual`, `multi`, `dubbed`, `edition`, `repack`,
and `lang_tag`. Only these are available. unshackle adds the title's own fields (`title`,
`year`, `season`, `episode`, and so on) after it evaluates the rules, so a condition cannot
+2 -2
View File
@@ -113,7 +113,7 @@ values.
| `{year}` `{season}` `{episode}` `{episode_name}` | Season and episode are plain numbers (`1`, `5`) |
| `{part}` | Part number of a split episode as a plain number (`1` for `Part.1`). Empty when the episode has no part |
| `{season_episode}` `{absolute}` `{date}` | As in the filename template: `{season_episode}` keeps the part suffix (`S01E05.Part.1`) while `{episode}` drops it |
| `{quality}` `{resolution}` `{hdr}` `{hfr}` `{vcodec}` `{acodec}` `{edition}` `{tag}` `{lang_tag}` | As in the filename template, for this file |
| `{quality}` `{resolution}` `{hdr}` `{hfr}` `{3d}` `{vcodec}` `{acodec}` `{edition}` `{tag}` `{lang_tag}` | As in the filename template, for this file |
| `{tmdb}` `{imdb}` `{tvdb}` | The tagging IDs in use, otherwise empty. See [Tagging IDs](#tagging-ids) |
| `{error}` | Failure message. Empty on success |
@@ -132,7 +132,7 @@ Every variable from your `output_template` is available here as well, including
music has none of them.
- **The naming context renders it empty on purpose.** `{hdr}` is empty for an SDR
file, because SDR is the absence of an HDR tag in a filename rather than a tag that
reads `SDR`. `{edition}`, `{atmos}`, `{multi}`, `{dual}` and `{hfr}` behave the same
reads `SDR`. `{edition}`, `{atmos}`, `{multi}`, `{dual}`, `{hfr}` and `{3d}` behave the same
way. If your script needs the word `SDR`, derive it from an empty `{hdr}` yourself.
- The post-script is a `failure` one. A failed download has no output file to read metadata
from, so everything that comes from the output file is empty. See
+6 -4
View File
@@ -223,9 +223,10 @@ the remote client's `-q` when the remote session starts:
error. The server checks the values the job runs with, after it applies the service defaults
and the `serve:` download defaults.
The limit protects the server's device, not the content keys. For a Widevine track, a server vault
still supplies a content key at every height. A PlayReady track never reads the vault, so above
the limit it always goes to the client's device. A licence for a lower track can carry the
The limit protects the server's device, not the content keys. A server vault still supplies a
content key at every height: for a Widevine track, the key of the track's own KID or of every KID in
its header; for a PlayReady track, only the key of the track's own KID. A PlayReady track whose
own KID the server does not know always goes to the client's device above the limit. A licence for a lower track can carry the
content key of a higher track when the service shares one content key between them. With server
accounts enabled, a client that reports a stronger device than it holds makes the server's
account request those manifests; the licence server still refuses its challenge.
@@ -233,7 +234,8 @@ account request those manifests; the licence server still refuses its challenge.
`server_vault` lets a remote session that the client's own device licenses take content keys
from the server vault. Set it to `true` for every service, or to a list of service tags. It is
`false` unless the entry sets it, and `server_cdm` for a service implies it. The server looks up
every KID of the track in its vault and makes no live licence. A track that the vault does not
the track's own KID in its vault, or every KID of the track when it does not know that KID, and
makes no live licence. A track that the vault does not
hold goes back to the client's device. When the service runs the remote session on the server's
own device (a service option, such as a server-identity flag), the client's own vaults, and the
server vault with this grant, are the only source, and a missing content key stops the download.
+13
View File
@@ -476,3 +476,16 @@ def test_import_keeps_every_period_when_no_exported_period_is_left(
assert video.data["dash"]["filtered_period_ids"] == []
assert "keeping every period" in caplog.text
def test_a_3d_video_stays_3d_through_export_and_import(tmp_path: Path) -> None:
export = tmp_path / "export.json"
title = make_title()
title.tracks.videos[0].stereo = True
runner = make_dl()
for track in [*title.tracks.videos, *title.tracks.audio]:
runner.write_export(export, title, track)
assert [row.get("stereo") for row in entry(export)["tracks"]][0] is True
svc = ImportService(import_ctx(), "EXAMPLE", "movie-1", str(export))
movie = next(iter(svc.get_titles()))
assert svc.get_tracks(movie).videos[0].stereo
+1 -1
View File
@@ -856,7 +856,7 @@ def video(
codec: Optional[Video.Codec] = Video.Codec.HEVC,
native_hybrid: bool = False,
) -> Any:
return SimpleNamespace(id=id_, height=height, range=range_, codec=codec, native_hybrid=native_hybrid)
return SimpleNamespace(id=id_, height=height, range=range_, codec=codec, native_hybrid=native_hybrid, stereo=False)
def test_hybrid_fails_with_neither_a_base_nor_dv() -> None:
+5
View File
@@ -297,3 +297,8 @@ def test_a_file_row_can_name_its_codec_and_range_by_name(tmp_path: Path) -> None
service = file_export(tmp_path, [row])
video = service.get_tracks(next(iter(service.get_titles()))).videos[0]
assert (video.codec, video.range) == (Video.Codec.HEVC, Video.Range.HDR10P)
def test_a_file_row_marked_stereo_imports_as_a_3d_video(tmp_path: Path) -> None:
service = file_export(tmp_path, [{**FILE_ROWS[0], "stereo": True}])
assert service.get_tracks(next(iter(service.get_titles()))).videos[0].stereo
+56 -1
View File
@@ -9,6 +9,7 @@ from types import SimpleNamespace
from typing import Any, Callable, Optional
from uuid import UUID
import click
import pytest
from pywidevine.pssh import PSSH
from rich.table import Table
@@ -71,7 +72,7 @@ def widevine(*kids: UUID) -> Widevine:
return Widevine(pssh=PSSH.new(system_id=PSSH.SystemId.Widevine, key_ids=list(kids)))
def prepare(cmd: dl, drm: Widevine, track_kid: UUID, table: Table, **kwargs: Any) -> None:
def prepare(cmd: dl, drm: Widevine, track_kid: Optional[UUID], table: Table, **kwargs: Any) -> None:
cmd.prepare_drm(
drm,
track=SimpleNamespace(drm=[drm]),
@@ -197,3 +198,57 @@ def test_vaults_only_stops_at_the_first_kid_no_vault_holds(cmd: dl, licences: li
assert asked == [KID_A]
assert licences == []
def refuse_licence(self: Widevine, **_: Any) -> None:
raise click.ClickException("Licence request failed")
def test_a_refused_licence_without_the_track_key_reports_the_licence_error(
cmd: dl, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(Widevine, "get_content_keys", refuse_licence)
monkeypatch.setattr(dl, "LICENSE_KEY_CACHE", {KID_B: KEYS[KID_B]})
monkeypatch.setattr(dl, "DRM_LOCKS", {})
table = new_table()
with pytest.raises(click.ClickException, match="^Licence request failed$"):
prepare(cmd, widevine(KID_A, KID_B), KID_A, table)
assert not [r for r in caplog.records if r.levelname == "WARNING"]
tree = list(table.columns[0].cells)[-1]
assert "Licence request failed" in str(tree.children[-1].label)
def test_a_held_track_key_needs_no_licence(cmd: dl, licences: list[str], monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(dl, "LICENSE_KEY_CACHE", {KID_A: KEYS[KID_A]})
drm = widevine(KID_A, KID_B)
prepare(cmd, drm, KID_A, new_table())
assert licences == []
assert drm.content_keys == {KID_A: KEYS[KID_A]}
def test_force_licenses_a_held_track_key_again(cmd: dl, licences: list[str], monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(dl, "LICENSE_KEY_CACHE", {KID_A: KEYS[KID_A]})
prepare(cmd, widevine(KID_A, KID_B), KID_A, new_table(), force=True)
assert len(licences) == 1
def test_a_refused_licence_keeps_the_known_keys_when_the_track_kid_is_unknown(
cmd: dl, monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
monkeypatch.setattr(Widevine, "get_content_keys", refuse_licence)
monkeypatch.setattr(dl, "LICENSE_KEY_CACHE", {KID_A: KEYS[KID_A]})
monkeypatch.setattr(dl, "DRM_LOCKS", {})
drm = widevine(KID_A, KID_B)
prepare(cmd, drm, None, new_table())
assert drm.content_keys == {KID_A: KEYS[KID_A]}
assert [r.getMessage() for r in caplog.records if r.levelname == "WARNING"] == [
"Licence failed, using the 1 known key(s): Licence request failed"
]
+2 -2
View File
@@ -96,7 +96,7 @@ def test_dts_uhd_audio_routes_the_mux_to_mp4(tmp_path: Path, monkeypatch: pytest
assert out.suffix == ".mp4", "a DTS-UHD title must not be delivered as Matroska"
assert returncode == 0 and not errors
command = captured[0].args
assert command[0] == "MP4Box"
assert command[:2] == ["MP4Box", "-for-test"]
assert command[-2:] == ["-new", str(out)]
assert f"{tmp_path / 'a.mp4'}:lang=en" in command
# the child must not inherit the terminal that the live progress tree is drawing on
@@ -170,7 +170,7 @@ def test_mp4_subtitles_get_the_script_their_region_implies(tmp_path: Path, monke
tracks.mux("Title", delete=False, output_path=tmp_path / "out.muxed.mkv")
command = captured[0].args
assert f"{tmp_path / 's.vtt'}:lang=zh-Hant-TW" in command
assert f"{tmp_path / 's.vtt'}:lang=zh-Hant-TW:tkhd=enable,movie,preview" in command
assert any(arg.endswith("=Chinese (Traditional)") for arg in command)
+86 -2
View File
@@ -1,5 +1,6 @@
from __future__ import annotations
import io
import subprocess
import sys
from pathlib import Path
@@ -14,9 +15,11 @@ from tests.test_segment_decrypt import read_argv
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.drm import ClearKeyCENC
from unshackle.core.drm import decrypters as decrypters_module
from unshackle.core.drm.decrypters import mp4decrypt_or_shaka
from unshackle.core.drm.key_args import shaka_noise, track_key_args
from unshackle.core.drm.segment_decrypt import SegmentDecrypter, can_use
from unshackle.core.drm.verify import kidless_track_kids, mp4decrypt_or_shaka
from unshackle.core.drm.verify import kidless_track_kids
KEY = "b5df2ae3f63b8dda9049d6e41422e075"
OTHER_KEY = "11" * 16
@@ -128,7 +131,18 @@ def test_segment_decrypt_needs_a_key_for_a_kidless_track(
drm.content_keys[key_kid] = KEY
assert can_use(drm, "mp4decrypt", movie(two_sinf(A, piff_first=True))) is usable
assert can_use(drm, "mp4decrypt", movie(encv(A))) is True
assert can_use(drm, "mp4decrypt", movie(encv(A))) is usable
def test_segment_decrypt_waits_for_a_key_to_every_kid_the_drm_names(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
monkeypatch.setattr(config, "decrypt_segments", True, raising=False)
monkeypatch.setattr(binaries, "Mp4decrypt", tmp_path / "mp4decrypt")
drm = _widevine(A, B)
drm.content_keys[A] = KEY
assert can_use(drm, "mp4decrypt", movie(encv(A))) is False
class StubDRM:
@@ -219,6 +233,76 @@ def test_shaka_non_zero_exit_raises(
drm.decrypt_with_shaka_packager(path)
class FakePackager:
"""A shaka-packager Popen that records its argv, prints ``log`` and exits with ``code``."""
code = 0
log = ""
argv: list[str] = []
def __init__(self, cmd: list[Any], **kwargs: Any) -> None:
FakePackager.argv = [str(c) for c in cmd]
self.stderr = io.StringIO(FakePackager.log)
self.returncode = FakePackager.code
def wait(self) -> int:
return self.returncode
@pytest.fixture
def fake_packager(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list[str]:
"""Route shaka-packager to ``FakePackager`` and collect what the console prints."""
printed: list[str] = []
FakePackager.code, FakePackager.log = 0, ""
monkeypatch.setattr(decrypters_module.subprocess, "Popen", FakePackager)
monkeypatch.setattr(binaries, "ShakaPackager", Path("packager"))
monkeypatch.setattr(config.directories, "temp", tmp_path)
monkeypatch.setattr(decrypters_module.console, "log", lambda text: printed.append(str(text)))
return printed
def shaka_run(tmp_path: Path, make: Callable[[], Any], raises: type[BaseException]) -> None:
"""Decrypt a one-key track of ``make()`` through ``FakePackager`` and expect ``raises``."""
path = tmp_path / "track.mp4"
path.write_bytes(b"encrypted")
drm = make()
drm.content_keys[A] = KEY
with pytest.raises(raises):
drm.decrypt_with_shaka_packager(path)
@pytest.mark.parametrize("make", DRMS)
def test_every_drm_passes_the_fake_clock(tmp_path: Path, fake_packager: list[str], make: Callable[[], Any]) -> None:
shaka_run(tmp_path, make, FileNotFoundError)
assert "--use_fake_clock_for_muxer" in FakePackager.argv
def test_clearkey_gives_no_zero_kid_fallback(tmp_path: Path, fake_packager: list[str]) -> None:
shaka_run(tmp_path, lambda: ClearKeyCENC(kids=[A]), FileNotFoundError)
assert FakePackager.argv[FakePackager.argv.index("--keys") + 1] == f"label=0:key_id={A.hex}:key={KEY}"
@pytest.mark.parametrize(
("make", "label"),
[
pytest.param(lambda: _widevine(A), "Widevine", id="widevine"),
pytest.param(lambda: _playready(A), "PlayReady", id="playready"),
pytest.param(lambda: ClearKeyCENC(kids=[A]), "ClearKey", id="clearkey"),
],
)
def test_a_shaka_failure_logs_under_the_drm_label(
tmp_path: Path, fake_packager: list[str], make: Callable[[], Any], label: str
) -> None:
FakePackager.code, FakePackager.log = 1, "[1103/153702:ERROR:main.cc(1)] Bad key\n"
shaka_run(tmp_path, make, subprocess.CalledProcessError)
assert len(fake_packager) == 1 and f"[{label}]: " in fake_packager[0] and "Bad key" in fake_packager[0]
def test_a_shaka_ctrl_c_exit_is_a_keyboard_interrupt(tmp_path: Path, fake_packager: list[str]) -> None:
FakePackager.code = 0xC000013A
shaka_run(tmp_path, lambda: _widevine(A), KeyboardInterrupt)
@pytest.mark.parametrize(
("line", "noise"),
[
+289
View File
@@ -0,0 +1,289 @@
from __future__ import annotations
import io
import itertools
import logging
import subprocess
from pathlib import Path
from types import SimpleNamespace
from typing import Any
import pytest
from langcodes import Language
from unshackle.commands.dl import dl, hybrid_plan, pick_video
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.drm.verify import decrypter_for
from unshackle.core.manifests import DASH, HLS
from unshackle.core.titles.movie import Movie
from unshackle.core.tracks import Audio, Chapter, Chapters, Track, Tracks, Video
from unshackle.core.tracks import tracks as tracks_module
VIDEO_IDS = itertools.count()
MASTER = """#EXTM3U
#EXT-X-VERSION:12
#EXT-X-STREAM-INF:BANDWIDTH=6786883,CODECS="dvh1.20.02",RESOLUTION=960x540,VIDEO-RANGE=PQ,REQ-VIDEO-LAYOUT="CH-MONO,CH-STEREO"
stereo.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=5000000,CODECS="hvc1.2.4.L150.90",RESOLUTION=1920x1080,REQ-VIDEO-LAYOUT="CH-STEREO"
mvhevc.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=4000000,CODECS="dvh1.08.06",RESOLUTION=1920x1080,VIDEO-RANGE=PQ
flat.m3u8
"""
@pytest.mark.parametrize(
("codecs", "layout", "expected"),
[
("dvh1.20.06,ec-3", None, True),
("dvhe.20.09", None, True),
("hvc1.2.4.L150.90", '"CH-MONO,CH-STEREO"', True),
("dvh1.08.06", '"CH-MONO"', False),
("dvh1.05.06", None, False),
("dvh1.200.06", None, False),
(None, None, False),
],
)
def test_is_stereo(codecs: str | None, layout: str | None, expected: bool) -> None:
assert Video.is_stereo(codecs, video_layout=layout) is expected
def test_hls_marks_stereo_tracks_from_the_layout_or_the_codec() -> None:
tracks = HLS.from_text(MASTER, url="https://cdn.example/master.m3u8").to_tracks(Language.get("en"))
stereo = {str(v.url).rsplit("/", 1)[-1]: v.stereo for v in tracks.videos}
assert stereo == {"stereo.m3u8": True, "mvhevc.m3u8": True, "flat.m3u8": False}
def test_a_stereo_track_lists_as_3d_and_keeps_the_flag_through_to_dict() -> None:
video = Video(url="https://x/v.mp4", language="en", codec=Video.Codec.HEVC, range_=Video.Range.DV, stereo=True)
assert str(video).startswith("VID | [H.265, DV, 3D] |")
assert Video.from_dict(video.to_dict()).stereo
def stereo_video(tmp_path: Path, stereo: bool = True, **kwargs: Any) -> Video:
n = next(VIDEO_IDS)
path = tmp_path / f"v{n}.mp4"
path.write_bytes(b"\x00" * 16)
video = Video(url=f"https://x/v{n}.mp4", language="en", codec=Video.Codec.HEVC, stereo=stereo, **kwargs)
video.path = path
return video
def test_dash_marks_a_dv_profile_20_representation_stereo() -> None:
mpd = """<?xml version="1.0"?>
<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" mediaPresentationDuration="PT10S">
<Period>
<AdaptationSet contentType="video" mimeType="video/mp4">
<SegmentTemplate media="$RepresentationID$/$Number$.m4s" initialization="$RepresentationID$/init.mp4" duration="2" />
<Representation id="v3d" codecs="dvh1.20.06" bandwidth="9000000" width="1920" height="1080" />
<Representation id="v2d" codecs="dvh1.08.06" bandwidth="5000000" width="1920" height="1080" />
</AdaptationSet>
</Period>
</MPD>"""
videos = DASH.from_text(mpd, url="https://cdn.example/manifest.mpd").to_tracks(language="en").videos
assert {v.data["dash"]["representation"].get("id"): v.stereo for v in videos} == {"v3d": True, "v2d": False}
def box(kind: bytes, payload: bytes = b"") -> bytes:
return (8 + len(payload)).to_bytes(4, "big") + kind + payload
def test_a_track_with_a_second_video_layer_decrypts_with_mp4decrypt(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
(layered := tmp_path / "3d.mp4").write_bytes(box(b"moov", box(b"lhvC")))
(flat := tmp_path / "2d.mp4").write_bytes(box(b"moov", box(b"hvcC")))
monkeypatch.setattr(config, "decryption", "Shaka")
assert decrypter_for(layered) == "mp4decrypt"
assert decrypter_for(flat) == "shaka"
monkeypatch.setattr(config, "decryption", "MP4DECRYPT")
assert decrypter_for(flat) == "mp4decrypt"
def test_a_3d_title_muxes_with_mp4box(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
calls: list[str] = []
monkeypatch.setattr(Tracks, "_mux_mp4", lambda self, **kwargs: calls.append(kwargs["title"]) or (Path("x"), 0, []))
assert Tracks(stereo_video(tmp_path)).mux("Film")[1] == 0
assert calls == ["Film"]
@pytest.fixture
def mp4box_commands(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> list[list[str]]:
commands: list[list[str]] = []
class FakePopen:
def __init__(self, command: list[str], **kwargs: Any) -> None:
commands.append([str(c) for c in command])
Path(command[-1]).write_bytes(b"mp4")
self.stdout = io.StringIO("")
def wait(self) -> int:
return 0
monkeypatch.setattr(binaries, "MP4Box", Path("/bin/MP4Box"), raising=False)
monkeypatch.setattr(tracks_module.subprocess, "Popen", FakePopen)
def run_step(command: list[Any], output: Path, **kwargs: Any) -> None:
commands.append([str(c) for c in command])
output.write_bytes(b"cut")
monkeypatch.setattr(tracks_module, "run_step", run_step)
monkeypatch.setattr(config, "sample", None)
monkeypatch.setattr(config.directories, "temp", tmp_path)
return commands
def test_a_3d_mp4_mux_keeps_the_sample_whole(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mp4box_commands: list[list[str]]
) -> None:
monkeypatch.setattr(config, "sample", (120.0, 180.0))
Tracks(stereo_video(tmp_path))._mux_mp4(title="Film", output_path=tmp_path / "out.mkv")
assert len(mp4box_commands) == 1
assert "-splitx" not in mp4box_commands[0]
assert "-for-test" in mp4box_commands[0]
def test_a_2d_mp4_sample_cut_keeps_no_date(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, mp4box_commands: list[list[str]]
) -> None:
monkeypatch.setattr(config, "sample", (120.0, 180.0))
Tracks(stereo_video(tmp_path, stereo=False))._mux_mp4(title="Film", output_path=tmp_path / "out.mkv")
assert mp4box_commands[1][1:3] == ["-for-test", "-splitx"]
def test_a_3d_mp4_mux_writes_quicktime_chapters(tmp_path: Path, mp4box_commands: list[list[str]]) -> None:
for stereo, flag in ((True, "-chapqt"), (False, "-chap")):
tracks = Tracks(stereo_video(tmp_path, stereo=stereo))
tracks.chapters = Chapters([Chapter(0.0, "Story")])
tracks._mux_mp4(title="Film", output_path=tmp_path / f"out{stereo}.mkv")
assert flag in mp4box_commands[-1] and ({"-chap", "-chapqt"} - {flag}).isdisjoint(mp4box_commands[-1])
def test_a_3d_track_is_never_a_hybrid_ingredient(tmp_path: Path) -> None:
flat, layered = (stereo_video(tmp_path, stereo=s, range_=Video.Range.DV) for s in (False, True))
candidates, standalone = Tracks.partition_hybrid_videos([flat, layered], [])
assert candidates == [flat] and standalone == []
def test_2d_ranks_over_3d_at_one_resolution_and_the_signatures_differ(tmp_path: Path) -> None:
flat = stereo_video(tmp_path, stereo=False, height=1080, bitrate=5_000_000)
layered = stereo_video(tmp_path, stereo=True, height=1080, bitrate=9_000_000)
uhd = stereo_video(tmp_path, stereo=True, height=2160, bitrate=40_000_000, width=3840)
tracks = Tracks(layered, flat, uhd)
tracks.sort_videos()
assert tracks.videos == [uhd, flat, layered]
assert Tracks.signature(flat) != Tracks.signature(
stereo_video(tmp_path, stereo=True, height=1080, bitrate=5_000_000)
)
def test_a_3d_track_gets_no_ffmpeg_vui_rewrite_or_repack(tmp_path: Path) -> None:
track = stereo_video(tmp_path, range_=Video.Range.HDR10)
assert track.vui_bsf() is None
assert track.repackage() is False
assert track.path.exists()
def test_the_filename_context_marks_a_3d_title(monkeypatch: pytest.MonkeyPatch) -> None:
monkeypatch.setattr(config, "tag_rules", [])
monkeypatch.setattr(config, "language_tags", {})
class Svc:
pass
def context(stereo: bool) -> str:
movie = Movie(id_="movie-0001", service=Svc, name="The Film", year=2024, language="en")
movie.tracks = Tracks(Video(url="https://x/v.mp4", language="en", codec=Video.Codec.HEVC, stereo=stereo))
video = SimpleNamespace(
width=3840,
height=2160,
format="HEVC",
scan_type="Progressive",
hdr_format_commercial="",
hdr_format="",
hdr_format_profile="",
transfer_characteristics="PQ",
transfer_characteristics_original=None,
frame_rate="47.952",
)
return str(movie.build_base_template_context(SimpleNamespace(video_tracks=[video], audio_tracks=[]))["3d"])
assert (context(True), context(False)) == ("3D", "")
def test_worst_prefers_the_2d_track_at_one_height(tmp_path: Path) -> None:
videos = [
stereo_video(tmp_path, stereo=s, height=2160, bitrate=b)
for s, b in ((False, 9_000_000), (False, 5_000_000), (True, 40_000_000), (True, 30_000_000))
]
ranked = Tracks(videos)
ranked.sort_videos()
assert pick_video(ranked.videos, worst=True) is videos[1]
assert pick_video(ranked.videos, worst=False) is videos[0]
assert pick_video([videos[3], videos[2]], worst=True) is videos[2]
assert pick_video([], worst=True) is None
def test_hybrid_never_pairs_a_3d_dv_track(tmp_path: Path) -> None:
base = stereo_video(tmp_path, stereo=False, range_=Video.Range.HDR10, height=2160)
layered = stereo_video(tmp_path, stereo=True, range_=Video.Range.DV, height=2160)
assert hybrid_plan([base, layered], generate_dv=False) == ([], None)
def test_mp4_names_skip_the_extra_tracks_of_a_3d_video_file(tmp_path: Path, mp4box_commands: list[list[str]]) -> None:
video = stereo_video(tmp_path)
video.path.write_bytes(box(b"moov", box(b"trak") + box(b"trak")))
audio = Audio(url="https://x/a.mp4", language="en", codec=Audio.Codec.EC3, name="English")
audio.path = tmp_path / "a.mp4"
audio.path.write_bytes(b"\x00" * 16)
Tracks([video, audio])._mux_mp4(title="Film", output_path=tmp_path / "out.mkv")
command = mp4box_commands[0]
assert command[command.index("-name") + 1] == "3=English"
def test_a_3d_mux_that_loses_its_second_view_warns(
tmp_path: Path, mp4box_commands: list[list[str]], caplog: pytest.LogCaptureFixture
) -> None:
Tracks(stereo_video(tmp_path))._mux_mp4(title="Film", output_path=tmp_path / "out.mkv")
assert "lost its second view" in caplog.text
def test_a_downloaded_file_with_a_second_view_marks_the_video_3d(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch
) -> None:
video = stereo_video(tmp_path, stereo=False)
monkeypatch.setattr(Track, "download", lambda self, *a, **k: self.path.write_bytes(box(b"moov", box(b"lhvC"))))
video.download(None, None)
assert video.stereo
def test_dash_reads_dv_profile_20_from_supplemental_codecs() -> None:
mpd = """<?xml version="1.0"?>
<MPD xmlns="urn:mpeg:dash:schema:mpd:2011" type="static" mediaPresentationDuration="PT10M">
<Period id="1"><AdaptationSet contentType="video" mimeType="video/mp4" lang="en">
<Representation id="v" codecs="hvc1.2.4.L153" supplementalCodecs="dvh1.20.09" bandwidth="1" width="3840" height="2160">
<BaseURL>v.mp4</BaseURL><SegmentBase indexRange="0-1"/>
</Representation>
</AdaptationSet></Period>
</MPD>"""
assert DASH.from_text(mpd, url="https://cdn.example/m.mpd").to_tracks(language="en").videos[0].stereo
def test_hls_reads_each_variant_layout_even_when_two_share_a_playlist() -> None:
master = """#EXTM3U
#EXT-X-STREAM-INF:BANDWIDTH=2,CODECS="hvc1.2.4.L150.90",RESOLUTION=1920x1080,REQ-VIDEO-LAYOUT="CH-STEREO"
shared.m3u8
#EXT-X-STREAM-INF:BANDWIDTH=1,CODECS="hvc1.2.4.L150.90",RESOLUTION=1920x1080,AUDIO="a"
shared.m3u8
"""
tracks = HLS.from_text(master, url="https://cdn.example/master.m3u8").to_tracks(Language.get("en"))
assert any(v.stereo for v in tracks.videos)
def test_a_sample_is_cut_only_from_a_matroska_file(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
runner = dl.__new__(dl)
runner.log = logging.getLogger("download")
monkeypatch.setattr(subprocess, "run", lambda *a, **k: pytest.fail("mkvmerge ran on an MP4"))
runner.write_sample(tmp_path / "Film.mp4", SimpleNamespace(), SimpleNamespace())
assert not (tmp_path / "Sample").exists()
+26
View File
@@ -384,6 +384,32 @@ def test_stale_backup_from_a_killed_run_is_not_restored(tmp_path: Path, monkeypa
assert path.read_bytes() == b"cipher"
@pytest.mark.parametrize(
("windows", "licensed"),
[({KID: [(0.0, 2.0)], SIBLING: [(2.0, 4.0)]}, [SIBLING]), ({KID: [(0.0, 2.0)]}, []), (None, [SIBLING])],
ids=["file-uses-it", "file-does-not", "unmapped"],
)
def test_an_unkeyed_kid_the_file_uses_is_licensed_before_the_decrypt(
tmp_path: Path, monkeypatch: pytest.MonkeyPatch, windows: Optional[dict], licensed: list[UUID]
) -> None:
cmd = make_cmd(monkeypatch, Vaults("SVC"))
drm = FakeDRM({KID: GOOD})
drm.kids = [KID, SIBLING]
monkeypatch.setattr(
dl_module.verify,
"kid_windows",
lambda path: None if windows is None else dl_module.verify.KidMap(windows, True),
)
path = tmp_path / "video.mp4"
path.write_bytes(b"cipher")
licences: list[UUID] = []
cmd.decrypt_verified(drm, path, lambda drm, track_kid: licences.append(track_kid), KID)
assert licences == licensed
assert drm.decrypted_with == [{KID: GOOD}]
def test_drm_without_content_keys_still_decrypts(tmp_path: Path, monkeypatch: pytest.MonkeyPatch) -> None:
"""HLS AES-128 ClearKey has no KID:KEY map; the hook must pass it straight through."""
cmd = make_cmd(monkeypatch, Vaults("SVC"))
+5 -4
View File
@@ -99,7 +99,7 @@ def test_several_keys_and_no_single_own_kid_give_no_zero_kid_entry(make) -> None
def test_shaka_decrypt_passes_the_built_keys(monkeypatch: pytest.MonkeyPatch, tmp_path) -> None:
"""The shaka-packager call uses the same builder the tests above cover."""
from unshackle.core.drm import widevine as widevine_module
from unshackle.core.drm import decrypters as decrypters_module
seen: list[list[str]] = []
@@ -110,9 +110,9 @@ def test_shaka_decrypt_passes_the_built_keys(monkeypatch: pytest.MonkeyPatch, tm
seen.append([str(c) for c in cmd])
raise StopPopen
monkeypatch.setattr(widevine_module.subprocess, "Popen", fake_popen)
monkeypatch.setattr(widevine_module.binaries, "ShakaPackager", "shaka-packager")
monkeypatch.setattr(widevine_module.config.directories, "temp", tmp_path)
monkeypatch.setattr(decrypters_module.subprocess, "Popen", fake_popen)
monkeypatch.setattr(decrypters_module.binaries, "ShakaPackager", "shaka-packager")
monkeypatch.setattr(decrypters_module.config.directories, "temp", tmp_path)
drm = _with_keys(_widevine(VIDEO_KID), BOTH_KEYS)
media = tmp_path / "track.mp4"
media.write_bytes(b"x")
@@ -122,6 +122,7 @@ def test_shaka_decrypt_passes_the_built_keys(monkeypatch: pytest.MonkeyPatch, tm
keys = seen[0][seen[0].index("--keys") + 1]
assert _shaka_zero_entries(keys) == [VIDEO_KEY]
assert "--use_fake_clock_for_muxer" in seen[0]
def test_a_rebuilt_widevine_does_not_take_the_first_pssh_kid_as_its_own() -> None:
@@ -56,7 +56,15 @@ async def _run_batch(monkeypatch, tracks, cdm_type, calls):
return session
def fake_single(
service, title, track, pssh_str, drm_type, request, sources=None, refusal=None, vault_only=False, shared=None
service,
title,
track,
pssh_str,
drm_type,
request,
sources=None,
refusal=None,
**_,
):
calls.append((pssh_str, drm_type))
return {"00" * 16: "key"}
+8
View File
@@ -12,6 +12,7 @@ import pytest
from unshackle.core.api.progress import (
DOWNLOAD_PROGRESS_CEILING,
build_job_progress_callables,
track_log_label,
track_progress_label,
track_progress_weight,
)
@@ -48,6 +49,13 @@ def noop(**kwargs):
pass
def test_track_log_label_adds_the_track_id():
video = Video(2160, "DV")
video.id = "592464997"
assert track_log_label(video) == "video 2160p DV (592464997)"
def test_track_progress_label():
assert track_progress_label(Video(2160, "DV")) == "video 2160p DV"
assert track_progress_label(Video(1080, "HDR10+")) == "video 1080p HDR10+"
+26
View File
@@ -169,6 +169,32 @@ def test_remote_error_carries_its_reason_and_exits_a_cli_run():
assert not remote_service.RemoteError("Server error [NOT_FOUND]: no title", "NOT_FOUND").session_broke
def test_quiet_mutes_a_remote_error_the_caller_reports(caplog):
with caplog.at_level("DEBUG", logger="remote_service"), remote_service.quiet():
with remote_service.quiet():
remote_service.RemoteError("Server error [SERVICE_ERROR]: refused", "SERVICE_ERROR")
remote_service.RemoteError("Remote auth failed: bad key", "AUTH")
remote_service.RemoteError("Server error [NOT_FOUND]: outside", "NOT_FOUND")
assert [(r.levelname, r.getMessage()) for r in caplog.records] == [
("DEBUG", "Server error [SERVICE_ERROR]: refused"),
("DEBUG", "Remote auth failed: bad key"),
("ERROR", "Server error [NOT_FOUND]: outside"),
]
def test_an_unreachable_server_list_logs_one_warning(monkeypatch, caplog):
class Down:
def get(self, endpoint):
raise remote_service.RemoteError("Could not connect to remote server at http://srv", "CONNECT")
monkeypatch.setattr(remote_service, "remote_client", lambda name: Down())
monkeypatch.setattr(remote_service, "_remote_lists", {})
assert remote_service.remote_service_list("down") is None
assert [r.levelname for r in caplog.records if r.levelname in ("WARNING", "ERROR")] == ["WARNING"]
def test_a_geofenced_remote_service_gets_its_first_allowed_region_as_proxy(monkeypatch):
monkeypatch.setattr(handlers, "server_proxy_allowed", lambda request=None: True)
fenced = {"tag": "EXAMPLE", "geofence": ["US", "ca"], "geoblock": ["us"]}
@@ -58,6 +58,11 @@ def test_a_dv_track_keeps_its_hdr10_base_label_over_remote() -> None:
assert deserialize_video(serialize_video_track(v)).range_label == "DV+HDR10"
def test_a_stereo_track_stays_3d_over_remote() -> None:
v = Video(url="https://x/v.mp4", language="en", codec=Video.Codec.HEVC, range_=Video.Range.DV, stereo=True)
assert deserialize_video(serialize_video_track(v)).stereo
def test_a_track_note_shows_over_remote_and_export_and_is_not_an_edition() -> None:
from unshackle.core.tracks.track import Track
@@ -96,3 +96,48 @@ def test_proxy_mode_keeps_the_callers_type():
assert licence == b"lic"
assert svc.client.posts[0]["drm_type"] == "playready"
assert svc.client.posts[0]["pssh"] == "PR"
def test_a_refused_relayed_licence_raises_a_click_exception_with_the_server_message(caplog):
import click
from unshackle.core.remote_service import RemoteError
svc = _service("widevine", {})
svc._server_cdm = False
svc.drain_server_logs = lambda: None
def refuse(endpoint, payload, expect=()):
raise RemoteError("Licence request failed: refused", "SERVICE_ERROR")
svc.client.post = refuse
track = SimpleNamespace(id="vid", drm=[_PlayReady()])
with pytest.raises(click.ClickException, match="^Licence request failed: refused$"):
svc.proxy_license(b"chal", track, "playready")
assert not [r for r in caplog.records if r.levelname == "ERROR"]
def test_a_refused_server_cdm_licence_logs_one_warning(caplog):
from unshackle.core.remote_service import RemoteError
svc = _service("widevine", {})
def refuse(endpoint, payload, expect=()):
raise RemoteError("Server error [SERVICE_ERROR]: refused", "SERVICE_ERROR")
svc.client.post = refuse
track = SimpleNamespace(id="vid", drm=[_Widevine()])
assert svc.proxy_license(b"chal", track, "widevine") == b"chal"
assert [r.levelname for r in caplog.records if r.levelname in ("WARNING", "ERROR")] == ["WARNING"]
def test_the_kid_dl_needs_goes_to_the_server():
svc = _service("widevine", {"keys": {KID.hex: "k"}, "drm_type": "widevine"})
track = SimpleNamespace(id="vid", drm=[_Widevine()])
svc.get_widevine_license(challenge=b"", title=None, track=track, kid=KID)
assert svc.client.posts[0]["kid"] == KID.hex
@@ -115,12 +115,25 @@ async def test_batch_reports_vault_keys_from_the_init_segment_pssh(monkeypatch):
session = SessionEntry(session_id="sess", service_tag="SVC", service_instance=SimpleNamespace())
session.tracks = {"vid": track}
probed: list = []
async def validated(sid, req):
return session
def fake_single(
service, title, track, pssh_str, drm_type, request, sources=None, refusal=None, vault_only=False, shared=None
service,
title,
track,
pssh_str,
drm_type,
request,
sources=None,
refusal=None,
vault_only=False,
shared=None,
track_kid=None,
):
probed.append(track_kid)
if pssh_str == "init":
sources[CDM_KID.hex] = "sqlite"
return {CDM_KID.hex: "v"}
@@ -147,6 +160,66 @@ async def test_batch_reports_vault_keys_from_the_init_segment_pssh(monkeypatch):
assert payload["keys"] == {"vid": {CDM_KID.hex: "v"}}
assert payload["vault_keys"] == [CDM_KID.hex]
assert session.served_keys == {CDM_KID.hex: ("v", "sqlite")}
assert probed == [CDM_KID, CDM_KID]
async def test_batch_asks_again_when_a_cached_vault_answer_lacks_the_track_kid(monkeypatch):
"""Renditions that share a header: a vault answer for one track's KID does not serve a track with another KID."""
kids = {"a": KID, "b": CDM_KID, "c": CDM_KID}
session = SessionEntry(session_id="sess", service_tag="SVC", service_instance=SimpleNamespace())
session.tracks = {
tid: SimpleNamespace(id=tid, drm=[_WV("shared", [KID, CDM_KID])], get_key_id=lambda init, k=kid: k)
for tid, kid in kids.items()
}
asked: list = []
async def validated(sid, req):
return session
def fake_single(
service, title, track, pssh_str, drm_type, request, sources=None, refusal=None, track_kid=None, **_
):
asked.append(track_kid)
if track_kid == KID:
sources[KID.hex] = "sqlite"
return {KID.hex: "v"}
return {KID.hex: "v", CDM_KID.hex: "c"}
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: b"init")
monkeypatch.setattr(handlers, "handle_single_server_cdm", fake_single)
monkeypatch.setattr(handlers.config, "serve", {"users": {}}, raising=False)
resp = await handlers.session_license_handler(
{"mode": "server_cdm", "track_ids": list(kids), "drm_type": "widevine"}, "sess", None
)
payload = json.loads(resp.body)
assert asked == [KID, CDM_KID]
assert payload["keys"]["c"] == {KID.hex: "v", CDM_KID.hex: "c"}
@pytest.mark.parametrize("drm", [["drm"], []], ids=["with-drm", "clear"])
def test_the_server_reads_the_head_of_a_one_file_track_with_drm(drm):
from unshackle.core.tracks import Video
fetched: list = []
track = SimpleNamespace(
id="v",
data={},
descriptor=Video.Descriptor.URL,
drm=drm,
get_init_segment=lambda session: fetched.append(session) or b"head",
)
head = handlers.fetch_init_segment(track, "svc-session")
assert (head, fetched) == ((b"head", ["svc-session"]) if drm else (None, []))
async def test_server_tells_a_remote_vault_only_when_it_served_the_pair(monkeypatch, session):
@@ -182,7 +255,7 @@ def test_resolve_server_keys_does_not_warn_for_clear_tracks(monkeypatch):
svc.resolve_server_keys(SimpleNamespace(tracks=Tracks([video_track, audio_track])))
assert warnings == ["Server CDM returned no content keys for track v-1"]
assert warnings == ["Server CDM returned no content keys for track simplenamespace (v-1)"]
def resolve(manifest_drm, drm_type="widevine"):
@@ -27,6 +27,8 @@ from unshackle.core.drm import PlayReady
pytestmark = pytest.mark.unit
REAL_CHECK_VAULTS = handlers.check_vaults
KID_A = UUID("161219ec3df64e0eadbb3827c0ccc46d")
KID_B = UUID("43de497c6d1945f49acc2951696ec628")
@@ -161,3 +163,59 @@ async def test_uncovered_kid_reaches_the_client(server_env, fake_playready, monk
)
assert any(KID_B.hex in r["message"] for r in buf.since(0)), "client gets no reason for the missing key"
class _VaultsHolding:
"""Server vaults that hold a content key for ``held`` only."""
def __init__(self, *held: UUID):
self.held = held
self.vaults = [SimpleNamespace(name="v1")]
def get_key(self, kid):
return ("k" + kid.hex[:4], self.vaults[0]) if kid in self.held else (None, None)
@pytest.mark.parametrize(
("track_kid", "found"), [(None, None), (KID_A, {KID_A.hex}), (KID_B, None)], ids=["every-kid", "held", "not-held"]
)
def test_check_vaults_answers_for_the_track_kid(monkeypatch, track_kid, found):
monkeypatch.setattr(handlers, "load_server_vaults", lambda name: _VaultsHolding(KID_A))
hit = handlers.check_vaults([KID_A, KID_B], "SVC", track_kid=track_kid)
assert (set(hit[0]) if hit else None) == found
@pytest.mark.parametrize("track_kid", [KID_A, None], ids=["track-kid-held", "track-kid-unknown"])
def test_a_playready_track_takes_the_vault_key_only_for_its_known_kid(server_env, monkeypatch, track_kid):
monkeypatch.setattr(handlers, "check_vaults", REAL_CHECK_VAULTS)
monkeypatch.setattr(handlers, "load_server_vaults", lambda name: _VaultsHolding(KID_A))
licensed: list = []
monkeypatch.setattr(
PlayReady,
"get_content_keys",
lambda self, cdm, certificate, licence: licensed.append(self) or self.content_keys.update({KID_A: "cdm"}),
)
pro_b64 = _kidless_pro_b64()
track = SimpleNamespace(
id="vid", drm=[PlayReady(pssh=RealPSSH(base64.b64decode(pro_b64)), kid=KID_A, pssh_b64=pro_b64)]
)
sources: dict = {}
keys = handlers.handle_single_server_cdm(
SimpleNamespace(), SimpleNamespace(), track, pro_b64, "playready", None, sources, track_kid=track_kid
)
assert bool(licensed) is (track_kid is None)
assert sources == ({KID_A.hex: "v1"} if track_kid else {})
assert set(keys) == {KID_A.hex}
def test_the_client_kid_counts_only_when_the_track_drm_names_it():
track = SimpleNamespace(drm=[SimpleNamespace(kids=[KID_A])])
assert handlers.client_track_kid(KID_A.hex, track) == KID_A
assert handlers.client_track_kid(KID_B.hex, track) is None
assert handlers.client_track_kid("not-hex", track) is None
assert handlers.client_track_kid(None, track) is None
@@ -397,7 +397,15 @@ def licence_env(monkeypatch):
vault = {}
def fake_single(
service, title, track, pssh_str, drm_type, request, sources=None, refusal=None, vault_only=False, shared=None
service,
title,
track,
pssh_str,
drm_type,
request,
sources=None,
refusal=None,
**_,
):
calls.append((track.id, pssh_str))
if pssh_str in vault:
@@ -134,7 +134,15 @@ async def test_batch_shares_full_bundle_per_pssh(monkeypatch):
bundle_b = {"aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa": "k_b"}
def fake_single(
service, title, track, pssh_str, drm_type, request, sources=None, refusal=None, vault_only=False, shared=None
service,
title,
track,
pssh_str,
drm_type,
request,
sources=None,
refusal=None,
**_,
):
calls.append(pssh_str)
return dict(bundle_a) if pssh_str == pssh_a else dict(bundle_b)
+29 -1
View File
@@ -11,7 +11,7 @@ import requests
import responses
from langcodes import Language
from unshackle.core.remote_service import RemoteClient, RemoteError, RemoteService, build_title
from unshackle.core.remote_service import RemoteClient, RemoteError, RemoteService, build_title, quiet
from unshackle.core.tracks import Audio
from unshackle.core.tracks.track import Track
@@ -124,3 +124,31 @@ def test_a_gateway_error_page_counts_as_an_unreachable_server():
client.get("/api/session/s1/titles")
assert exc.value.kind == "CONNECT"
def fail_reopen(session_id, kind):
raise RemoteError("Server error [NOT_FOUND]: no title", "NOT_FOUND")
@pytest.mark.parametrize("recover", [lambda session_id, kind: False, fail_reopen], ids=["lost", "reopen-failed"])
@responses.activate
def test_a_session_lost_under_quiet_is_left_to_the_caller(monkeypatch, caplog, recover):
service = make_service()
monkeypatch.setattr(service, "recover", recover)
responses.add(responses.POST, f"{SERVER}/api/session/s1/license", json=GONE, status=404)
with quiet(), pytest.raises(RemoteError):
license_call(service, audio("a-old"))
assert not [r for r in caplog.records if r.levelname == "ERROR"]
@responses.activate
def test_closing_a_session_the_server_lost_logs_no_error(caplog):
service = make_service()
responses.add(responses.DELETE, f"{SERVER}/api/session/s1", json=GONE, status=404)
service.close()
assert service._session_id is None
assert not [r for r in caplog.records if r.levelname in ("WARNING", "ERROR")]
+55 -27
View File
@@ -146,7 +146,7 @@ from unshackle.core.utils.click_types import (
from unshackle.core.utils.collections import ci_get, merge_dict
from unshackle.core.utils.hdr10plus import probed_tracks
from unshackle.core.utils.post_scripts import NO_POST_SCRIPTS, build_context, dispatch, season_context
from unshackle.core.utils.redact import mask_proxy
from unshackle.core.utils.redact import mask_proxy, redact_secrets
from unshackle.core.utils.selector import select_multiple
from unshackle.core.utils.subprocess import ffmpeg_decodes, ffprobe, run_step
from unshackle.core.vault import Vault
@@ -336,18 +336,26 @@ def hybrid_gate_failure(videos: Sequence[Video], generate_dv: bool) -> Optional[
def hybrid_plan(videos: Sequence[Video], generate_dv: bool) -> tuple[list[Video], Optional[Video]]:
"""The hybrid bases and the DV track to pair with them.
"""The hybrid bases and the 2D DV track to pair with them.
With no DV track, a base qualifies only with ``--generate-dv``. A resolution with a
finished hybrid (``native_hybrid``) gets no base: that track is its hybrid.
"""
dv_tracks = [v for v in videos if v.range == Video.Range.DV]
dv_tracks = [v for v in videos if v.range == Video.Range.DV and not v.stereo]
native_heights = {v.height for v in videos if v.native_hybrid}
base_pool = [v for v in videos if v.height not in native_heights and (dv_tracks or generate_dv)]
matching_dv = min(dv_tracks, key=lambda v: v.height or 0) if dv_tracks else None
return select_hybrid_bases(base_pool), matching_dv
def pick_video(candidates: list[Video], worst: bool) -> Optional[Video]:
"""The best candidate, or with ``worst`` the last one, preferring 2D: a 3D track sorts after 2D at one height."""
if worst:
pool = [v for v in candidates if not v.stereo] or candidates
return pool[-1] if pool else None
return next(iter(candidates), None)
def hybrid_source_tracks(base: Video, dv: Optional[Video], base_container: Optional[Path] = None) -> list[Video]:
"""Copies of the base and DV tracks for the hybrid step, with the base on its container file.
@@ -2874,7 +2882,7 @@ class dl:
and (not codec or t.codec == codec)
and (vlang is None or str(t.language) == vlang)
]
match = candidates[-1] if worst and candidates else next(iter(candidates), None)
match = pick_video(candidates, worst)
if match and match not in non_hybrid_selected:
non_hybrid_selected.append(match)
@@ -2903,7 +2911,7 @@ class dl:
and (not codec or t.codec == codec)
and (vlang is None or str(t.language) == vlang)
]
match = candidates[-1] if worst and candidates else next(iter(candidates), None)
match = pick_video(candidates, worst)
if match and match not in selected_videos:
selected_videos.append(match)
title.tracks.videos = selected_videos
@@ -2941,7 +2949,7 @@ class dl:
and (not codec or t.codec == codec)
and (vlang is None or str(t.language) == vlang)
]
match = candidates[-1] if worst and candidates else next(iter(candidates), None)
match = pick_video(candidates, worst)
if match and match not in fallback_selected:
fallback_selected.append(match)
title.tracks.videos = fallback_selected
@@ -3448,13 +3456,9 @@ class dl:
self.suggest_missing_fonts(missing_fonts)
# Handle DRM decryption BEFORE repacking (must decrypt first!)
service_name = service.__class__.__name__.upper()
decryption_method = resolve_decryption(config.decryption_map, config.decryption, service_name)
decrypt_tool = "mp4decrypt" if decryption_method.lower() == "mp4decrypt" else "Shaka Packager"
drm_tracks = [track for track in title.tracks if track.drm]
if drm_tracks:
with console.status(f"Decrypting tracks with {decrypt_tool}..."):
with console.status("Decrypting tracks..."):
has_decrypted = False
for track in drm_tracks:
drm = track.get_drm_for_cdm(self.cdm)
@@ -3470,7 +3474,7 @@ class dl:
f"No matching DRM found for track {track} with CDM type {type(self.cdm).__name__}"
)
if has_decrypted:
self.log.info(f"Decrypted tracks with {decrypt_tool}")
self.log.info("Decrypted tracks")
if (
not no_subs
@@ -4050,7 +4054,11 @@ class dl:
"""Cut a 60 second sample from a finished file into a ``Sample`` folder next to it.
The sample starts 2 minutes in. A file shorter than 3 minutes gives its middle 60 seconds.
mkvmerge cuts only Matroska: an MP4 (3D video, or audio Matroska cannot carry) gets no sample.
"""
if source.suffix.lower() != ".mkv":
self.log.warning(f"No sample for {source.name}: mkvmerge cuts only Matroska files")
return
duration = float(media_info.general_tracks[0].duration or 0) / 1000
start: float
start, end = DEFAULT_SAMPLE
@@ -4518,6 +4526,8 @@ class dl:
}
if isinstance(t, Video):
row.update(width=t.width, height=t.height, range=t.range.name.lower() if t.range else "")
if t.stereo:
row["stereo"] = True
elif isinstance(t, Audio):
row.update(channels=str(t.channels) if t.channels else "", atmos=bool(t.joc), descriptive=t.descriptive)
elif isinstance(t, Subtitle):
@@ -4614,6 +4624,9 @@ class dl:
With ``decrypt=False`` the downloader already decrypted the segments in place, so a
failure can only flag and raise; the next run skips the flagged pair.
Before the decrypt, each KID the DRM names without a content key is licensed when the file
uses it, or when the file cannot be mapped, because ``prepare_drm`` licensed only for the track's KID.
"""
keys: Any = getattr(drm, "content_keys", None)
if keys is None:
@@ -4665,6 +4678,15 @@ class dl:
if known_bad and licence:
relicense({kid: drop(kid) for kid in known_bad}, {})
unkeyed = [kid for kid in getattr(drm, "kids", None) or [] if kid not in keys]
mapped = bool(unkeyed and decrypt and licence)
kid_map = verify.kid_windows(path) if mapped else None
if mapped and licence:
for kid in unkeyed:
if kid_map is None or kid in kid_map.windows:
licence(drm, track_kid=kid)
keys = getattr(drm, "content_keys", keys)
def vault_kids() -> dict[UUID, Vault]:
"""The KIDs whose key on the track came from a vault, with that vault."""
found = {kid: self.vaults.sources.get(kid) for kid in keys}
@@ -4678,7 +4700,7 @@ class dl:
os.link(path, backup)
except OSError as e:
self.log.debug(f"Cannot keep the ciphertext for a decrypt retry: {e!r}")
kid_map = verify.kid_windows(path) if checking else None
kid_map = (kid_map if mapped else verify.kid_windows(path)) if checking else None
windows = kid_map.windows if kid_map else {}
video = kid_map.video if kid_map else None
@@ -4995,6 +5017,8 @@ class dl:
) -> None:
"""Give a Widevine, PlayReady or ClearKey DRM its content keys: run cache, then key vaults, then ``get_keys``.
A held key for ``track_kid`` skips the licence, as the file is encrypted with that KID;
``decrypt_verified`` licenses any other KID the file turns out to use.
``force`` skips the run cache, so the licence runs again for each KID no vault answers.
A licence can answer a KID with a blank or null key; a vault key for that KID wins.
"""
@@ -5050,7 +5074,7 @@ class dl:
show_error(msg)
raise drm.Exceptions.CEKNotFound(msg)
need_license = bool(missing)
need_license = bool(missing) and (force or not track_kid or track_kid not in drm.content_keys)
if need_license:
# force licenses again, so the turn takes no cached keys
self.take_licence_turn(licence_turn, drm, [] if force else all_kids, track, track_kid, cek_tree)
@@ -5072,8 +5096,10 @@ class dl:
try:
get_keys()
except Exception as e:
if drm.content_keys:
self.log.debug(f"License call failed but keys already in content_keys: {e}")
if drm.content_keys and (not track_kid or track_kid in drm.content_keys):
self.log.warning(
f"Licence failed, using the {len(drm.content_keys)} known key(s): {redact_secrets(str(e))}"
)
else:
if isinstance(e, (drm.Exceptions.EmptyLicense, drm.Exceptions.CEKNotFound)):
msg = str(e)
@@ -5092,18 +5118,19 @@ class dl:
)
show_error(msg)
raise
else:
log_event(
"license_keys_retrieved",
level="INFO",
service=self.service,
context={
"track": str(track),
"drm_type": type(drm).__name__,
"keys_count": len(drm.content_keys),
"kids": [k.hex for k in drm.content_keys.keys()],
},
)
log_event(
"license_keys_retrieved",
level="INFO",
service=self.service,
context={
"track": str(track),
"drm_type": type(drm).__name__,
"keys_count": len(drm.content_keys),
"kids": [k.hex for k in drm.content_keys.keys()],
},
)
for kid, key in drm.content_keys.items():
self.add_key_label(cek_tree, kid, f"[red]{key}[/]" if is_null_key(key) else key, track_kid)
drop_null_keys(drm)
@@ -5302,6 +5329,7 @@ class dl:
licence(
drm_system="playready" if drm.__class__.__name__ == "PlayReady" else "widevine",
challenge=b"",
kid=track_kid,
)
except click.ClickException:
raise
+7
View File
@@ -48,6 +48,13 @@ def get_dependencies() -> list[dict]:
"desc": "MKV metadata",
"cat": "Core",
},
{
"name": "MP4Box",
"binary": binaries.MP4Box,
"required": False,
"desc": "MP4 muxing",
"cat": "Core",
},
{
"name": "shaka-packager",
"binary": binaries.ShakaPackager,
+86 -56
View File
@@ -27,6 +27,7 @@ from aiohttp import web
from unshackle.core.api.compression import safe_inflate
from unshackle.core.api.errors import APIError, APIErrorCode, categorize_exception, handle_api_exception
from unshackle.core.api.input_bridge import AuthStatus, InputBridge, NoUserBridge
from unshackle.core.api.progress import track_log_label
from unshackle.core.api.sanitize import MAX_SESSION_CACHE_KEYS, safe_cache_key, safe_cache_payload, sanitize_log
from unshackle.core.api.session_log import SessionLogBuffer, SessionLogMirror, capture_service_logs
from unshackle.core.api.session_store import SessionStore
@@ -1744,6 +1745,7 @@ def serialize_video_track(track: Video, include_url: bool = False) -> Dict[str,
"dv_compatible_bitstream": bool(getattr(track, "dv_compatible_bitstream", False)),
"hdr10plus": bool(getattr(track, "hdr10plus", False)),
"hdr10_base": bool(getattr(track, "hdr10_base", False)),
"stereo": bool(getattr(track, "stereo", False)),
}
result.update(serialize_track_common(track))
if include_url and hasattr(track, "url") and track.url:
@@ -4633,17 +4635,23 @@ def extract_pssh_from_track(track: Any, drm_type: str) -> Optional[str]:
def fetch_init_segment(track: Any, session: Any = None) -> Optional[bytes]:
"""The init segment of a DASH track; None for a track that is not DASH, or when the fetch fails."""
"""The init segment of a DASH track, or the head of a one-file track with DRM; None otherwise, or when the fetch fails."""
from unshackle.core.manifests import DASH as DASHManifest
dash = (getattr(track, "data", None) or {}).get("dash") or {}
if not all(dash.get(k) is not None for k in ("period", "adaptation_set", "representation", "manifest")):
return None
session = getattr(track, "session", None) or session
if session is None:
return None
dash = (getattr(track, "data", None) or {}).get("dash") or {}
if not all(dash.get(k) is not None for k in ("period", "adaptation_set", "representation", "manifest")):
if getattr(track, "descriptor", None) is not Video.Descriptor.URL or not getattr(track, "drm", None):
return None
try:
return track.get_init_segment(session=session)
except Exception as e:
log.debug(f"Init segment fetch failed for {track.id}: {e!r}")
return None
try:
return DASHManifest.get_period_segments(
period=dash["period"],
@@ -4803,7 +4811,7 @@ def server_drm_candidates(
candidates.insert(0, preferred)
else:
warn(
f"Track {sanitize_log(str(track.id)[:12])} wants {preferred} DRM "
f"Track {track_log_label(track)} wants {preferred} DRM "
"but the server has no device for it, using the configured DRM instead"
)
return candidates
@@ -4821,7 +4829,7 @@ def pick_server_pssh(track: Any, candidates: List[str]) -> Optional[tuple[str, s
pssh_str = extract_pssh_from_track(track, "playready")
if pssh_str:
log.info(
f"Track {sanitize_log(str(track.id)[:12])} has only a PlayReady PSSH, licensing it with the Widevine device"
f"Track {track_log_label(track)} has only a PlayReady PSSH, licensing it with the Widevine device"
)
if pssh_str:
return candidate, pssh_str
@@ -4854,28 +4862,29 @@ def once(shared: Dict[Any, Any], key: Any, make: Callable[[], Any]) -> Any:
def check_vaults(
kids: list, service_name: str, shared: Optional[Dict[str, Any]] = None
kids: list, service_name: str, shared: Optional[Dict[str, Any]] = None, track_kid: Optional[uuid.UUID] = None
) -> Optional[tuple[Dict[str, str], Dict[str, str]]]:
"""Examine the server vaults for existing content keys that match all KIDs.
"""Examine the server vaults for existing content keys that match the KIDs.
Returns `(KID:KEY, KID:vault name)` if ALL KIDs are found, None otherwise. The vault
name travels to the client, which is the only side that can prove the content key decrypts.
Returns `(KID:KEY, KID:vault name)` when every KID is found, or with ``track_kid`` when that
KID is found, with the other KIDs the vaults hold; None otherwise. The vault name travels to
the client, which is the only side that can prove the content key decrypts.
"""
from uuid import UUID
try:
vaults = once({} if shared is None else shared, "vaults", lambda: load_server_vaults(service_name))
if not vaults.vaults:
return None
keys: Dict[str, str] = {}
sources: Dict[str, str] = {}
for kid in kids:
kid_uuid = kid if isinstance(kid, UUID) else UUID(hex=str(kid))
wanted = [kid if isinstance(kid, uuid.UUID) else uuid.UUID(hex=str(kid)) for kid in kids]
if track_kid:
wanted = [track_kid, *(kid for kid in wanted if kid != track_kid)]
for kid_uuid in wanted:
content_key, vault_used = vaults.get_key(kid_uuid)
if content_key:
keys[kid_uuid.hex] = content_key
sources[kid_uuid.hex] = vault_used.name if vault_used else "unknown"
else:
elif track_kid in (None, kid_uuid):
return None
if keys:
log.info(f"Vault hit: {len(keys)} key(s) from server vaults, skipping CDM")
@@ -4950,6 +4959,17 @@ def license_with_track_fallback(drm: Any, track: Any, **kwargs: Any) -> Dict[str
return {kid: key for kid, key in keys.items() if not is_null_key(key)}
def client_track_kid(kid_hex: Any, track: Any) -> Optional[uuid.UUID]:
"""The KID a client asks the server CDM for, when the track's DRM names it; None for any other value."""
if not isinstance(kid_hex, str):
return None
try:
kid = uuid.UUID(hex=kid_hex)
except ValueError:
return None
return kid if any(kid in (getattr(d, "kids", None) or []) for d in getattr(track, "drm", None) or []) else None
def pssh_kids(pssh_b64: str, drm_type: str) -> set:
"""The KIDs a base64 PSSH names, for either DRM system."""
if drm_type == "playready":
@@ -4998,13 +5018,15 @@ def handle_single_server_cdm(
refusal: Optional[APIError] = None,
vault_only: bool = False,
shared: Optional[Dict[str, Any]] = None,
track_kid: Optional[uuid.UUID] = None,
) -> Dict[str, str]:
"""Do the single-track server_cdm licensing with the DRM class get_content_keys() flow.
``sources`` is filled with the vault name for every returned content key a server vault
supplied. A content key the CDM licensed gets no entry. It raises ``refusal`` before a live licence:
after a Widevine vault miss, or at once for PlayReady, whose licence can carry keys its header does not name.
With ``vault_only`` no server device is involved: the header's KIDs come from the vault, else ``refusal``.
supplied. A content key the CDM licensed gets no entry. A vault answers in place of the licence
when it holds ``track_kid``, the KID the track is encrypted with, or else every KID of the header.
PlayReady without ``track_kid`` always licenses, as its licence can carry keys its header does not name.
It raises ``refusal`` before a live licence. With ``vault_only`` no server device is involved.
``shared`` keeps the vaults and CDMs one ``license_response`` call built, for its next track.
"""
import base64
@@ -5021,12 +5043,16 @@ def handle_single_server_cdm(
if not pssh_b64:
raise APIError(APIErrorCode.INVALID_INPUT, "No PSSH available for server_cdm licensing")
def from_vaults(kids: Iterable[uuid.UUID]) -> Optional[Dict[str, str]]:
"""The vault keys for ``kids`` (see ``check_vaults``), with their sources noted."""
hit = check_vaults(list(kids), service_name, shared=shared, track_kid=track_kid)
if hit and sources is not None:
sources.update(hit[1])
return hit[0] if hit else None
if vault_only:
vault_hit = check_vaults(list(pssh_kids(pssh_b64, drm_type)), service_name, shared=shared)
if vault_hit:
if sources is not None:
sources.update(vault_hit[1])
return vault_hit[0]
if vault_keys := from_vaults(pssh_kids(pssh_b64, drm_type)):
return vault_keys
raise refusal or APIError(APIErrorCode.NO_CONTENT, "The server vault holds no content key for this track")
api_key = caller_key(request)
@@ -5047,6 +5073,8 @@ def handle_single_server_cdm(
# Gate on the caller's CDM device first: no device, no keys from the vault or CDM.
device_name = resolve_device_name(user_config, drm_type, service_name)
if track_kid and (vault_keys := from_vaults(pr_drm.kids)):
return vault_keys
if refusal:
raise refusal
@@ -5074,11 +5102,7 @@ def handle_single_server_cdm(
# harvest server-side keys from the vault fallback below.
device_name = resolve_device_name(user_config, drm_type, service_name)
vault_hit = check_vaults(wv_drm.kids, service_name, shared=shared)
if vault_hit:
vault_keys, vault_sources = vault_hit
if sources is not None:
sources.update(vault_sources)
if vault_keys := from_vaults(wv_drm.kids):
return vault_keys
if refusal:
raise refusal
@@ -5351,23 +5375,26 @@ def license_response(
session.log_buffer.append(logging.WARNING, message)
def license_track(
track: Any, title: Any, candidates: list
track: Any, title: Any, candidates: list, track_kid: Optional[uuid.UUID]
) -> tuple[Dict[str, str], Optional[str], Optional[str]]:
"""`(keys, drm_type, pssh)` for the track's current DRM, licensing once per unique PSSH.
A track the server must not license live goes into ``capped_tracks`` with no keys, and its PSSH
stays uncached, as a track under the cap can share it.
A cached vault answer that lacks ``track_kid`` is asked again for that KID. A track the server
must not license live goes into ``capped_tracks`` with no keys, and its PSSH stays uncached,
as a track under the cap can share it.
"""
picked = pick_server_pssh(track, candidates)
if not picked:
warn(f"No PSSH on track {sanitize_log(str(track.id)[:12])} for {', '.join(candidates) or 'any CDM'}")
warn(f"No PSSH on track {track_log_label(track)} for {', '.join(candidates) or 'any CDM'}")
return {}, None, None
candidate, pssh_str = picked
cache_key = (pssh_str, pssh_set(track))
if cache_key not in keys_by_pssh:
keys_by_pssh[cache_key] = {}
sources_by_pssh[cache_key] = {}
cached = keys_by_pssh.get(cache_key)
from_vault = bool(cached) and set(cached or {}) <= set(sources_by_pssh.get(cache_key, {}))
if cached is None or (track_kid and track_kid.hex not in cached and from_vault):
keys_by_pssh.setdefault(cache_key, {})
sources_by_pssh.setdefault(cache_key, {})
try:
keys = handle_single_server_cdm(
service,
@@ -5380,18 +5407,19 @@ def license_response(
live_licence_refusal(session, track),
vault_only=vault_only,
shared=shared,
track_kid=track_kid,
)
if keys:
keys_by_pssh[cache_key] = keys
keys_by_pssh[cache_key] = {**keys_by_pssh[cache_key], **keys}
drm_type_by_pssh[cache_key] = candidate
except SystemExit:
warn(f"Service exited while resolving keys for track {sanitize_log(str(track.id)[:12])}, skipping")
warn(f"Service exited while resolving keys for track {track_log_label(track)}, skipping")
except (Exception, SystemExit) as e:
if isinstance(e, APIError) and e.error_code is APIErrorCode.SERVER_CDM_CAPPED:
del keys_by_pssh[cache_key], sources_by_pssh[cache_key]
capped_tracks[str(track.id)] = e.details
return {}, None, pssh_str
warn(f"Failed to resolve keys for track {sanitize_log(str(track.id)[:12])}: {redact_all(str(e))}")
warn(f"Failed to resolve keys for track {track_log_label(track)}: {redact_all(str(e))}")
return keys_by_pssh[cache_key], drm_type_by_pssh.get(cache_key), pssh_str
for tid in track_ids:
@@ -5403,7 +5431,7 @@ def license_response(
init_data = fetch_init_segment(track, svc_session)
ensure_track_drm(track, svc_session, init_data)
if not track.drm or all(isinstance(d, BlowfishStripe) for d in track.drm):
log.info(f"Track {sanitize_log(tid[:12])} needs no licence, so it has no keys to resolve")
log.info(f"Track {track_log_label(track)} needs no licence, so it has no keys to resolve")
clear_tracks.append(tid)
continue
@@ -5415,34 +5443,34 @@ def license_response(
else server_drm_candidates(service_tag, user_config, drm_type, track, warn)
)
keys, track_drm_type, pssh_str = license_track(track, title, candidates)
probed: Optional[uuid.UUID] = None
if init_data:
try:
probed = track.get_key_id(init_data)
except Exception as e:
log.debug(f"KID probe failed for {sanitize_log(tid[:12])}: {e!r}")
keys, track_drm_type, pssh_str = license_track(track, title, candidates, probed)
# a refused track still tries its init segment PSSH, where a server vault can hold the key
refusal = capped_tracks.pop(tid, None)
track_kid = None
if init_data:
try:
track_kid = track.get_key_id(init_data)
except Exception as e:
log.debug(f"KID probe failed for {sanitize_log(tid[:12])}: {e!r}")
if track_kid and track_kid.hex not in keys:
if probed and probed.hex not in keys:
manifest_drm = track.drm
track.drm = drm_from_init_segment(track, init_data=init_data) or manifest_drm
init_keys, init_drm_type, init_pssh = license_track(track, title, candidates)
init_keys, init_drm_type, init_pssh = license_track(track, title, candidates, probed)
init_refusal = capped_tracks.pop(tid, None)
if init_pssh != pssh_str:
log.info(
f"The manifest PSSH gave no content key for KID {track_kid.hex} of track "
f"{sanitize_log(tid[:12])}, tried the init segment PSSH"
f"The manifest PSSH gave no content key for KID {probed.hex} of track "
f"{track_log_label(track)}, tried the init segment PSSH"
)
if track_kid.hex in init_keys:
if probed.hex in init_keys:
keys, track_drm_type, pssh_str = init_keys, init_drm_type, init_pssh
refusal = None
else:
track.drm = manifest_drm
refusal = refusal or init_refusal
if not refusal:
warn(f"No content key for KID {track_kid.hex} of track {sanitize_log(tid[:12])}")
warn(f"No content key for KID {probed.hex} of track {track_log_label(track)}")
if refusal:
capped_tracks[tid] = refusal
continue
@@ -5502,12 +5530,13 @@ def license_response(
if not picked:
raise APIError(
APIErrorCode.INVALID_INPUT,
f"No PSSH on track {sanitize_log(track_id[:12])} for {', '.join(candidates) or 'any CDM'}",
f"No PSSH on track {track_log_label(track)} for {', '.join(candidates) or 'any CDM'}",
)
server_drm_type, server_pssh = picked
asked_kid = client_track_kid(data.get("kid"), track)
if pssh_b64 and server_drm_type != drm_type:
log.info(
f"Client asked for {sanitize_log(drm_type)} on track {sanitize_log(track_id[:12])}, "
f"Client asked for {sanitize_log(drm_type)} on track {track_log_label(track)}, "
f"licensing with the server's {server_drm_type} CDM"
)
pssh_b64 = None
@@ -5526,8 +5555,9 @@ def license_response(
key_sources,
live_licence_refusal(session, track),
vault_only=vault_only,
track_kid=asked_kid,
)
log.info(f"Server CDM resolved {len(keys)} key(s) for track {sanitize_log(track_id[:12])}")
log.info(f"Server CDM resolved {len(keys)} key(s) for track {track_log_label(track)}")
note_served_keys(session, keys, key_sources)
return web.json_response({"keys": keys, "vault_keys": list(key_sources), "drm_type": server_drm_type})
+6
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import time
from typing import Any, Callable, Optional
from unshackle.core.api.sanitize import sanitize_log
from unshackle.core.constants import AnyTrack
from unshackle.core.utilities import standard_resolution
@@ -49,6 +50,11 @@ def track_progress_label(track: AnyTrack) -> str:
return track_type.lower()
def track_log_label(track: AnyTrack) -> str:
"""Track label for a log line: its short label and its ID, e.g. "video 360p DV (592464997)"."""
return f"{track_progress_label(track)} ({sanitize_log(str(track.id)[:12])})"
def track_progress_weight(track: AnyTrack) -> float:
"""Track weight in the aggregate (its bitrate in bits/s), so video/audio dominate subtitles."""
bitrate = getattr(track, "bitrate", None)
+2
View File
@@ -79,6 +79,7 @@ __binaries = {
"DoviTool": ("dovi_tool",),
"HDR10PlusTool": ("hdr10plus_tool", "HDR10Plus_tool"),
"Mp4decrypt": ("mp4decrypt",),
"MP4Box": ("MP4Box", "mp4box"),
"Docker": ("docker",),
"Git": ("git",),
}
@@ -226,6 +227,7 @@ __all__ = (
"DoviTool",
"HDR10PlusTool",
"Mp4decrypt",
"MP4Box",
"Docker",
"Git",
"find",
+1
View File
@@ -279,6 +279,7 @@ class Config:
"video",
"hdr",
"hfr",
"3d",
"edition",
"repack",
"lang_tag",
+11 -151
View File
@@ -2,22 +2,15 @@ from __future__ import annotations
import base64
import json
import shutil
import subprocess
import textwrap
import time
from pathlib import Path
from typing import Any, Callable, Iterable, Optional, Union
from uuid import UUID
from requests import Session
from rich.text import Text
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.console import console
from unshackle.core.drm import key_args as drm_key_args
from unshackle.core.drm.verify import kidless_track_kids, mp4decrypt_or_shaka, read_top_level_box
from unshackle.core.drm.decrypters import decrypt_cenc, mp4decrypt, shaka_decrypt
from unshackle.core.utilities import log_event
from unshackle.core.utils import run_proxy
@@ -181,50 +174,8 @@ class ClearKeyCENC:
)
def decrypt(self, path: Path) -> None:
"""
Decrypt a Track with ClearKey DRM (standard CENC).
Args:
path: Path to the encrypted file to decrypt
Raises:
EnvironmentError if the required decryption executable could not be found.
ValueError if unshackle has not yet downloaded the track.
SubprocessError if the decryption process returned a non-zero exit code.
"""
if not self.content_keys:
raise ValueError("Cannot decrypt a Track without any Content Keys...")
if not path or not path.exists():
raise ValueError("Tried to decrypt a file that does not exist.")
decrypter = str(getattr(config, "decryption", "")).lower()
tool = "mp4decrypt" if decrypter == "mp4decrypt" else "shaka-packager"
log_event(
"drm_decrypt",
level="DEBUG",
message=f"Decrypting {path.name} with {tool}",
drm_type="ClearKeyCENC",
tool=tool,
file=path.name,
key_count=len(self.content_keys),
)
decrypt_start = time.monotonic()
if decrypter == "mp4decrypt":
mp4decrypt_or_shaka(self, path)
else:
self.decrypt_with_shaka_packager(path)
log_event(
"drm_decrypt_complete",
level="DEBUG",
message=f"Decrypted {path.name} with {tool}",
drm_type="ClearKeyCENC",
tool=tool,
file=path.name,
duration_ms=round((time.monotonic() - decrypt_start) * 1000, 1),
output_size=path.stat().st_size if path.exists() else 0,
)
"""Decrypt the file at ``path`` in place with this track's ClearKey (standard CENC) content keys."""
decrypt_cenc(self, path, "ClearKeyCENC")
def mp4decrypt_key_args(self, track_kids: Optional[dict[int, UUID]] = None) -> list[str]:
"""Build the mp4decrypt --key arguments for every content key, and by track ID for each ``track_kids`` KID."""
@@ -234,107 +185,16 @@ class ClearKeyCENC:
return key_args + drm_key_args.track_key_args(self.content_keys, track_kids)
def decrypt_with_mp4decrypt(self, path: Path) -> None:
"""Decrypt using mp4decrypt"""
if not binaries.Mp4decrypt:
raise EnvironmentError("mp4decrypt executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
key_args = self.mp4decrypt_key_args(kidless_track_kids(read_top_level_box(path, b"moov") or b""))
cmd = [
str(binaries.Mp4decrypt),
"--show-progress",
*key_args,
str(path),
str(output_path),
]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else f"mp4decrypt failed with exit code {e.returncode}"
raise subprocess.CalledProcessError(e.returncode, cmd, output=e.stdout, stderr=error_msg)
if not output_path.exists():
raise RuntimeError(f"mp4decrypt failed: output file {output_path} was not created")
if output_path.stat().st_size == 0:
raise RuntimeError(f"mp4decrypt failed: output file {output_path} is empty")
path.unlink()
shutil.move(output_path, path)
"""Decrypt with mp4decrypt."""
mp4decrypt(path, self.mp4decrypt_key_args)
def decrypt_with_shaka_packager(self, path: Path) -> None:
"""Decrypt with shaka-packager"""
if not binaries.ShakaPackager:
raise EnvironmentError("Shaka Packager executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
config.directories.temp.mkdir(parents=True, exist_ok=True)
try:
arguments: list[str | Path] = [
f"input={path},stream=0,output={output_path},output_format=MP4",
"--enable_raw_key_decryption",
"--keys",
",".join(
"label={}:key_id={}:key={}".format(i, kid.hex, key.lower())
for i, (kid, key) in enumerate(self.content_keys.items())
),
"--temp_dir",
config.directories.temp,
]
p = subprocess.Popen(
[binaries.ShakaPackager, *arguments],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
universal_newlines=True,
encoding="utf-8",
errors="replace",
)
stream_skipped = False
shaka_log_buffer = ""
assert p.stderr is not None
for line in iter(p.stderr.readline, ""):
line = line.strip()
if not line:
continue
if "Skip stream" in line:
# file/segment was so small that it didn't have any actual data, ignore
stream_skipped = True
if drm_key_args.shaka_noise(line):
continue
shaka_log_buffer += f"{line.strip()}\n"
p.wait()
if p.returncode != 0:
if shaka_log_buffer:
# wrap to console width - padding - '[ClearKey]: '
shaka_log_buffer = "\n ".join(
textwrap.wrap(shaka_log_buffer.rstrip(), width=console.width - 22, initial_indent="")
)
console.log(Text.from_ansi("\n[ClearKey]: " + shaka_log_buffer))
raise subprocess.CalledProcessError(p.returncode, [binaries.ShakaPackager, *arguments])
path.unlink()
if not stream_skipped:
shutil.move(output_path, path)
except subprocess.CalledProcessError as e:
if e.returncode == 0xC000013A: # STATUS_CONTROL_C_EXIT
raise KeyboardInterrupt()
raise
"""Decrypt with shaka-packager."""
keys = ",".join(
"label={}:key_id={}:key={}".format(i, kid.hex, key.lower())
for i, (kid, key) in enumerate(self.content_keys.items())
)
shaka_decrypt(path, keys, "ClearKey")
class Exceptions:
class KIDNotFound(Exception):
+172
View File
@@ -0,0 +1,172 @@
"""Whole-file CENC decryption with mp4decrypt or shaka-packager, shared by Widevine, PlayReady and ClearKeyCENC."""
import logging
import shutil
import subprocess
import textwrap
from pathlib import Path
from time import monotonic
from typing import Any, Callable
from uuid import UUID
from rich.text import Text
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.console import console
from unshackle.core.drm.key_args import shaka_noise
from unshackle.core.drm.verify import (
decrypter_for,
has_encrypted_sample_entry,
kidless_track_kids,
read_top_level_box,
)
from unshackle.core.utilities import log_event
log = logging.getLogger(__name__)
def mp4decrypt_or_shaka(drm: Any, path: Path) -> None:
"""Decrypt ``path`` with mp4decrypt, and again with shaka-packager when mp4decrypt copied it through.
mp4decrypt copies a track it has no content key for unchanged, so its first moof stays the same.
"""
first_moof = read_top_level_box(path, b"moof")
drm.decrypt_with_mp4decrypt(path)
if has_encrypted_sample_entry(path) and read_top_level_box(path, b"moof") == first_moof:
log.warning(f"mp4decrypt left {path.name} encrypted, decrypting it with shaka-packager")
drm.decrypt_with_shaka_packager(path)
def decrypt_cenc(drm: Any, path: Path, drm_type: str) -> None:
"""Decrypt ``path`` in place with the decrypter ``decrypter_for`` picks, using the content keys of ``drm``."""
if not drm.content_keys:
raise ValueError("Cannot decrypt a Track without any Content Keys...")
if not path or not path.exists():
raise ValueError("Tried to decrypt a file that does not exist.")
decrypter = decrypter_for(path)
tool = "mp4decrypt" if decrypter == "mp4decrypt" else "shaka-packager"
log_event(
"drm_decrypt",
level="DEBUG",
message=f"Decrypting {path.name} with {tool}",
drm_type=drm_type,
tool=tool,
file=path.name,
key_count=len(drm.content_keys),
)
decrypt_start = monotonic()
if decrypter == "mp4decrypt":
mp4decrypt_or_shaka(drm, path)
else:
drm.decrypt_with_shaka_packager(path)
log_event(
"drm_decrypt_complete",
level="DEBUG",
message=f"Decrypted {path.name} with {tool}",
drm_type=drm_type,
tool=tool,
file=path.name,
duration_ms=round((monotonic() - decrypt_start) * 1000, 1),
output_size=path.stat().st_size if path.exists() else 0,
)
def mp4decrypt(path: Path, key_args_for: Callable[[dict[int, UUID]], list[str]]) -> None:
"""Decrypt ``path`` in place with mp4decrypt, taking the ``--key`` arguments from ``key_args_for(track_kids)``."""
if not binaries.Mp4decrypt:
raise EnvironmentError("mp4decrypt executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
key_args = key_args_for(kidless_track_kids(read_top_level_box(path, b"moov") or b""))
cmd = [str(binaries.Mp4decrypt), "--show-progress", *key_args, str(path), str(output_path)]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else f"mp4decrypt failed with exit code {e.returncode}"
raise subprocess.CalledProcessError(e.returncode, cmd, output=e.stdout, stderr=error_msg)
if not output_path.exists():
raise RuntimeError(f"mp4decrypt failed: output file {output_path} was not created")
if output_path.stat().st_size == 0:
raise RuntimeError(f"mp4decrypt failed: output file {output_path} is empty")
path.unlink()
shutil.move(output_path, path)
def shaka_decrypt(path: Path, keys: str, label: str) -> None:
"""Decrypt ``path`` in place with shaka-packager and its ``--keys`` value ``keys``.
Only a non-zero exit is a failure; its non-noise log lines go to the console under ``[label]``.
"""
if not binaries.ShakaPackager:
raise EnvironmentError("Shaka Packager executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
config.directories.temp.mkdir(parents=True, exist_ok=True)
arguments: list[str | Path] = [
f"input={path},stream=0,output={output_path},output_format=MP4",
"--enable_raw_key_decryption",
"--keys",
keys,
"--use_fake_clock_for_muxer",
"--temp_dir",
config.directories.temp,
]
try:
p = subprocess.Popen(
[binaries.ShakaPackager, *arguments],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
stream_skipped = False
shaka_log_buffer = ""
assert p.stderr is not None
for line in iter(p.stderr.readline, ""):
line = line.strip()
if not line:
continue
if "Skip stream" in line:
# file/segment was so small that it didn't have any actual data, ignore
stream_skipped = True
if shaka_noise(line):
continue
shaka_log_buffer += f"{line}\n"
p.wait()
if p.returncode != 0:
if shaka_log_buffer:
shaka_log_buffer = "\n ".join(
textwrap.wrap(shaka_log_buffer.rstrip(), width=console.width - 22, initial_indent="")
)
console.log(Text.from_ansi(f"\n[{label}]: " + shaka_log_buffer))
raise subprocess.CalledProcessError(p.returncode, [binaries.ShakaPackager, *arguments])
path.unlink()
if not stream_skipped:
shutil.move(output_path, path)
except subprocess.CalledProcessError as e:
if e.returncode == 0xC000013A: # STATUS_CONTROL_C_EXIT
raise KeyboardInterrupt()
raise
+7 -145
View File
@@ -2,10 +2,6 @@ from __future__ import annotations
import base64
import re
import shutil
import subprocess
import textwrap
import time
from pathlib import Path
from threading import Lock
from typing import Any, Callable, Optional, Union
@@ -18,14 +14,11 @@ from pyplayready.cdm import Cdm as PlayReadyCdm
from pyplayready.misc.exceptions import InvalidPssh
from pyplayready.system.pssh import PSSH
from pyplayready.system.wrmheader import WRMHeader
from rich.text import Text
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.console import console
from unshackle.core.constants import AnyTrack
from unshackle.core.drm import key_args as drm_key_args
from unshackle.core.drm.verify import kidless_track_kids, mp4decrypt_or_shaka, read_top_level_box
from unshackle.core.drm.decrypters import decrypt_cenc, mp4decrypt, shaka_decrypt
from unshackle.core.session import AnySession, utf8_text
from unshackle.core.utilities import call_licence, get_boxes, log_event
from unshackle.core.utils import run_proxy
@@ -542,50 +535,8 @@ class PlayReady:
return None
def decrypt(self, path: Path) -> None:
"""
Decrypt a Track with PlayReady DRM.
Args:
path: Path to the encrypted file to decrypt
Raises:
EnvironmentError if the required decryption executable could not be found.
ValueError if unshackle has not yet downloaded the track.
SubprocessError if the decryption process returned a non-zero exit code.
"""
if not self.content_keys:
raise ValueError("Cannot decrypt a Track without any Content Keys...")
if not path or not path.exists():
raise ValueError("Tried to decrypt a file that does not exist.")
decrypter = str(getattr(config, "decryption", "")).lower()
tool = "mp4decrypt" if decrypter == "mp4decrypt" else "shaka-packager"
log_event(
"drm_decrypt",
level="DEBUG",
message=f"Decrypting {path.name} with {tool}",
drm_type="PlayReady",
tool=tool,
file=path.name,
key_count=len(self.content_keys),
)
decrypt_start = time.monotonic()
if decrypter == "mp4decrypt":
mp4decrypt_or_shaka(self, path)
else:
self.decrypt_with_shaka_packager(path)
log_event(
"drm_decrypt_complete",
level="DEBUG",
message=f"Decrypted {path.name} with {tool}",
drm_type="PlayReady",
tool=tool,
file=path.name,
duration_ms=round((time.monotonic() - decrypt_start) * 1000, 1),
output_size=path.stat().st_size if path.exists() else 0,
)
"""Decrypt the file at ``path`` in place with this track's PlayReady content keys."""
decrypt_cenc(self, path, "PlayReady")
def own_kids(self) -> list[UUID]:
"""Return the KIDs that identify this track's own content key.
@@ -602,101 +553,12 @@ class PlayReady:
return drm_key_args.mp4decrypt_key_args(self.content_keys, self.own_kids(), track_kids)
def decrypt_with_mp4decrypt(self, path: Path) -> None:
"""Decrypt using mp4decrypt"""
if not binaries.Mp4decrypt:
raise EnvironmentError("mp4decrypt executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
key_args = self.mp4decrypt_key_args(kidless_track_kids(read_top_level_box(path, b"moov") or b""))
cmd = [
str(binaries.Mp4decrypt),
"--show-progress",
*key_args,
str(path),
str(output_path),
]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else f"mp4decrypt failed with exit code {e.returncode}"
raise subprocess.CalledProcessError(e.returncode, cmd, output=e.stdout, stderr=error_msg)
if not output_path.exists():
raise RuntimeError(f"mp4decrypt failed: output file {output_path} was not created")
if output_path.stat().st_size == 0:
raise RuntimeError(f"mp4decrypt failed: output file {output_path} is empty")
path.unlink()
shutil.move(output_path, path)
"""Decrypt with mp4decrypt."""
mp4decrypt(path, self.mp4decrypt_key_args)
def decrypt_with_shaka_packager(self, path: Path) -> None:
"""Decrypt with shaka-packager (original method)"""
if not binaries.ShakaPackager:
raise EnvironmentError("Shaka Packager executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
config.directories.temp.mkdir(parents=True, exist_ok=True)
try:
arguments: list[str | Path] = [
f"input={path},stream=0,output={output_path},output_format=MP4",
"--enable_raw_key_decryption",
"--keys",
drm_key_args.shaka_keys(self.content_keys, self.own_kids()),
"--temp_dir",
config.directories.temp,
]
p = subprocess.Popen(
[binaries.ShakaPackager, *arguments],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
stream_skipped = False
shaka_log_buffer = ""
assert p.stderr is not None
for line in iter(p.stderr.readline, ""):
line = line.strip()
if not line:
continue
if "Skip stream" in line:
stream_skipped = True
if drm_key_args.shaka_noise(line):
continue
shaka_log_buffer += f"{line.strip()}\n"
p.wait()
if p.returncode != 0:
if shaka_log_buffer:
shaka_log_buffer = "\n ".join(
textwrap.wrap(shaka_log_buffer.rstrip(), width=console.width - 22, initial_indent="")
)
console.log(Text.from_ansi("\n[PlayReady]: " + shaka_log_buffer))
raise subprocess.CalledProcessError(p.returncode, [binaries.ShakaPackager, *arguments])
path.unlink()
if not stream_skipped:
shutil.move(output_path, path)
except subprocess.CalledProcessError as e:
if e.returncode == 0xC000013A:
raise KeyboardInterrupt()
raise
"""Decrypt with shaka-packager."""
shaka_decrypt(path, drm_key_args.shaka_keys(self.content_keys, self.own_kids()), "PlayReady")
class Exceptions:
class PSSHNotFound(Exception):
+2
View File
@@ -24,6 +24,7 @@ def can_use(drm: Any, decryption: str, init: Optional[bytes] = None) -> bool:
"""Whether unshackle can decrypt a track's segments one by one as they download.
A track whose ``init`` mp4decrypt cannot key goes to the whole-file decrypt, which has the shaka fallback.
So does a track whose DRM names a KID with no content key, as only that decrypt can license a KID the file uses.
"""
return (
isinstance(drm, (Widevine, PlayReady, ClearKeyCENC))
@@ -31,6 +32,7 @@ def can_use(drm: Any, decryption: str, init: Optional[bytes] = None) -> bool:
and config.decrypt_segments
and bool(binaries.Mp4decrypt)
and not (init and set(kidless_track_kids(init).values()) - set(drm.content_keys))
and set(getattr(drm, "kids", None) or []) <= set(drm.content_keys)
)
+29 -8
View File
@@ -5,6 +5,8 @@ from pathlib import Path
from typing import Any, BinaryIO, Callable, Iterator, NamedTuple, Optional
from uuid import UUID
from unshackle.core.config import config
log = logging.getLogger(__name__)
DECRYPT_HOOK: Optional[Callable[..., None]] = None
@@ -221,16 +223,35 @@ def kidless_track_kids(data: bytes) -> dict[int, UUID]:
return kids
def mp4decrypt_or_shaka(drm: Any, path: Path) -> None:
"""Decrypt ``path`` with mp4decrypt, and again with shaka-packager when mp4decrypt copied it through.
def find_child_box(buf: bytes, box_type: bytes) -> Optional[bytes]:
"""Return the first box of the given type nested anywhere in buf, header included."""
i = buf.find(box_type, 4)
while i != -1:
size = int.from_bytes(buf[i - 4 : i], "big")
if 8 <= size <= len(buf) - (i - 4):
return buf[i - 4 : i - 4 + size]
i = buf.find(box_type, i + 1)
return None
mp4decrypt copies a track it has no content key for unchanged, so its first moof stays the same.
def has_second_view(path: Path) -> bool:
"""True if the MP4 holds a second video layer (an ``lhvC`` box, as stereoscopic MV-HEVC has)."""
try:
return find_child_box(read_top_level_box(path, b"moov") or b"", b"lhvC") is not None
except OSError:
return False
def decrypter_for(path: Path) -> str:
"""The configured decrypter, or mp4decrypt for a multi-layer (MV-HEVC) track.
shaka-packager refuses a version 3 dvvC (Dolby Vision profile 20) and, before v3.5.0, drops lhvC and vexu.
"""
first_moof = read_top_level_box(path, b"moof")
drm.decrypt_with_mp4decrypt(path)
if has_encrypted_sample_entry(path) and read_top_level_box(path, b"moof") == first_moof:
log.warning(f"mp4decrypt left {path.name} encrypted, decrypting it with shaka-packager")
drm.decrypt_with_shaka_packager(path)
configured = str(getattr(config, "decryption", "")).lower()
if configured != "mp4decrypt" and has_second_view(path):
log.debug(f"Decrypting {path.name} with mp4decrypt: it holds a second video layer (lhvC)")
return "mp4decrypt"
return configured
def senc_protects_samples(buf: bytes, body: int, end: int) -> bool:
+7 -148
View File
@@ -1,10 +1,6 @@
from __future__ import annotations
import base64
import shutil
import subprocess
import textwrap
import time
from pathlib import Path
from typing import Any, Callable, Optional, Union
from uuid import UUID
@@ -14,14 +10,11 @@ from construct import Container
from pymp4.parser import Box
from pywidevine.cdm import Cdm as WidevineCdm
from pywidevine.pssh import PSSH
from rich.text import Text
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.console import console
from unshackle.core.constants import AnyTrack
from unshackle.core.drm import key_args as drm_key_args
from unshackle.core.drm.verify import kidless_track_kids, mp4decrypt_or_shaka, read_top_level_box
from unshackle.core.drm.decrypters import decrypt_cenc, mp4decrypt, shaka_decrypt
from unshackle.core.session import AnySession, utf8_text
from unshackle.core.utilities import call_licence, get_boxes, log_event
from unshackle.core.utils import run_proxy
@@ -292,50 +285,8 @@ class Widevine:
cdm.close(session_id)
def decrypt(self, path: Path) -> None:
"""
Decrypt a Track with Widevine DRM.
Args:
path: Path to the encrypted file to decrypt
Raises:
EnvironmentError if the required decryption executable could not be found.
ValueError if unshackle has not yet downloaded the track.
SubprocessError if the decryption process returned a non-zero exit code.
"""
if not self.content_keys:
raise ValueError("Cannot decrypt a Track without any Content Keys...")
if not path or not path.exists():
raise ValueError("Tried to decrypt a file that does not exist.")
decrypter = str(getattr(config, "decryption", "")).lower()
tool = "mp4decrypt" if decrypter == "mp4decrypt" else "shaka-packager"
log_event(
"drm_decrypt",
level="DEBUG",
message=f"Decrypting {path.name} with {tool}",
drm_type="Widevine",
tool=tool,
file=path.name,
key_count=len(self.content_keys),
)
decrypt_start = time.monotonic()
if decrypter == "mp4decrypt":
mp4decrypt_or_shaka(self, path)
else:
self.decrypt_with_shaka_packager(path)
log_event(
"drm_decrypt_complete",
level="DEBUG",
message=f"Decrypted {path.name} with {tool}",
drm_type="Widevine",
tool=tool,
file=path.name,
duration_ms=round((time.monotonic() - decrypt_start) * 1000, 1),
output_size=path.stat().st_size if path.exists() else 0,
)
"""Decrypt the file at ``path`` in place with this track's Widevine content keys."""
decrypt_cenc(self, path, "Widevine")
def own_kids(self) -> list[UUID]:
"""Return the KIDs that identify this track's own content key.
@@ -352,104 +303,12 @@ class Widevine:
return drm_key_args.mp4decrypt_key_args(self.content_keys, self.own_kids(), track_kids)
def decrypt_with_mp4decrypt(self, path: Path) -> None:
"""Decrypt using mp4decrypt"""
if not binaries.Mp4decrypt:
raise EnvironmentError("mp4decrypt executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
key_args = self.mp4decrypt_key_args(kidless_track_kids(read_top_level_box(path, b"moov") or b""))
cmd = [
str(binaries.Mp4decrypt),
"--show-progress",
*key_args,
str(path),
str(output_path),
]
try:
subprocess.run(
cmd,
check=True,
stdout=subprocess.PIPE,
stderr=subprocess.PIPE,
text=True,
encoding="utf-8",
errors="replace",
)
except subprocess.CalledProcessError as e:
error_msg = e.stderr if e.stderr else f"mp4decrypt failed with exit code {e.returncode}"
raise subprocess.CalledProcessError(e.returncode, cmd, output=e.stdout, stderr=error_msg)
if not output_path.exists():
raise RuntimeError(f"mp4decrypt failed: output file {output_path} was not created")
if output_path.stat().st_size == 0:
raise RuntimeError(f"mp4decrypt failed: output file {output_path} is empty")
path.unlink()
shutil.move(output_path, path)
"""Decrypt with mp4decrypt."""
mp4decrypt(path, self.mp4decrypt_key_args)
def decrypt_with_shaka_packager(self, path: Path) -> None:
"""Decrypt with shaka-packager (original method)"""
if not binaries.ShakaPackager:
raise EnvironmentError("Shaka Packager executable not found but is required.")
output_path = path.with_stem(f"{path.stem}_decrypted")
config.directories.temp.mkdir(parents=True, exist_ok=True)
try:
arguments: list[str | Path] = [
f"input={path},stream=0,output={output_path},output_format=MP4",
"--enable_raw_key_decryption",
"--keys",
drm_key_args.shaka_keys(self.content_keys, self.own_kids()),
"--temp_dir",
config.directories.temp,
]
p = subprocess.Popen(
[binaries.ShakaPackager, *arguments],
stdout=subprocess.DEVNULL,
stderr=subprocess.PIPE,
universal_newlines=True,
encoding="utf-8",
errors="replace",
)
stream_skipped = False
shaka_log_buffer = ""
assert p.stderr is not None
for line in iter(p.stderr.readline, ""):
line = line.strip()
if not line:
continue
if "Skip stream" in line:
# file/segment was so small that it didn't have any actual data, ignore
stream_skipped = True
if drm_key_args.shaka_noise(line):
continue
shaka_log_buffer += f"{line.strip()}\n"
p.wait()
if p.returncode != 0:
if shaka_log_buffer:
# wrap to console width - padding - '[Widevine]: '
shaka_log_buffer = "\n ".join(
textwrap.wrap(shaka_log_buffer.rstrip(), width=console.width - 22, initial_indent="")
)
console.log(Text.from_ansi("\n[Widevine]: " + shaka_log_buffer))
raise subprocess.CalledProcessError(p.returncode, [binaries.ShakaPackager, *arguments])
path.unlink()
if not stream_skipped:
shutil.move(output_path, path)
except subprocess.CalledProcessError as e:
if e.returncode == 0xC000013A: # STATUS_CONTROL_C_EXIT
raise KeyboardInterrupt()
raise
"""Decrypt with shaka-packager."""
shaka_decrypt(path, drm_key_args.shaka_keys(self.content_keys, self.own_kids()), "Widevine")
class Exceptions:
class PSSHNotFound(Exception):
+1
View File
@@ -562,6 +562,7 @@ class ImportService:
width=int_or_none(row.get("width")),
height=int_or_none(row.get("height")),
fps=row.get("fps") or None,
stereo=bool(row.get("stereo")),
)
else:
track = Audio(
+1
View File
@@ -333,6 +333,7 @@ class DASH:
video_args["dv_compatible_bitstream"] = Video.is_dv_composite(
track_codec, codecs, get("supplementalCodecs")
)
video_args["stereo"] = Video.is_stereo(codecs, get("supplementalCodecs"))
elif content_type == "audio":
track_type = Audio
track_codec = Audio.Codec.from_codecs(codecs or "")
+8 -1
View File
@@ -222,7 +222,11 @@ class HLS:
supplemental_codecs = self.supplemental_codecs_by_uri()
for playlist in self.manifest.playlists:
video_layouts = [
(raw.get("stream_info") or {}).get("req_video_layout")
for raw in (getattr(self.manifest, "data", None) or {}).get("playlists", [])
]
for playlist, video_layout in zip(self.manifest.playlists, video_layouts):
video_codec: Optional[Video.Codec] = None
primary_track_type: type[Audio] | type[Video]
try:
@@ -272,6 +276,9 @@ class HLS:
(playlist.stream_info.closed_captions or "").strip('"'), []
),
dv_compatible_bitstream=dv_compatible_bitstream,
stereo=Video.is_stereo(
playlist.stream_info.codecs, supplemental_codecs.get(playlist.uri), video_layout
),
)
if primary_track_type is Video
else {}
+81 -34
View File
@@ -77,7 +77,15 @@ RATE_LIMIT_TRIES = 3
TUNNEL_SLOT_POLL = 15
REOPEN_WAIT = 1800.0 # seconds a lost session waits for its server to come back; covers a deploy's whole drain
_recovering = local()
class _Mute(local):
"""This thread's RemoteError log state: inside ``session_call`` (``recovering``) and under ``quiet()``."""
recovering = False
quiet = False
_mute = _Mute()
T = TypeVar("T")
@@ -85,11 +93,22 @@ T = TypeVar("T")
@contextlib.contextmanager
def recovering() -> Iterator[None]:
"""Mark this thread's remote calls as recoverable, so a RECOVERABLE error logs at debug level."""
_recovering.on = True
_mute.recovering = True
try:
yield
finally:
_recovering.on = False
_mute.recovering = False
@contextlib.contextmanager
def quiet() -> Iterator[None]:
"""Log this thread's RemoteErrors at debug level, because the caller reports them."""
before = _mute.quiet
_mute.quiet = True
try:
yield
finally:
_mute.quiet = before
def redact_secrets(text: str, data: Optional[Dict[str, Any]] = None) -> str:
@@ -119,7 +138,7 @@ class RemoteError(SystemExit):
"""A remote call that failed: logged, then exits a CLI run like ``SystemExit(1)``.
A serve relaying for a REST client catches it instead and answers with ``message``; ``kind`` is the
remote server's error code, or CONNECT / TIMEOUT / AUTH when the session could not be reached or opened.
remote server's error code, or CONNECT / TIMEOUT / AUTH when the remote session could not be reached or opened.
(Not ``code``: SystemExit's ``code`` is the exit status.)
"""
@@ -127,12 +146,15 @@ class RemoteError(SystemExit):
super().__init__(1)
self.message = message
self.kind = kind
quiet = getattr(_recovering, "on", False) and kind in RECOVERABLE
(log.debug if quiet else log.error)(message)
muted = _mute.quiet or (_mute.recovering and kind in RECOVERABLE)
(log.debug if muted else log.error)(message)
def __str__(self) -> str:
return self.message
@property
def session_broke(self) -> bool:
"""Whether the remote session itself is gone, not just this one call."""
"""Whether the remote session itself is gone, not only this one call."""
return self.kind in SESSION_LOST
@@ -348,6 +370,7 @@ def deserialize_video(data: Dict[str, Any]) -> Video:
dv_compatible_bitstream=bool(data.get("dv_compatible_bitstream", False)),
hdr10plus=bool(data.get("hdr10plus", False)),
hdr10_base=bool(data.get("hdr10_base", False)),
stereo=bool(data.get("stereo", False)),
)
@@ -740,7 +763,8 @@ def remote_service_list(server_name: Optional[str]) -> Optional[list[dict]]:
return hit[1]
services: Optional[list[dict]] = None
try:
services = remote_client(server_name).get("/api/services").get("services") or []
with quiet():
services = remote_client(server_name).get("/api/services").get("services") or []
except (Exception, SystemExit) as e:
log.warning(f"Could not list the services of remote server '{server_name}': {e}")
_remote_lists[server_name] = (time.monotonic(), services)
@@ -1459,7 +1483,7 @@ class RemoteService:
def session_call(self, call: Callable[[Optional[str]], T]) -> T:
"""Run ``call`` with the session id, and once more on a new session when the server lost the old one."""
session_id = self._session_id
if getattr(_recovering, "on", False):
if _mute.recovering:
return call(session_id)
try:
with recovering():
@@ -1469,7 +1493,8 @@ class RemoteService:
raise
lost = e
if not self.recover(session_id, lost.kind):
log.error(lost.message)
if not _mute.quiet:
log.error(lost.message)
raise lost
return call(self._session_id)
@@ -1591,7 +1616,8 @@ class RemoteService:
return
with self._log_drain_lock:
try:
resp = self.client.get_optional(f"/api/session/{self._session_id}/logs?since={self._log_seq}")
with quiet():
resp = self.client.get_optional(f"/api/session/{self._session_id}/logs?since={self._log_seq}")
except (Exception, SystemExit):
return
if not resp:
@@ -1700,7 +1726,8 @@ class RemoteService:
try:
call = self.client.session.get(url, timeout=30).json().get("cdm_call")
if isinstance(call, dict):
self.answer_cdm_call(call, sessions)
with quiet():
self.answer_cdm_call(call, sessions)
except (requests.RequestException, ValueError, SystemExit) as e:
self.log.debug(f"CDM call poll failed: {e}")
self.close_cdm_sessions(sessions)
@@ -1892,7 +1919,9 @@ class RemoteService:
track_keys = keys_by_track.get(server_id, {})
if not track_keys:
if server_id in track_ids and server_id not in clear_tracks | set(capped):
self.log.warning(f"Server CDM returned no content keys for track {track.id}")
from unshackle.core.api.progress import track_log_label
self.log.warning(f"Server CDM returned no content keys for track {track_log_label(track)}")
continue
named = {kid.hex for drm in track.drm or [] for kid in real_kids(drm)} & track_keys.keys()
@@ -1963,7 +1992,9 @@ class RemoteService:
drm_class = other
track.drm_preference = drm_class.lower()
self.client_licensed.add(str(track.id))
self.log.info(f"{why}; licensing {track.id} with your own {drm_class} device")
from unshackle.core.api.progress import track_log_label
self.log.info(f"{why}; licensing track {track_log_label(track)} with your own {drm_class} device")
@property
def server_cdm(self) -> bool:
@@ -1990,11 +2021,12 @@ class RemoteService:
def report_bad_key(self, kid: UUID, key: str) -> None:
"""Tell the server a vault key it served did not decrypt, so it flags its own row."""
try:
self.session_call(
lambda sid: self.client.post_optional(f"/api/session/{sid}/keys/bad", {"kid": kid.hex, "key": key})
)
with quiet():
self.session_call(
lambda sid: self.client.post_optional(f"/api/session/{sid}/keys/bad", {"kid": kid.hex, "key": key})
)
except (Exception, RemoteError) as e:
self.log.warning(f"Could not report the bad content key to the server: {e!r}")
self.log.debug(f"Could not report the bad content key to the server: {e}")
@staticmethod
def create_drm_stub(drm_type: str, kid_hexes: list[str]) -> Any:
@@ -2049,13 +2081,15 @@ class RemoteService:
raw = self._chapters_by_title.get(title_id, [])
return Chapters([Chapter(ch["timestamp"], ch.get("name")) for ch in raw])
def get_widevine_license(self, *, challenge: bytes, title: Title_T, track: AnyTrack) -> Optional[Union[bytes, str]]:
return self.proxy_license(challenge, track, "widevine")
def get_widevine_license(
self, *, challenge: bytes, title: Title_T, track: AnyTrack, kid: Optional[UUID] = None
) -> Optional[Union[bytes, str]]:
return self.proxy_license(challenge, track, "widevine", kid)
def get_playready_license(
self, *, challenge: bytes, title: Title_T, track: AnyTrack
self, *, challenge: bytes, title: Title_T, track: AnyTrack, kid: Optional[UUID] = None
) -> Optional[Union[bytes, str]]:
return self.proxy_license(challenge, track, "playready")
return self.proxy_license(challenge, track, "playready", kid)
def get_clearkey_license(
self, *, challenge: bytes, title: Title_T, track: AnyTrack
@@ -2083,7 +2117,13 @@ class RemoteService:
)
return base64.b64decode(resp["license"]) if resp.get("license") else None
def proxy_license(self, challenge: Union[bytes, str], track: AnyTrack, drm_type: str) -> bytes:
def proxy_license(
self, challenge: Union[bytes, str], track: AnyTrack, drm_type: str, kid: Optional[UUID] = None
) -> bytes:
"""Relay a licence challenge to the server. A failed relay raises ClickException, so dl's licence fallbacks see it.
``kid`` is the KID the server CDM must answer; a server vault key for it stands in for the licence.
"""
if isinstance(challenge, str):
challenge = challenge.encode("utf-8")
@@ -2103,13 +2143,16 @@ class RemoteService:
if pssh_b64:
try:
request = {"drm_type": drm_type, "mode": "server_cdm", "pssh": pssh_b64}
resp = self.session_call(
lambda sid: self.client.post(
f"/api/session/{sid}/license",
{**request, "track_id": self.server_track_id(track.id)},
expect=("SERVER_CDM_CAPPED",),
if kid:
request["kid"] = kid.hex
with quiet():
resp = self.session_call(
lambda sid: self.client.post(
f"/api/session/{sid}/license",
{**request, "track_id": self.server_track_id(track.id)},
expect=("SERVER_CDM_CAPPED",),
)
)
)
if resp.get("error_code") == "SERVER_CDM_CAPPED":
refusal = resp.get("details") or {}
self.server_cdm_type = resp.get("drm_type", server_type)
@@ -2140,11 +2183,14 @@ class RemoteService:
payload["pssh"] = pssh_b64
try:
resp = self.session_call(
lambda sid: self.client.post(
f"/api/session/{sid}/license", {**payload, "track_id": self.server_track_id(track.id)}
with quiet():
resp = self.session_call(
lambda sid: self.client.post(
f"/api/session/{sid}/license", {**payload, "track_id": self.server_track_id(track.id)}
)
)
)
except RemoteError as e:
raise click.ClickException(e.message) from None
finally:
self.drain_server_logs()
return base64.b64decode(resp["license"])
@@ -2171,7 +2217,8 @@ class RemoteService:
session_id, self._session_id = self._session_id, None
if session_id:
try:
result = self.client.delete(f"/api/session/{session_id}")
with quiet():
result = self.client.delete(f"/api/session/{session_id}")
profile = self.ctx.parent.params.get("profile") if self.ctx.parent else None
self.save_returned_cache(result.get("cache", {}), profile)
except SystemExit:
+2
View File
@@ -106,6 +106,7 @@ class Title:
"video": "",
"hdr": "",
"hfr": "",
"3d": "",
"edition": "",
"lang_tag": "",
"title_type": {"Movie": "movie", "Episode": "series", "Song": "music"}.get(
@@ -163,6 +164,7 @@ class Title:
frame_rate = float(primary_video_track.frame_rate) if primary_video_track.frame_rate else 0.0
context["hfr"] = "HFR" if frame_rate > 30 else ""
context["3d"] = "3D" if any(v.stereo for v in self.tracks.videos[:1]) else ""
if primary_audio_track:
codec = primary_audio_track.format
+1 -11
View File
@@ -30,6 +30,7 @@ from unshackle.core.downloaders.requests import requests
from unshackle.core.drm import DRM_T, BlowfishStripe, ClearKeyCENC, PlayReady, Widevine
from unshackle.core.drm.verify import (
decrypt_track,
find_child_box,
has_encrypted_sample_entry,
read_top_level_box,
senc_protects_samples,
@@ -114,17 +115,6 @@ def iter_top_level_boxes(path: Path) -> Iterable[tuple[bytes, int, int]]:
offset += size
def find_child_box(buf: bytes, box_type: bytes) -> Optional[bytes]:
"""Return the first box of the given type nested anywhere in buf, header included."""
i = buf.find(box_type, 4)
while i != -1:
size = int.from_bytes(buf[i - 4 : i], "big")
if 8 <= size <= len(buf) - (i - 4):
return buf[i - 4 : i - 4 + size]
i = buf.find(box_type, i + 1)
return None
def strip_duplicate_init_boxes(src: Path, dst: Path) -> int:
"""Write src to dst without its repeated ftyp/moov pairs, returning how many it dropped.
+49 -20
View File
@@ -21,6 +21,7 @@ from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.console import GradientPulseBarColumn, console
from unshackle.core.constants import AnyTrack, TrackT
from unshackle.core.drm.verify import children, has_second_view, read_top_level_box
from unshackle.core.events import events
from unshackle.core.tracks.attachment import Attachment
from unshackle.core.tracks.audio import Audio
@@ -68,6 +69,11 @@ def file_track_ids(path: Path) -> dict[str, int]:
return ids or {"video": 0, "audio": 1}
def mp4_track_count(moov: Optional[bytes]) -> int:
"""How many tracks ``MP4Box -add`` takes from a file: each ``trak`` of its moov, or 1 for a file with none."""
return sum(1 for kind, _, _ in children(moov or b"", 8) if kind == b"trak") or 1
class Tracks:
"""
Video, Audio, Subtitle, Chapter, and Attachment Track Store.
@@ -287,7 +293,17 @@ class Tracks:
def signature(track: AnyTrack) -> tuple:
"""The attributes that listing and selection read. Two tracks with one signature look and select the same."""
if isinstance(track, Video):
return Video, track.codec, track.range, track.width, track.height, track.fps, track.bitrate, track.name
return (
Video,
track.codec,
track.range,
track.stereo,
track.width,
track.height,
track.fps,
track.bitrate,
track.name,
)
if isinstance(track, Audio):
return (
Audio,
@@ -333,11 +349,13 @@ class Tracks:
def sort_videos(
self, by_language: Optional[Sequence[Union[str, Language]]] = None, exact_match: bool = False
) -> None:
"""Sort video tracks by resolution then bitrate, and optionally language."""
"""Sort video tracks by resolution, 2D before 3D, then bitrate, and optionally language.
A 3D track carries two views, so its higher bitrate would otherwise rank it over the 2D one.
"""
if not self.videos:
return
# resolution first, then bitrate (unknown-bitrate tracks still rank by resolution)
self.videos.sort(key=lambda x: (x.height or 0, float(x.bitrate or 0.0)), reverse=True)
self.videos.sort(key=lambda x: (x.height or 0, not x.stereo, float(x.bitrate or 0.0)), reverse=True)
for language in reversed(by_language or []):
if str(language) in ("all", "best"):
language = self._original_language(self.videos, exact_match)
@@ -486,10 +504,10 @@ class Tracks:
HDR10/HDR10+/DV tracks are hybrid ingredients. They only enter the standalone
pool when the user explicitly requested their range alongside HYBRID, so for
example `-r HYBRID` muxes only the hybrid while `-r HYBRID,HDR10P` also delivers
HDR10+.
HDR10+. A 3D track is never an ingredient: dovi_tool does not handle a second view.
"""
ingredient_ranges = (Video.Range.HDR10, Video.Range.HDR10P, Video.Range.DV)
hybrid_candidates = [v for v in videos if v.range in ingredient_ranges]
hybrid_candidates = [v for v in videos if v.range in ingredient_ranges and not v.stereo]
non_hybrid = [v for v in videos if v.range not in ingredient_ranges or v.range in non_hybrid_ranges]
return hybrid_candidates, non_hybrid
@@ -654,20 +672,23 @@ class Tracks:
"""
Multiplex the Tracks into an MP4 Container with MP4Box.
Only used for a codec Matroska cannot carry, DTS-UHD or MPEG-H, because MP4 carries
far less: it drops the forced, hearing-impaired and original flags, holds no
attachments at all, and leaves out a subtitle it has no format for rather than
mangling it. Losing those beats shipping audio no player can name.
Only used for what Matroska cannot carry: DTS-UHD or MPEG-H audio, and stereoscopic 3D
(MV-HEVC) video, whose second view mkvmerge drops. MP4 carries far less: it drops the
forced, hearing-impaired and original flags, holds no attachments at all, and leaves
out a subtitle it has no format for rather than mangling it. Losing those beats
shipping audio no player can name or a 3D film as 2D. ``-for-test`` drops the dates and
the GPAC version tag; ``tkhd`` restores the flags it clears on a non-MP4 input.
Returns the same (path, returncode, errors) shape as the Matroska mux.
"""
mp4box = getattr(binaries, "MP4Box", None)
if not mp4box:
raise RuntimeError("MP4Box (GPAC) not found, muxing DTS-UHD or MPEG-H audio needs it")
raise RuntimeError("MP4Box (GPAC) not found, muxing DTS-UHD or MPEG-H audio or 3D video needs it")
progress = progress or partial(lambda **kwargs: None)
cl = [str(mp4box)]
stereo = any(v.stereo for v in self.videos)
cl = [str(mp4box), "-for-test"]
sample_base = self.sample_base
names: list[str] = []
cleaned: list[Path] = []
@@ -685,7 +706,6 @@ class Tracks:
)
continue
events.emit(events.Types.TRACK_MULTIPLEX, track=track)
track_id += 1
source = track.path
candidate = config.directories.temp / f"mp4mux_{track.id}_{os.getpid()}_{threading.get_ident()}.mp4"
@@ -703,10 +723,13 @@ class Tracks:
candidate.unlink()
delay = round((track.sample_start - sample_base) * 1000) if config.sample else 0
cl.extend(["-add", f"{source}:lang={track.mux_language}" + (f":delay={delay}" if delay else "")])
moov = read_top_level_box(source, b"moov")
flags = "" if moov else ":tkhd=enable,movie,preview"
cl.extend(["-add", f"{source}:lang={track.mux_language}{flags}" + (f":delay={delay}" if delay else "")])
name = "" if isinstance(track, Video) else (track.get_track_name() or "")
if name:
names.extend(["-name", f"{track_id}={name}"])
names.extend(["-name", f"{track_id + 1}={name}"])
track_id += mp4_track_count(moov)
if not track_id:
raise ValueError("No tracks provided, at least one track must be provided.")
@@ -719,7 +742,7 @@ class Tracks:
title=sanitize_filename(title), random=f"{self.chapters.id}_{os.getpid()}_{threading.get_ident()}"
)
self.chapters.dump(chapters_path, fallback_name=config.chapter_fallback_name)
cl.extend(["-chap", str(chapters_path)])
cl.extend(["-chapqt" if stereo else "-chap", str(chapters_path)])
if self.attachments:
log_event(
@@ -749,7 +772,9 @@ class Tracks:
"muxer": str(mp4box),
"command": full_command,
"container": "mp4",
"reason": "DTS-UHD or MPEG-H audio has no Matroska CodecID",
"reason": "3D video has no Matroska mapping"
if stereo
else "DTS-UHD or MPEG-H audio has no Matroska CodecID",
},
)
@@ -778,12 +803,12 @@ class Tracks:
returncode = p.wait()
if returncode < 2 and any(MP4BOX_OPEN_FAILED.search(line) for line in errors):
returncode = 2
if config.sample and returncode == 0:
if config.sample and returncode == 0 and not stereo:
start, end = (seconds - sample_base for seconds in config.sample)
cut_path = output_path.with_suffix(".cut.mp4")
try:
run_step(
[mp4box, "-splitx", f"{start:.3f}:{end:.3f}", output_path, "-out", cut_path],
[mp4box, "-for-test", "-splitx", f"{start:.3f}:{end:.3f}", output_path, "-out", cut_path],
output=cut_path,
label="MP4Box sample cut",
)
@@ -791,6 +816,8 @@ class Tracks:
except RuntimeError as e:
returncode = 2
errors.append(f"MP4Box could not cut the sample window: {e}")
if stereo and returncode == 0 and not has_second_view(output_path):
logging.getLogger("Tracks").warning(f"The 3D mux of {output_path.name} lost its second view")
mux_duration_ms = round((time.monotonic() - mux_start_time) * 1000, 1)
output_size = output_path.stat().st_size if output_path.exists() else 0
@@ -868,7 +895,9 @@ class Tracks:
video.needs_repack = True
video.data["audio_language"] = lang_code
if any(at.path and at.path.exists() and has_mp4_only_audio_entry(at.path) for at in self.audio):
if any(v.stereo for v in self.videos) or any(
at.path and at.path.exists() and has_mp4_only_audio_entry(at.path) for at in self.audio
):
return self._mux_mp4(
title=title,
delete=delete,
+41 -4
View File
@@ -15,6 +15,7 @@ from langcodes import Language
from unshackle.core import binaries
from unshackle.core.config import config
from unshackle.core.drm.verify import has_second_view
from unshackle.core.tracks.subtitle import Subtitle
from unshackle.core.tracks.track import Track
from unshackle.core.utilities import FPS, get_boxes, log_event
@@ -228,6 +229,7 @@ class Video(Track):
dv_compatible_bitstream: bool = False,
hdr10plus: bool = False,
hdr10_base: bool = False,
stereo: bool = False,
**kwargs: Any,
) -> None:
"""
@@ -248,6 +250,7 @@ class Video(Track):
All strings will be cast to either a number or float.
hdr10plus: The DV track also carries HDR10+ metadata. The HDR10+ probe sets it.
hdr10_base: The DV track has an HDR10 base layer, as DV profile 8.1 does.
stereo: The track is stereoscopic 3D (MV-HEVC), with a second view for the other eye.
Note: If codec, bitrate, width, height, or fps is not specified, unshackle can
skip some checks or assume a value. Give as much information as possible.
@@ -298,6 +301,7 @@ class Video(Track):
self.dv_compatible_bitstream = dv_compatible_bitstream
self.hdr10plus = hdr10plus
self.hdr10_base = hdr10_base
self.stereo = stereo
self.hybrid_base_only = False
self.native_hybrid = False
@@ -315,6 +319,7 @@ class Video(Track):
"dv_compatible_bitstream": self.dv_compatible_bitstream,
"hdr10plus": self.hdr10plus,
"hdr10_base": self.hdr10_base,
"stereo": self.stereo,
}
)
return data
@@ -334,6 +339,7 @@ class Video(Track):
dv_compatible_bitstream=data.get("dv_compatible_bitstream", False),
hdr10plus=data.get("hdr10plus", False),
hdr10_base=data.get("hdr10_base", False),
stereo=data.get("stereo", False),
)
@staticmethod
@@ -347,6 +353,16 @@ class Video(Track):
"""Whether a codecs list names a Dolby Vision codec."""
return bool(Video.dv_fourccs(codecs))
@staticmethod
def is_stereo(
codecs: Optional[str], supplemental_codecs: Optional[str] = None, video_layout: Optional[str] = None
) -> bool:
"""Whether a track is stereoscopic 3D: the HLS REQ-VIDEO-LAYOUT names CH-STEREO, or a codec is DV profile 20."""
tokens = re.split(r"[\s,]+", f"{codecs or ''},{supplemental_codecs or ''}".lower())
return "CH-STEREO" in (video_layout or "").upper() or any(
re.match(r"dvh[1e]\.20(\.|$)", token) for token in tokens
)
@staticmethod
def is_dv_composite(
codec: Optional[Video.Codec], codecs: Optional[str], supplemental_codecs: Optional[str]
@@ -377,12 +393,15 @@ class Video(Track):
return f"{label}+DV" if self.dv_compatible_bitstream and self.range != Video.Range.DV else label
def __str__(self) -> str:
labels = filter(
None, [self.codec.value if self.codec else None, self.range_label, "3D" if self.stereo else None]
)
return " | ".join(
filter(
None,
[
"VID",
"[" + (", ".join(filter(None, [self.codec.value if self.codec else None, self.range_label]))) + "]",
f"[{', '.join(labels)}]",
str(self.language),
", ".join(
filter(
@@ -459,16 +478,34 @@ class Video(Track):
self.path = output_path
original_path.unlink()
def download(self, *args: Any, **kwargs: Any) -> None:
"""Download the track and mark it 3D when the file holds a second view; a 3D track is never repacked."""
super().download(*args, **kwargs)
if not self.stereo and self.path and has_second_view(self.path):
self.stereo = True
if self.stereo:
self.needs_repack = False
def repackage(self, bsf_v: Optional[str] = None) -> bool:
"""Remux with FFmpeg, except a 3D track: an FFmpeg remux rewrites its stereo and DV boxes."""
if self.stereo:
logging.getLogger("Video").warning(
f"Skipped the FFmpeg repack of 3D track {self.id}: FFmpeg rewrites its stereo boxes"
)
return False
return super().repackage(bsf_v)
def vui_bsf(self) -> Optional[str]:
"""Return the ``-bsf:v`` value that rewrites SPS VUI colour metadata to match ``self.range``.
Returns None when a rewrite is not necessary: SDR/DV/HYBRID ranges, non-AVC/HEVC codecs, a
missing file, or when the bitstream already ships the correct colour metadata. Otherwise
Returns None when a rewrite is not necessary or not safe: SDR/DV/HYBRID ranges, non-AVC/HEVC
codecs, a 3D track (an FFmpeg remux rewrites its stereo and DV boxes), a missing file, or when the
bitstream already ships the correct colour metadata. Otherwise
returns a ``{h264,hevc}_metadata=...`` string suitable for FFmpeg ``-bsf:v``.
"""
if not self.path or not self.path.exists():
return None
if self.codec not in (Video.Codec.AVC, Video.Codec.HEVC):
if self.codec not in (Video.Codec.AVC, Video.Codec.HEVC) or self.stereo:
return None
if self.range in (Video.Range.SDR, Video.Range.DV, Video.Range.HYBRID):
return None
+4 -4
View File
@@ -22,12 +22,12 @@ tag: "NOGRP"
# ? suffix, e.g. {year?}.
output_template:
# Scene-style naming (dot-separated)
movies: "{title}.{year?}.{repack?}.{edition?}.{lang_tag?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
series: "{title}.{year?}.{season_episode}.{episode_name?}.{repack?}.{edition?}.{lang_tag?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
movies: "{title}.{year?}.{repack?}.{edition?}.{lang_tag?}.{3d?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
series: "{title}.{year?}.{season_episode}.{episode_name?}.{repack?}.{edition?}.{lang_tag?}.{3d?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
songs: "{track_number}.{title}.{repack?}.{edition?}.{lang_tag?}.{source?}.WEB-DL.{audio_full}.{atmos?}-{tag}"
folder:
movies: "{title}.{year?}.{repack?}.{edition?}.{lang_tag?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
series: "{title}.{year?}.{repack?}.{edition?}.{lang_tag?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
movies: "{title}.{year?}.{repack?}.{edition?}.{lang_tag?}.{3d?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
series: "{title}.{year?}.{repack?}.{edition?}.{lang_tag?}.{3d?}.{quality}.{source}.WEB-DL.{dual?}.{multi?}.{dubbed?}.{audio_full}.{atmos?}.{hdr?}.{hfr?}.{video}-{tag}"
albums: "{album_artist} - {album} ({year?})"
#