Compare commits

...
4 Commits
Author SHA1 Message Date
imSp4rky f86359de5d feat(serve): reload unshackle.yaml on SIGHUP without a restart
Extract the serve validation and app config builder so startup and reload share them. A bad file or a restart-only change keeps the running config; open CDM sessions survive.
2026-10-06 17:56:47 -06:00
imSp4rky eea4b41b70 feat(serve): log CDM calls for the dashboard
Add GET /api/dashboard/cdm-logs: one record per pywidevine/pyplayready call plus open sessions per device, with no key or licence material.
2026-10-06 14:58:08 -06:00
imSp4rky 94e5264396 fix(cdm): send a PlayReady licence without an XML declaration as XML
A remote CDM tried base64 first on any licence not starting with "<?xml", which can decode XML text into garbage. Also lists L3 as a device_name.
2026-10-06 13:59:40 -06:00
imSp4rky 3477d09cc2 feat(dl): add --audio-description-only and --forced-subs-only
Each keeps one kind of track inside its track type. A run of more than one title skips a title that has no such track. The REST API rejects a non-boolean value for a boolean dl param, and the DASH parser reads audio description from Role elements.
2026-10-06 12:45:52 -06:00
28 changed files with 1500 additions and 165 deletions
+47
View File
@@ -47,6 +47,7 @@ All are `GET`.
| `/api/dashboard/sessions` | Every live remote session: `id`, `owner` (username or masked key), `creator_ip`, `service`, `title_id`, `titles`, `tracks`, `title` (the display name of the title the client is on: the last one it asked tracks for, else the first resolved title), `auth_status`, `auth_error`, `server_account` (the server profile lent to the remote session, else `null`), `server_cdm` (`false` when the client's own device licenses the remote session), `server_cdm_max_height` (the tallest video the server CDM licenses live, else `null`), `client`, `actions`, `created_at` and `last_accessed` (ISO 8601) with `created_ts` and `last_accessed_ts` (Unix epoch, the same clock as log `ts`), `age_seconds`, `idle_seconds`, `log_seq`. `client` is whatever the remote client sent as `client` in its session create request. The CLI sends `version`, `code_hash` (the commit the client runs, `null` when its source cannot be read), `platform` and `argv`. `argv` is the command line the user ran: proxy and URL userinfo, secret query parameters and credential values become `***`, home and install paths shorten as in the logs, and the line is cut at 3000 characters. An older client sends less or nothing and the field is `{}`. `log_seq` is the last sequence number in the remote session's service log - poll `/api/dashboard/sessions/{id}/logs` when it changes. `actions` is the session's request log, newest last, capped at 500: `ts`, `method`, `action` (`titles`, `tracks`, `segments`, `license`, `prompt`, …), `query`, `status`, `ms`, `bytes_in`, `bytes_out`. |
| `/api/dashboard/jobs` | Every download job with full detail, regardless of owner. Empty in `--remote-only` mode. |
| `/api/dashboard/logs` | `{"seq": N, "records": [...]}`: the last 1000 log records. `?since=<seq>` returns only newer records; `?level=WARNING` sets the minimum level; `?logger=serve` keeps one logger and its children (`aiohttp.access` is the noisiest). `ts` is a Unix epoch in seconds. |
| `/api/dashboard/cdm-logs` | `{"seq": N, "records": [...], "sessions": [...]}`: the last 5000 calls to the Widevine and PlayReady CDM device routes, and the open sessions of each device. `?since=<seq>` returns only newer records. See [CDM calls](#cdm-calls). |
| `/api/dashboard/sessions/{id}/logs` | One remote session's service log: `{"session_id", "last_seq", "records": [...]}`. Each record has `seq`, `ts`, `level`, `message`. `?since=<seq>` returns only newer records. 404 when the remote session is unknown. |
| `/api/dashboard/keys` | Every configured API key: what it may do and what it has done. One row per key in `serve.users`, plus `serve.api_secret` when it is set. |
| `/api/dashboard/services` | Every service configured on this server and its load state, including the ones that failed to import. See [Services](#services) for what configured means. |
@@ -113,6 +114,52 @@ never keeps it alive or makes it look active.
publishes nothing. Poll this route when `log_seq` changes, and on a timer while a drawer
is open.
## CDM calls
A server in `full` mode also mounts the Widevine CDM device routes at `/{device}/...` and the
PlayReady ones at `/playready/{device}/...`. The server records each call to these routes in
a 5000-record buffer of its own, apart from the log records of `/api/dashboard/logs`.
```
GET /api/dashboard/cdm-logs?since=1230
{"seq": 1231,
"records": [{"seq": 1231, "ts": 1759750000.123, "drm": "widevine", "op": "get_license_challenge",
"device": "chrome_l3", "key_id": "ab12cd34ef56", "user": "tier1",
"session_id": "0a1b2c3d4e5f60718293a4b5c6d7e8f9",
"status": 200, "ok": true, "error": null, "ms": 12.4}],
"sessions": [{"drm": "widevine", "device": "chrome_l3", "key_id": "ab12cd34ef56", "user": "tier1",
"open": 3, "max": 17}]}
```
`since` and `seq` work as they do for `/api/dashboard/logs`: `records` holds the calls with a
`seq` above `since`, oldest first. `seq` starts again at 1 when the server restarts, so a `seq`
below your cursor means a restart: read again from 0.
| Field | Meaning |
|-------|---------|
| `ts` | Unix epoch in seconds, when the call arrived. |
| `drm` | `widevine`, or `playready` for a route below `/playready/`. |
| `op` | `open`, `close`, `set_service_certificate`, `get_service_certificate`, `get_license_challenge`, `parse_license` or `get_keys`. The licence type or content key type that follows in the path is not part of it. |
| `device` | The device name from the path, cut at 255 characters. |
| `key_id`, `user` | The `id` and `label` of the caller's row in `/api/dashboard/keys`. Both are `anonymous` when the API key is missing or unknown. |
| `session_id` | The CDM session, lowercase hexadecimal, so the calls of one licence exchange group together. `null` when there is none, for example a refused `open`. |
| `status`, `ok` | The HTTP status of the reply. `ok` is `true` below 400. |
| `error` | `null` when `ok`. Otherwise the `message` of the reply, cut at 300 characters. A handler that raised an error gives the error's class name only. |
| `ms` | Time the server took to answer, in milliseconds. |
A record never holds the API key, `init_data`, a challenge, a certificate, a licence body or a
content key. A refused call is in the log too, a 401 for a missing or unknown API key included.
`sessions` is a snapshot at read time, not a history. It has one row for each device an API
key has opened since the server started: `open` is the number of sessions open now, and `max`
is the number of sessions the device permits at one time. A PlayReady session older than the
library's session timeout does not count: the library drops it on the next `open`. The
list is `[]` until the first `open`, and on a server that mounts no CDM device routes.
!!! note "Polled only"
The event stream does not carry these records. One licence exchange is five calls, and that
volume would push `session` events out of the stream's history.
## Keys
```
+2
View File
@@ -435,12 +435,14 @@ Make a download job. It requires `service` and `title_id`. Every other field is
| `require_video` | string[] | `[]` | Video languages that must exist. The job fails if one is missing, even with `best_available`. |
| `require_subs` | string[] | `[]` | Subtitle languages that must exist. The job fails if one is missing. `s_lang` still decides what to keep. |
| `forced_subs` | boolean | `false` | Include forced subtitles. |
| `forced_subs_only` | boolean | `false` | Download forced subtitles and no other subtitle; implies `forced_subs`. Not valid with `no_subs`, `video_only`, `audio_only` or `chapters_only`. |
| `forced_s_lang` | string[] | `[]` | Forced subtitle language(s); implies `forced_subs`. |
| `exact_lang` | boolean | `false` | Exact language matching. |
| `sub_format` | string | `null` | Output subtitle format. |
| `video_only` / `audio_only` / `subs_only` / `chapters_only` | boolean | `false` | Track-type restriction (at most one). |
| `no_subs` / `no_audio` / `no_chapters` / `no_video` / `no_attachments` | boolean | `false` | Skip a track type. |
| `audio_description` | boolean | `false` | Download audio description. |
| `audio_description_only` | boolean | `false` | Download audio description and no standard audio. Not valid with `no_audio`, `video_only`, `subs_only` or `chapters_only`. |
| `slow` | boolean/string | `null` | Randomized delay between downloads. |
| `split_audio` | boolean | `null` | Separate files per audio codec. |
| `skip_dl` | boolean | `false` | Only fetch keys, do not download. |
+3 -2
View File
@@ -438,8 +438,9 @@ server drops an unknown codec name. An empty array is a valid value and stays
empty.
The server does not forward the `dl` mode flags: `video_only`, `audio_only`,
`subs_only`, `chapters_only`, `list_`, `skip_dl`, `sub_format` and
`no_attachments`. A service that reads one of them to skip work fetches more on
`subs_only`, `chapters_only`, `list_`, `skip_dl`, `sub_format`,
`no_attachments`, `audio_description`, `audio_description_only` and
`forced_subs_only`. A service that reads one of them to skip work fetches more on
the server, but the result is the same.
In a client-login session (not a server account), the server gives the service
+2
View File
@@ -102,6 +102,7 @@ unshackle dl [OPTIONS] SERVICE [SERVICE ARGS...]
| `--require-video` | - | Video langs that must exist. The title fails even with `--best-available`. Does not select tracks. |
| `--require-subs` | - | Subtitle langs that must exist. Fails the title if one is missing. `--s-lang` still decides what to keep. |
| `-fs`, `--forced-subs` | off | Include forced subtitle tracks. |
| `-fso`, `--forced-subs-only` | off | Keep only forced subtitle tracks; drop the other subtitles. Implies `-fs`. |
| `-fsl`, `--forced-s-lang` | none | Language(s) wanted for forced subtitles; implies `-fs`. A `-` prefix excludes. |
| `--exact-lang` | off | Exact matching only: `-l es-419` matches `es-419`, not `es-ES`. Applies to selection and to sort order. |
| `--sub-format` | - | Output subtitle format (`SRT`/`srt`, `VTT`/`webvtt`, `ASS`/`ssa`, `TTML`, `SMI`, ...), or `original` to keep the source format. |
@@ -139,6 +140,7 @@ Keep only certain track types, or skip certain track types.
| Flag | Description |
|---|---|
| `-ad`, `--audio-description` | Include descriptive (audio-description) tracks. |
| `-ado`, `--audio-description-only` | Keep only descriptive (audio-description) tracks; drop the standard audio. Does not need `-ad`. |
| `--skip-subtitle-errors` | Skip a failed subtitle instead of aborting the title. Video/audio failures remain fatal. |
### Output, muxing & files
+22
View File
@@ -416,6 +416,8 @@ video track counts. `--no-audio` and `--no-video` drop the matching requirement.
`-fs`). Works independently of `--s-lang`, so `-sl all -fsl en` grabs every full
subtitle but only the English forced track. It accepts exclusions too: `-fsl all,-es`
keeps every forced subtitle except the Spanish one, and `-fsl -es` means the same.
- `-fso` / `--forced-subs-only`: keep only the forced subtitle tracks and drop every
other subtitle (implies `-fs`). `--s-lang` and `-fsl` still select the languages.
- `--sub-format`: set the output subtitle format, converting only when necessary.
Accepts codec names/values and common aliases (`srt`, `vtt`, `ass`, `ssa`, `ttml`,
and the other codec aliases), or the literal `original` to keep the source format.
@@ -637,9 +639,29 @@ Additional track-type flags:
- `-ad` / `--audio-description`: include descriptive (audio-description) tracks, which
unshackle drops by default.
- `-ado` / `--audio-description-only`: keep only the descriptive tracks and drop the
standard audio. It does not need `-ad`. Add `-A` to download no video or subtitles.
- `--skip-subtitle-errors`: if a subtitle fails to download, skip it and continue rather
than aborting the whole title. Video and audio failures remain fatal.
!!! note "The `-only` kind flags"
`-ado` and `-fso` each keep one kind of track inside its track type. They do not drop
the other track types: `-ado` still downloads the video, and `-fso` still downloads
video and audio. Add `-A` or `-S` for that.
- The language flags still apply. With the default `-l orig`, `-ado` fails when the
descriptive track is not in the original language. Select the language with `-al`.
- A flag that drops the same tracks is an error, for example `-ado` with `--no-audio`
or `-fso` with `-A`.
- A title with no track of that kind stops a run of one title. A run of more than one
title skips that title with a warning and continues.
- unshackle reads the kind from the manifest or from the service. A track that
neither of them marks as descriptive or forced does not match.
```shell title="Audio description only, in English"
unshackle dl -A -ado -al en EXAMPLE 81234567
```
```shell title="Subtitles only"
unshackle dl -S -sl en EXAMPLE 81234567
```
+1 -1
View File
@@ -336,7 +336,7 @@ includes an intelligent key-caching layer:
remote_cdm:
- name: decryptlabs
type: decrypt_labs
device_name: L1 # ChromeCDM, L1, L2 (Widevine) or SL2, SL3 (PlayReady)
device_name: L1 # ChromeCDM, L1, L2, L3 (Widevine) or SL2, SL3 (PlayReady)
# host: https://keyxtractor.decryptlabs.com # optional, this is the default
# secret: ... # optional if decrypt_labs_api_key is set globally
+12
View File
@@ -74,6 +74,7 @@ Subtitle selection happens through `dl` flags. The most important is language se
| `-sl`, `--s-lang` | Language(s) wanted for subtitles. Defaults to `all`. A `-` prefix excludes, e.g. `all,-es`. |
| `--require-subs` | Require these languages to exist; if present, download **all** subtitles. Cannot be combined with `--s-lang`. |
| `-fs`, `--forced-subs` | Include forced subtitle tracks (excluded by default). |
| `-fso`, `--forced-subs-only` | Keep only forced subtitle tracks and drop the other subtitles; implies `-fs`. |
| `-fsl`, `--forced-s-lang` | Language(s) wanted for forced subtitles; implies `-fs`. A `-` prefix excludes. |
| `--exact-lang` | Exact language matching, with no regional variants. |
| `-S`, `--subs-only` | Download only subtitle tracks. |
@@ -138,6 +139,17 @@ If you pass both flags, `-fsl` wins and keeps only forced tracks in its language
It also understands `orig` (the title's original language) and `all` (every forced
track, same as plain `-fs`).
To download the forced tracks and no other subtitle, use `-fso` / `--forced-subs-only`.
It implies `-fs`, and `--s-lang` and `-fsl` still select the languages:
```shell title="Only the English forced track"
unshackle dl --s-lang en --forced-subs-only EXAMPLE 81234567
```
`-fso` does not drop video or audio. Add `-S` for a subtitle-only download. A title with
no forced track stops a run of one title. A run of more than one title skips that title
with a warning.
## Track types: forced, SDH, and CC
Every subtitle track carries flags describing what kind of captions it holds. These
+3
View File
@@ -43,6 +43,9 @@ Common config keys. This is a useful subset, and every `dl` flag works:
| `sub_format` | str | *(unset)* | Convert subtitles to this format (`srt`, `vtt`, `original`, ...). |
| `forced_subs` | bool | `false` | Include forced subtitle tracks. |
| `forced_s_lang` | list | `[]` | Forced subtitle language(s); implies `forced_subs`. A `-` prefix excludes. |
| `forced_subs_only` | bool | `false` | Keep only forced subtitle tracks; implies `forced_subs`. |
| `audio_description` | bool | `false` | Include descriptive (audio-description) tracks. |
| `audio_description_only` | bool | `false` | Keep only descriptive (audio-description) tracks. |
| `no_subs` / `no_audio` / `no_chapters` / `no_attachments` | bool | `false` | Skip that track type. `no_attachments` also skips attaching subtitle fonts. |
| `downloads` | int | `1` | Tracks downloaded concurrently. |
| `workers` | int | *(downloader default)* | Threads per track. |
+1 -1
View File
@@ -117,7 +117,7 @@ styles: `Device Type`/`device_type`, `System ID`/`system_id`, `Security Level`/`
| Field | Default | Notes |
|-------|---------|-------|
| `host` | `https://keyxtractor.decryptlabs.com` | |
| `device_name` | `ChromeCDM` | `ChromeCDM`, `L1`, `L2` (Widevine) or `SL2`, `SL3` (PlayReady) |
| `device_name` | `ChromeCDM` | `ChromeCDM`, `L1`, `L2`, `L3` (Widevine) or `SL2`, `SL3` (PlayReady). The host decides which of these names it serves. |
| `secret` | from [`decrypt_labs_api_key`](misc.md#external-api-keys) | Sent as the `decrypt-labs-api-key` header. An error is raised if neither is set. |
| `system_id` | `26830` (Widevine), `0` (PlayReady) | |
| `security_level` | Widevine `3`; PlayReady `2000` for `SL2`, else `3000` | |
+10
View File
@@ -155,6 +155,16 @@ Each entry under `users` uses that user's API key as its name, and can set its o
A user with no `playready_devices` config key gets no PlayReady access at all, not the global
list.
!!! info "Reload `serve` settings without a restart"
Send `SIGHUP` to a running server (`systemctl reload` with
`ExecReload=kill -HUP $MAINPID` in the unit) after you edit the file. The server reads
the whole file again, globs the device directories again, and swaps in the new `users`,
`tiers`, `services` and device lists in one step. Open CDM sessions and remote sessions
continue; an API key you removed gets `401` on its next request. A file that does not parse
or validate keeps the running settings and logs the reason. A change to `remote_only`,
`services_refresh_interval`, the first PlayReady device, or any command-line flag needs
a restart, and the log says so. Windows has no `SIGHUP`: restart the server there.
`server_cdm` decides whether the server runs the CDM licensing for that API key. Set it to
`true` to enable every service, or to a list of service tags to enable only those. It is `false`
unless the entry sets it. For a service the API key does not cover, the server tells a remote
+10
View File
@@ -37,3 +37,13 @@ def test_non_descriptive_accessibility_values(scheme: str, value: str) -> None:
def test_no_accessibility_element() -> None:
assert DASH.is_descriptive(ET.fromstring("<AdaptationSet/>")) is False
def test_descriptive_role_element() -> None:
# some manifests mark audio description with a Role and no Accessibility element
element = ET.fromstring(
'<AdaptationSet><Role schemeIdUri="urn:mpeg:dash:role:2011" value="description"/></AdaptationSet>'
)
assert DASH.is_descriptive(element) is True
main = ET.fromstring('<AdaptationSet><Role schemeIdUri="urn:mpeg:dash:role:2011" value="main"/></AdaptationSet>')
assert DASH.is_descriptive(main) is False
@@ -0,0 +1,77 @@
"""A remote CDM must send a PlayReady licence to the host as the XML it received.
A licence server may return the SOAP envelope with or without an XML declaration.
``base64.b64decode`` drops every character it cannot read, so XML text can decode
without an error and the host then receives bytes that are not a licence.
"""
from __future__ import annotations
import base64
from typing import Any
import pytest
from unshackle.core.cdm.custom_remote_cdm import CustomRemoteCDM
from unshackle.core.cdm.decrypt_labs_remote_cdm import DecryptLabsRemoteCDM
pytestmark = pytest.mark.unit
ENVELOPE = (
'<soap:Envelope xmlns:soap="http://schemas.xmlsoap.org/soap/envelope/"><soap:Body>'
"<AcquireLicenseResponse><License>AAAA</License></AcquireLicenseResponse>"
"</soap:Body></soap:Envelope>"
)
class Sent(Exception):
"""Stops ``parse_license`` after the request body has been captured."""
def decrypt_labs_cdm() -> tuple[Any, str]:
return DecryptLabsRemoteCDM(secret="x", host="https://cdm.test", device_name="SL3"), "decrypt_labs_session_id"
def custom_cdm() -> tuple[Any, str]:
cdm = CustomRemoteCDM(host="https://cdm.test", device={"name": "SL3000", "type": "PLAYREADY"})
return cdm, "remote_session_id"
def licence_sent(cdm: Any, remote_id_key: str, licence: str) -> bytes:
"""Run ``parse_license`` on a session that has a challenge, and return the licence bytes it posts."""
session_id = cdm.open()
cdm.set_pssh_b64("UFNTSA==", session_id=session_id)
cdm._sessions[session_id].update({"challenge": b"challenge", remote_id_key: "remote", "pssh": object()})
body: dict[str, Any] = {}
def capture(_url: str, json: dict[str, Any], **_: Any) -> None:
body.update(json)
raise Sent
cdm._http_session.post = capture
with pytest.raises(Sent):
cdm.parse_license(session_id, licence)
return base64.b64decode(body["license_response"])
@pytest.mark.parametrize("make_cdm", [decrypt_labs_cdm, custom_cdm])
@pytest.mark.parametrize(
"licence",
[
ENVELOPE,
'<?xml version="1.0" encoding="utf-8"?>' + ENVELOPE,
"\n " + ENVELOPE,
# Each length makes a different count of base64 characters, so one of them decodes without an error.
*(ENVELOPE.replace("AAAA", "A" * n) for n in range(5, 8)),
],
)
def test_a_playready_xml_licence_is_sent_as_it_is(make_cdm: Any, licence: str) -> None:
cdm, remote_id_key = make_cdm()
assert licence_sent(cdm, remote_id_key, licence) == licence.encode("utf-8")
@pytest.mark.parametrize("make_cdm", [decrypt_labs_cdm, custom_cdm])
def test_a_base64_licence_is_still_decoded(make_cdm: Any) -> None:
cdm, remote_id_key = make_cdm()
assert licence_sent(cdm, remote_id_key, base64.b64encode(b"binary licence").decode()) == b"binary licence"
@@ -0,0 +1,90 @@
"""The kind-only flags: --audio-description-only and --forced-subs-only.
The filters in dl.result are not extractable without executing the whole download loop, so
their wiring is pinned against the source, the same way test_required_langs_gate does it.
"""
from __future__ import annotations
import inspect
import pytest
from unshackle.commands.dl import dl
from unshackle.core.api.handlers import validate_download_parameters
from unshackle.core.utilities import kind_only_conflict
KEYS = ["audio_description_only", "forced_subs_only"]
BASE = {"service": "EXAMPLE", "title": "x"}
@pytest.mark.parametrize(
"args,only_key,base_key,expected",
[
(["-ado"], "audio_description_only", "audio_description", (True, False)),
(["-ad"], "audio_description_only", "audio_description", (False, True)),
([], "audio_description_only", "audio_description", (False, False)),
(["-fso"], "forced_subs_only", "forced_subs", (True, False)),
(["-fs"], "forced_subs_only", "forced_subs", (False, True)),
],
)
def test_only_flag_does_not_parse_as_its_base_flag(args, only_key, base_key, expected):
params = dl.cli.make_context("dl", args, resilient_parsing=True).params
assert (params[only_key], params[base_key]) == expected
@pytest.mark.parametrize(
"params,expected",
[
({"audio_description_only": True, "audio_only": True}, None),
({"audio_description_only": True}, None),
({"audio_description_only": True, "no_audio": True}, ("audio_description_only", "no_audio")),
({"audio_description_only": True, "video_only": True}, ("audio_description_only", "video_only")),
({"forced_subs_only": True, "subs_only": True}, None),
({"forced_subs_only": True, "no_subs": True}, ("forced_subs_only", "no_subs")),
({"forced_subs_only": True, "audio_only": True}, ("forced_subs_only", "audio_only")),
({"no_audio": True, "video_only": True}, None),
# the *_only track-type flags combine: -A -S keeps audio, so -ado is not dropped
({"audio_description_only": True, "audio_only": True, "subs_only": True}, None),
({"forced_subs_only": True, "subs_only": True, "audio_only": True}, None),
({"audio_description_only": True, "subs_only": True}, ("audio_description_only", "subs_only")),
],
)
def test_kind_only_conflict(params, expected):
assert kind_only_conflict(params) == expected
def test_api_rejects_a_conflict():
assert validate_download_parameters({**BASE, "audio_description_only": True, "no_audio": True}) == (
"Cannot use both audio_description_only and no_audio"
)
assert validate_download_parameters({**BASE, "audio_description_only": True, "audio_only": True}) is None
@pytest.mark.parametrize("key", [*KEYS, "audio_description", "audio_only"])
def test_api_rejects_a_non_boolean_flag(key):
# the string "false" is truthy in dl.result, so it would turn the flag on
assert validate_download_parameters({**BASE, key: "false"}) == f"{key} must be a boolean"
assert validate_download_parameters({**BASE, key: None}) is None
class TestFilterWiring:
source = inspect.getsource(dl.result)
def test_ado_runs_when_the_title_has_no_separate_audio(self):
# embedded audio is standard audio: the filter must sit before the `len(...) > 0` guard
ado = self.source.index("if keep_audio and audio_description_only:")
guard = self.source.index("if keep_audio and len(title.tracks.audio) > 0:")
assert ado < guard
def test_ado_does_not_count_embedded_audio(self):
# both the --require-audio gate and the missing-language check
assert self.source.count("keep_videos and not audio_description_only") == 2
def test_only_filters_keep_one_kind(self):
assert "title.tracks.select_audio(lambda x: x.descriptive)" in self.source
assert "title.tracks.select_subtitles(lambda x: x.forced)" in self.source
def test_fso_implies_forced_subs(self):
# services read forced_subs from the ctx params to fetch forced tracks at all
assert "if forced_s_lang or fsl_excl or forced_subs_only:" in self.source
@@ -0,0 +1,149 @@
"""Run the kind-only filters through the real dl.result() selection loop.
A fake service hands in-memory tracks to the loop, and the run stops at the first
title that reaches ``Tracks.tree``, the step right after selection. A title with no
track of the kind never gets there, so what the hook records is what the run kept.
"""
from __future__ import annotations
import inspect
import logging
from typing import Any, cast
import pytest
from unshackle.commands.dl import dl
from unshackle.core.api.handlers import DEFAULT_DOWNLOAD_PARAMS
from unshackle.core.service import Service
from unshackle.core.titles import Episode, Series
from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
class Selected(BaseException):
"""Carries the kept track ids out of the loop."""
def __init__(self, tracks: Tracks) -> None:
self.ids = {
"video": [t.id for t in tracks.videos],
"audio": [t.id for t in tracks.audio],
"subs": [t.id for t in tracks.subtitles],
}
class Harness(dl):
def __getattr__(self, name: str) -> None:
# every optional dl attribute (tmdb ids, cdm, vaults, ...) reads as unset
return None
class FakeService:
"""One episode per entry; ``ad`` adds a descriptive track, ``forced`` a forced subtitle."""
def __init__(self, episodes: dict[int, dict[str, bool]]) -> None:
self.episodes = episodes
self.session = None
def authenticate(self, cookies: Any = None, credential: Any = None) -> None:
pass
def get_titles_cached(self) -> Series:
return Series(
Episode(id_=f"show-s01-e{n:03}", service=FakeService, title="Show", season=1, number=n, language="en")
for n in self.episodes
)
def get_tracks(self, title: Episode) -> Tracks:
kinds = self.episodes[title.number]
tracks = Tracks()
tracks.add(Video(id_="v", url="https://x/v", language="en", codec=Video.Codec.AVC, range_=Video.Range.SDR))
tracks.add(Audio(id_="a", url="https://x/a", language="en", codec=Audio.Codec.AAC, bitrate=128_000))
if kinds.get("ad"):
tracks.add(
Audio(id_="ad", url="https://x/ad", language="en", codec=Audio.Codec.AAC, bitrate=1, descriptive=True)
)
tracks.add(Subtitle(id_="s", url="https://x/s", language="en", codec=Subtitle.Codec.WebVTT))
if kinds.get("forced"):
tracks.add(Subtitle(id_="sf", url="https://x/sf", language="en", codec=Subtitle.Codec.WebVTT, forced=True))
return tracks
def get_chapters(self, title: Episode) -> list:
return []
def __getattr__(self, name: str) -> Any:
if name.startswith("on_"):
# the event hooks (on_track_downloaded, ...) the loop subscribes
return lambda *a, **k: None
raise AttributeError(name)
def run(monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, episodes: dict, **flags: Any) -> Any:
"""Return the Selected ids, the exit code, or None for a run that completes."""
monkeypatch.setattr(Tracks, "tree", lambda self, *a, **k: (_ for _ in ()).throw(Selected(self)))
cmd = Harness.__new__(Harness)
cmd.log = logging.getLogger("test_kind_only")
cmd.service = "FAKE"
cmd.tmdb_searched = True
cmd.completed_files = []
kwargs: dict[str, Any] = {}
for name, param in inspect.signature(dl.result).parameters.items():
if name in DEFAULT_DOWNLOAD_PARAMS:
kwargs[name] = DEFAULT_DOWNLOAD_PARAMS[name]
elif name not in ("self", "service"):
kwargs[name] = None if param.default is inspect.Parameter.empty else param.default
kwargs.update(quality=[], vcodec=[], acodec=[], range_=[], wanted=[], cdm_only=None, downloads=1)
kwargs.update(flags)
caplog.set_level(logging.WARNING, logger="test_kind_only")
try:
cmd.result(cast(Service, FakeService(episodes)), **kwargs)
except Selected as sel:
return sel.ids
except SystemExit as exc:
return exc.code
return None
def test_ado_keeps_the_descriptive_track_and_the_other_types(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, audio_description_only=True)
assert ids == {"video": ["v"], "audio": ["ad"], "subs": ["s"]}
def test_fso_keeps_the_forced_track_and_the_other_types(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, forced_subs_only=True)
assert ids == {"video": ["v"], "audio": ["a"], "subs": ["sf"]}
def test_only_flags_with_their_type_flags_give_a_single_kind(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, audio_description_only=True, audio_only=True)
assert ids == {"video": [], "audio": ["ad"], "subs": []}
ids = run(monkeypatch, caplog, {1: {"ad": True, "forced": True}}, forced_subs_only=True, subs_only=True)
assert ids == {"video": [], "audio": [], "subs": ["sf"]}
@pytest.mark.parametrize("flag", ["audio_description_only", "forced_subs_only"])
def test_a_single_title_with_no_such_track_exits(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture, flag: str
) -> None:
assert run(monkeypatch, caplog, {1: {}}, **{flag: True}) == 1
assert "Skipping" not in caplog.text
def test_a_multi_title_run_skips_the_title_and_continues(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
ids = run(monkeypatch, caplog, {1: {}, 2: {"ad": True}}, audio_description_only=True)
assert ids["audio"] == ["ad"]
assert "Skipping Show S01E01" in caplog.text
def test_a_multi_title_run_with_every_title_skipped_completes(
monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture
) -> None:
assert run(monkeypatch, caplog, {1: {}, 2: {}}, forced_subs_only=True) is None
assert caplog.text.count("Skipping Show S01E") == 2
+9 -8
View File
@@ -5,6 +5,7 @@ import inspect
import pytest
from unshackle.commands.dl import dl
from unshackle.core.api import download_manager
from unshackle.core.api.handlers import DEFAULT_DOWNLOAD_PARAMS
from unshackle.core.utilities import missing_required_langs
@@ -102,15 +103,15 @@ def test_api_accepts_the_require_flags():
)
@pytest.mark.parametrize("key", ["require_audio", "require_video"])
def test_api_reaches_dl_result(key, monkeypatch):
@pytest.mark.parametrize(
"key,default",
[("require_audio", []), ("require_video", []), ("audio_description_only", False), ("forced_subs_only", False)],
)
def test_api_reaches_dl_result(key, default):
# download_manager builds the dl.result kwargs by hand, so a missing key is a silent drop
import inspect as _inspect
from unshackle.core.api import download_manager
source = _inspect.getsource(download_manager)
assert f'{key}=params.get("{key}", [])' in source
assert DEFAULT_DOWNLOAD_PARAMS[key] == default
assert key in inspect.signature(dl.result).parameters
assert f'{key}=params.get("{key}", {default!r})' in inspect.getsource(download_manager)
@pytest.mark.parametrize(
@@ -48,7 +48,7 @@ class TestGateWiring:
assert "require_video if keep_videos else []" in self.source
def test_gate_counts_audio_embedded_in_kept_video(self):
assert "embedded_audio_langs(title.tracks.videos, keep_videos)" in self.source
assert "embedded_audio_langs(title.tracks.videos, keep_videos and not audio_description_only)" in self.source
def test_gate_honours_exact_lang(self):
assert "missing_required_langs(required, available, title.language, exact=exact_lang)" in self.source
+301
View File
@@ -0,0 +1,301 @@
"""The CDM call log of `unshackle serve` and its `/api/dashboard/cdm-logs` route."""
from __future__ import annotations
import json
import time
from pathlib import Path
from typing import Any
import pytest
from aiohttp import web
from click.testing import CliRunner
from unshackle.commands.serve import serve
from unshackle.core.api.stats import SeqRing, cdm_calls, cdm_log_middleware, cdm_sessions, key_id
from unshackle.core.config import config
pytestmark = pytest.mark.unit
USER = {"X-Secret-Key": "tier-key"}
DASH = {"X-Secret-Key": "dash-secret"}
INIT_DATA = "AAAAaW5pdC1kYXRhLW1hcmtlcg=="
CHALLENGE_B64_PREFIX = "Y2hhbGxlbmdl"
LICENCE = "bGljZW5jZS1tYXJrZXI="
CONTENT_KEY = "00112233445566778899aabbccddeeff"
class FakeSession:
def __init__(self) -> None:
self.opened_at = time.time()
class Cdm:
"""A stand-in for both libraries' Cdm. The class name keeps the ``_Cdm__sessions`` mangling."""
MAX_NUM_OF_SESSIONS = 16
system_id = 1234
security_level = 3
def __init__(self) -> None:
self.__sessions: dict[bytes, FakeSession] = {}
@classmethod
def from_device(cls, _device: Any) -> Cdm:
return cls()
def open(self) -> bytes:
session_id = bytes([len(self.__sessions) + 1]) * 16
self.__sessions[session_id] = FakeSession()
return session_id
def close(self, session_id: bytes) -> None:
del self.__sessions[session_id]
def get_service_certificate(self, session_id: bytes) -> None:
return None
def get_license_challenge(self, session_id: bytes, **_: Any) -> Any:
return b"challenge-marker" * 40
def parse_license(self, session_id: bytes, _message: Any) -> None:
return None
def age(self, seconds: float) -> None:
for session in self.__sessions.values():
session.opened_at -= seconds
class PlayReadyCdm(Cdm):
SESSION_TIMEOUT = 30
def get_license_challenge(self, session_id: bytes, **_: Any) -> Any:
return "challenge-marker" * 40
@pytest.fixture
def serve_app(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> web.Application:
"""The app a keyed, full-mode `unshackle serve` builds, with both CDM libraries stubbed."""
from pyplayready.remote import serve as pyplayready_serve
from pywidevine import serve as pywidevine_serve
(tmp_path / "chrome_l3.wvd").touch()
(tmp_path / "sl2000.prd").touch()
monkeypatch.setattr(config.directories, "wvds", tmp_path)
monkeypatch.setattr(config.directories, "prds", tmp_path)
monkeypatch.setattr(
config,
"serve",
{
"api_secret": "master-key",
"dashboard": {"key": "dash-secret"},
"users": {
"tier-key": {"username": "tier1", "devices": ["chrome_l3"], "playready_devices": ["sl2000"]},
},
},
)
for module, cdm in ((pywidevine_serve, Cdm), (pyplayready_serve, PlayReadyCdm)):
monkeypatch.setattr(module, "Cdm", cdm)
monkeypatch.setattr(module.Device, "load", staticmethod(lambda _path: object()))
monkeypatch.setattr(pywidevine_serve, "PSSH", lambda data: data)
monkeypatch.setattr(pyplayready_serve, "PSSH", lambda _data: type("P", (), {"wrm_headers": ["header"]})())
captured: dict[str, web.Application] = {}
monkeypatch.setattr(web, "run_app", lambda app, **kwargs: captured.update(app=app))
result = CliRunner().invoke(serve, [])
assert result.exit_code == 0, result.output
cdm_calls.records.clear()
cdm_calls.seq = 0
return captured["app"]
async def cdm_log(client: Any, since: int = 0) -> dict[str, Any]:
resp = await client.get(f"/api/dashboard/cdm-logs?since={since}", headers=DASH)
assert resp.status == 200
return await resp.json()
async def test_widevine_exchange_is_recorded_per_call(aiohttp_client, serve_app) -> None:
client = await aiohttp_client(serve_app)
opened = await (await client.get("/chrome_l3/open", headers=USER)).json()
session_id = opened["data"]["session_id"]
assert (await cdm_log(client))["sessions"] == [
{"drm": "widevine", "device": "chrome_l3", "key_id": key_id("tier-key"), "user": "tier1", "open": 1, "max": 17}
]
# Accept-Encoding makes compression_middleware gzip the long challenge reply.
challenge = await client.post(
"/chrome_l3/get_license_challenge/STREAMING",
headers={**USER, "Accept-Encoding": "gzip"},
json={"session_id": session_id.upper(), "init_data": INIT_DATA},
)
assert challenge.status == 200, await challenge.text()
parsed = await client.post(
"/chrome_l3/parse_license", headers=USER, json={"session_id": session_id, "license_message": LICENCE}
)
assert parsed.status == 200, await parsed.text()
assert (await client.get(f"/chrome_l3/close/{session_id}", headers=USER)).status == 200
page = await cdm_log(client)
records = page["records"]
assert [r["op"] for r in records] == ["open", "get_license_challenge", "parse_license", "close"]
assert [r["seq"] for r in records] == [1, 2, 3, 4] and page["seq"] == 4
for record in records:
assert record["drm"] == "widevine" and record["device"] == "chrome_l3"
assert record["session_id"] == session_id
assert record["user"] == "tier1" and record["key_id"] == key_id("tier-key")
assert record["status"] == 200 and record["ok"] is True and record["error"] is None
assert isinstance(record["ms"], float) and abs(record["ts"] - time.time()) < 60
assert set(record) == {
"seq", "ts", "drm", "op", "device", "key_id", "user", "session_id", "status", "ok", "error", "ms",
} # fmt: skip
assert page["sessions"][0]["open"] == 0
dump = json.dumps(page)
for secret in ("tier-key", INIT_DATA, CHALLENGE_B64_PREFIX, "challenge-marker", LICENCE, CONTENT_KEY):
assert secret not in dump
async def test_playready_sub_app_calls_are_tagged_playready(aiohttp_client, serve_app) -> None:
client = await aiohttp_client(serve_app)
opened = await (await client.get("/playready/sl2000/open", headers=USER)).json()
session_id = opened["data"]["session_id"]
challenge = await client.post(
"/playready/sl2000/get_license_challenge", headers=USER, json={"session_id": session_id, "init_data": INIT_DATA}
)
assert challenge.status == 200, await challenge.text()
page = await cdm_log(client)
assert [(r["drm"], r["op"], r["device"], r["session_id"]) for r in page["records"]] == [
("playready", "open", "sl2000", session_id),
("playready", "get_license_challenge", "sl2000", session_id),
]
assert page["sessions"] == [
{"drm": "playready", "device": "sl2000", "key_id": key_id("tier-key"), "user": "tier1", "open": 1, "max": 17}
]
# A session past SESSION_TIMEOUT is one the library drops on its next open.
serve_app["playready_app"]["cdms"][("tier-key", "sl2000")].age(PlayReadyCdm.SESSION_TIMEOUT + 1)
assert (await cdm_log(client))["sessions"][0]["open"] == 0
async def test_refusals_are_recorded_with_their_message(aiohttp_client, serve_app) -> None:
client = await aiohttp_client(serve_app)
assert (await client.get("/other_device/open", headers=USER)).status == 403
assert (await client.get("/chrome_l3/open", headers={"X-Secret-Key": "wrong-key"})).status == 401
assert (await client.get("/chrome_l3/open")).status == 401
assert (await client.get("/playready/sl2000/open", headers={"X-Secret-Key": "wrong-key"})).status == 401
unopened = await client.post("/chrome_l3/get_keys/CONTENT", headers=USER, json={"session_id": "AB" * 16})
assert unopened.status == 400
assert (await client.post("/chrome_l3/parse_license", headers=USER, json={})).status == 400
page = await cdm_log(client)
forbidden, invalid, empty, playready, no_cdm, no_field = page["records"]
assert forbidden["status"] == 403 and forbidden["ok"] is False and forbidden["user"] == "tier1"
assert forbidden["device"] == "other_device" and "not authorized" in forbidden["error"]
assert forbidden["session_id"] is None
assert invalid["status"] == 401 and invalid["ok"] is False and "Invalid" in invalid["error"]
assert invalid["user"] == "anonymous" and invalid["key_id"] == "anonymous"
assert empty["status"] == 401 and "Empty" in empty["error"] and empty["user"] == "anonymous"
assert playready["drm"] == "playready" and playready["status"] == 401 and "Invalid" in playready["error"]
assert no_cdm["op"] == "get_keys" and no_cdm["status"] == 400 and no_cdm["session_id"] == "ab" * 16
assert no_field["op"] == "parse_license" and no_field["session_id"] is None and "session_id" in no_field["error"]
assert page["sessions"] == [] and "wrong-key" not in json.dumps(page)
async def test_gzipped_refusal_is_read_and_capped(aiohttp_client, serve_app) -> None:
"""compression_middleware runs inside the log middleware, so a long refusal arrives as gzip."""
client = await aiohttp_client(serve_app)
device = "d" * 400
resp = await client.get(f"/{device}/open", headers={**USER, "Accept-Encoding": "gzip"})
assert resp.status == 403 and resp.headers["Content-Encoding"] == "gzip"
(record,) = (await cdm_log(client))["records"]
assert record["device"] == device[:255] and len(record["error"]) == 300
assert record["error"].startswith("Device 'ddd")
async def test_refused_call_leaves_the_request_body_unread(aiohttp_client, serve_app) -> None:
"""A 401 is decided before the handler, so the body stays unread: no cost, and no session id, for an unknown key."""
client = await aiohttp_client(serve_app)
body = {"session_id": "ab" * 16, "license_message": LICENCE}
assert (
await client.post("/chrome_l3/parse_license", headers={"X-Secret-Key": "wrong-key"}, json=body)
).status == 401
assert (await client.post("/chrome_l3/parse_license", headers=USER, json=body)).status == 400
refused, unopened = (await cdm_log(client))["records"]
assert refused["status"] == 401 and refused["session_id"] is None
assert unopened["status"] == 400 and unopened["session_id"] == "ab" * 16
async def test_raised_http_exception_is_recorded_with_its_status(aiohttp_client) -> None:
"""aiohttp answers a raised ``HTTPException`` itself, so the record carries that status, not 500."""
async def too_large(_request: web.Request) -> web.Response:
raise web.HTTPRequestEntityTooLarge(max_size=1, actual_size=2)
app = web.Application(middlewares=[cdm_log_middleware])
app.router.add_post("/{device}/parse_license", too_large)
before = cdm_calls.seq
client = await aiohttp_client(app)
resp = await client.post("/chrome_l3/parse_license", json={"session_id": "ab" * 16})
(record,) = cdm_calls.since(before)
assert resp.status == 413 and record["status"] == 413 and record["ok"] is False
assert record["error"] == resp.reason
async def test_since_cursor_and_dashboard_key(aiohttp_client, serve_app) -> None:
client = await aiohttp_client(serve_app)
for _ in range(3):
await client.get("/chrome_l3/open", headers=USER)
assert (await client.get("/api/dashboard/cdm-logs")).status == 401
assert (await client.get("/api/dashboard/cdm-logs", headers=USER)).status == 401
# Only the device routes are CDM calls: not the API, not the ping, not this route.
await client.get("/api/health")
await client.get("/", headers=USER)
page = await cdm_log(client, since=1)
assert [r["seq"] for r in page["records"]] == [2, 3] and page["seq"] == 3
assert await cdm_log(client, since=3) == {"seq": 3, "records": [], "sessions": page["sessions"]}
resp = await client.get("/api/dashboard/cdm-logs?since=junk", headers=DASH)
assert len((await resp.json())["records"]) == 3
assert page["sessions"][0]["open"] == 3
async def test_handler_exception_is_recorded_without_its_text(aiohttp_client) -> None:
"""An unhandled error still answers 500, and its text (it can hold the API key) stays out."""
async def broken(request: web.Request) -> web.Response:
raise KeyError(request.headers["X-Secret-Key"])
app = web.Application(middlewares=[cdm_log_middleware])
app.router.add_get("/{device}/open", broken)
before = cdm_calls.seq
client = await aiohttp_client(app)
assert (await client.get("/chrome_l3/open", headers={"X-Secret-Key": "leaky-key"})).status == 500
(record,) = cdm_calls.since(before)
assert record["status"] == 500 and record["ok"] is False and record["error"] == "KeyError"
async def test_recording_failure_never_fails_the_cdm_call(aiohttp_client, monkeypatch) -> None:
async def opened(_request: web.Request) -> web.Response:
return web.json_response({"message": "Success"})
def refuse(_record: dict[str, Any]) -> None:
raise RuntimeError("log is broken")
monkeypatch.setattr(cdm_calls, "append", refuse)
app = web.Application(middlewares=[cdm_log_middleware])
app.router.add_get("/{device}/open", opened)
client = await aiohttp_client(app)
assert (await client.get("/chrome_l3/open")).status == 200
def test_call_log_keeps_the_newest_and_counts_on() -> None:
log = SeqRing(maxlen=2)
for op in ("open", "get_keys", "close"):
log.append({"op": op})
assert [(r["seq"], r["op"]) for r in log.since()] == [(2, "get_keys"), (3, "close")]
assert log.since(2) == [{"seq": 3, "op": "close"}] and log.seq == 3
def test_sessions_snapshot_is_empty_without_a_cdm() -> None:
assert cdm_sessions(web.Application()) == []
+78
View File
@@ -0,0 +1,78 @@
"""Pins the SIGHUP config reload: a valid file swaps the users in, a bad or restart-only file leaves the old one."""
from __future__ import annotations
import logging
from pathlib import Path
import pytest
import yaml
from aiohttp import web
from unshackle.commands import serve as serve_cmd
from unshackle.core import config as config_module
from unshackle.core.config import config
pytestmark = pytest.mark.unit
LOG = logging.getLogger("serve")
@pytest.fixture
def running(monkeypatch: pytest.MonkeyPatch, tmp_path: Path) -> tuple[web.Application, Path]:
"""An app serving the key ``old`` with one open CDM session, and a writable config path."""
wvds = tmp_path / "wvds"
prds = tmp_path / "prds"
wvds.mkdir()
prds.mkdir()
(wvds / "dev.wvd").write_bytes(b"x")
monkeypatch.setattr(config.directories, "wvds", wvds)
monkeypatch.setattr(config.directories, "prds", prds)
monkeypatch.setattr(config, "serve", {"api_secret": "api", "users": {"old": {"devices": ["dev"]}}})
path = tmp_path / "unshackle.yaml"
monkeypatch.setattr(config_module, "config_path", path)
app = web.Application()
app["config"] = {"users": {"old": {"devices": ["dev"]}, "api": {"devices": ["dev"]}}, "devices": {}}
app["cdms"] = {("old", "dev"): object()}
return app, path
def _write(path: Path, **serve: object) -> None:
path.write_text(yaml.safe_dump({"serve": {"api_secret": "api", **serve}}), encoding="utf8")
def _reload(app: web.Application) -> bool:
return serve_cmd.reload_config(app, no_key=False, serve_widevine=True, serve_playready=True)
def test_valid_reload_swaps_the_accepted_keys(running: tuple[web.Application, Path]) -> None:
app, path = running
_write(path, users={"new": {"devices": ["dev"]}})
assert _reload(app) is True
assert set(app["config"]["users"]) == {"new", "api"}
assert app["config"]["devices"] == {"dev": path.parent / "wvds" / "dev.wvd"}
assert config.serve["users"] == {"new": {"devices": ["dev"]}}
assert ("old", "dev") in app["cdms"]
def test_invalid_file_keeps_the_running_config(running: tuple[web.Application, Path]) -> None:
app, path = running
_write(path, users={"new": {"devices": ["dev"], "tier": "gold"}})
assert _reload(app) is False
assert set(app["config"]["users"]) == {"old", "api"}
assert config.serve["users"] == {"old": {"devices": ["dev"]}}
def test_restart_only_change_is_reported_not_applied(
running: tuple[web.Application, Path], caplog: pytest.LogCaptureFixture
) -> None:
app, path = running
_write(path, users={"new": {"devices": ["dev"]}}, remote_only=True)
with caplog.at_level(logging.WARNING, logger=LOG.name):
assert _reload(app) is False
assert "restart" in caplog.text and "remote_only" in caplog.text
assert set(app["config"]["users"]) == {"old", "api"}
+80 -12
View File
@@ -93,6 +93,7 @@ from unshackle.core.utilities import (
is_close_match,
is_exact_match,
keep_forced_subtitle,
kind_only_conflict,
log_event,
missing_required_langs,
partition_exclusions,
@@ -756,6 +757,13 @@ class dl:
help="Subtitle languages that must exist. The title fails if one is missing. -sl still decides what to keep.",
)
@click.option("-fs", "--forced-subs", is_flag=True, default=False, help="Include forced subtitle tracks.")
@click.option(
"-fso",
"--forced-subs-only",
is_flag=True,
default=False,
help="Download forced subtitle tracks and no other subtitle tracks.",
)
@click.option(
"-fsl",
"--forced-s-lang",
@@ -882,6 +890,13 @@ class dl:
help="Do not download or mux attachments (cover art, subtitle fonts, and files the service attaches).",
)
@click.option("-ad", "--audio-description", is_flag=True, default=False, help="Download audio description tracks.")
@click.option(
"-ado",
"--audio-description-only",
is_flag=True,
default=False,
help="Download audio description tracks and no standard audio tracks.",
)
@click.option(
"--slow",
type=SLOW_DELAY_RANGE,
@@ -1523,6 +1538,7 @@ class dl:
require_video: list[str],
require_subs: list[str],
forced_subs: bool,
forced_subs_only: bool,
forced_s_lang: list[str],
exact_lang: bool,
sub_format: Optional[Union[Subtitle.Codec, str]],
@@ -1537,6 +1553,7 @@ class dl:
no_video: bool,
no_attachments: bool,
audio_description: bool,
audio_description_only: bool,
slow: Optional[tuple[int, int]],
list_: bool,
list_titles: bool,
@@ -1639,7 +1656,27 @@ class dl:
if excludes and name in parent_params:
parent_params[name] = includes
if forced_s_lang or fsl_excl:
conflict = kind_only_conflict(
{
"audio_description_only": audio_description_only,
"forced_subs_only": forced_subs_only,
"no_audio": no_audio,
"no_subs": no_subs,
"video_only": video_only,
"audio_only": audio_only,
"subs_only": subs_only,
"chapters_only": chapters_only,
}
)
if conflict:
first, second = (f"--{name.replace('_', '-')}" for name in conflict)
self.log.error(f"{first} cannot be used with {second}: {second} drops the tracks that {first} selects.")
sys.exit(1)
if audio_description_only:
audio_description = True
if forced_s_lang or fsl_excl or forced_subs_only:
# services read forced_subs from raw ctx params, so the implication must be pushed there too
forced_subs = True
if "all" in forced_s_lang:
@@ -2044,6 +2081,27 @@ class dl:
post_script_last: dict[Any, dict[Path, dict[str, str]]] = {}
post_script_folders: list[Path] = []
post_script_sample: dict[str, str] = {}
wanted_total = sum(post_script_pending.values())
def settle_post_script_group(title: Any) -> None:
"""Count one title of its group as done; send the season event after the last one."""
group_key = post_script_group(title)
post_script_pending[group_key] = post_script_pending.get(group_key, 1) - 1
if post_script_pending[group_key] <= 0:
for folder, context in post_script_last.get(group_key, {}).items():
dispatch("success", "season", season_context(context, folder), postscript)
def skip_title(title: Any, reason: str) -> None:
"""Skip a title that has no track of the kind an only flag selects.
A run with one title stops instead, because it has nothing else to download.
"""
if wanted_total <= 1:
self.log.error(reason)
sys.exit(1)
self.log.warning(f"{reason} Skipping {title}")
if not no_mux:
settle_post_script_group(title)
for i, title in enumerate(titles):
v_lang, a_lang, s_lang, range_ = base_selection
@@ -2316,13 +2374,14 @@ class dl:
(
"audio",
require_audio if keep_audio else [],
[a.language for a in title.tracks.audio] + embedded_audio_langs(title.tracks.videos, keep_videos),
[a.language for a in title.tracks.audio if a.descriptive or not audio_description_only]
+ embedded_audio_langs(title.tracks.videos, keep_videos and not audio_description_only),
),
("video", require_video if keep_videos else [], [v.language for v in title.tracks.videos]),
(
"subtitle",
require_subs if keep_subtitles else [],
[t.language for t in title.tracks.subtitles],
[t.language for t in title.tracks.subtitles if t.forced or not forced_subs_only],
),
):
missing_required = missing_required_langs(required, available, title.language, exact=exact_lang)
@@ -2749,7 +2808,17 @@ class dl:
fsl_excl_r, [t.language for t in title.tracks.subtitles if t.forced], exact_lang
)
title.tracks.select_subtitles(lambda x: not (x.forced and str(x.language) in drop))
if forced_subs_only:
title.tracks.select_subtitles(lambda x: x.forced)
if not title.tracks.subtitles:
skip_title(title, "There's no forced subtitle track for the selected languages.")
continue
if keep_audio and audio_description_only:
title.tracks.select_audio(lambda x: x.descriptive)
if not title.tracks.audio:
skip_title(title, "There's no audio description track.")
continue
# might have no audio tracks if part of the video, e.g. transport stream hls
if keep_audio and len(title.tracks.audio) > 0:
if not audio_description:
@@ -2818,7 +2887,10 @@ class dl:
if a_orig_token in audio_languages:
a_orig_token = None
embedded_langs = embedded_audio_langs(title.tracks.videos, keep_videos)
embedded_langs = embedded_audio_langs(
title.tracks.videos, keep_videos and not audio_description_only
)
audio_noun = "audio description tracks" if audio_description_only else "audio tracks"
if not any(tok in processed_lang for tok in ("best", "all")):
missing_a_langs = find_missing_langs(
@@ -2832,17 +2904,17 @@ class dl:
remaining = [tok for tok in processed_lang if tok not in missing_a_langs]
if remaining:
self.log.warning(
f"{missing_str} not found in audio tracks, "
f"{missing_str} not found in {audio_noun}, "
f"continuing with: {as_requested(remaining, a_orig_token)}"
)
processed_lang = remaining
else:
self.log.error(
f"{missing_str} not found in audio tracks and no fallback available"
f"{missing_str} not found in {audio_noun} and no fallback available"
)
sys.exit(1)
else:
self.log.error(missing_str + " not found in audio tracks")
self.log.error(f"{missing_str} not found in {audio_noun}")
sys.exit(1)
title.tracks.audio = select_best_audio(
@@ -3700,11 +3772,7 @@ class dl:
post_script_folders.append(final_path.parent)
if not no_mux:
group_key = post_script_group(title)
post_script_pending[group_key] = post_script_pending.get(group_key, 1) - 1
if post_script_pending[group_key] <= 0:
for folder, context in post_script_last.get(group_key, {}).items():
dispatch("success", "season", season_context(context, folder), postscript)
settle_post_script_group(title)
title_dl_time = time_elapsed_since(dl_start_time)
downloaded_label = "Track" if isinstance(title, Song) else "Title"
+221 -103
View File
@@ -2,14 +2,19 @@ import asyncio
import hmac
import logging
import re
import signal
import subprocess
import sys
from contextlib import suppress
from datetime import datetime
from functools import partial
from pathlib import Path
from typing import Any, Optional
import click
from aiohttp import web
import unshackle.core.config as config_module
from unshackle.core import binaries
from unshackle.core.api import cors_middleware, setup_routes, setup_swagger
from unshackle.core.api.compression import compression_middleware
@@ -25,8 +30,16 @@ from unshackle.core.api.handlers import (
server_cdm_max_height_error,
validate_server_accounts,
)
from unshackle.core.api.stats import key_rate_limit, rate_limit_error, ring, stats, stats_middleware
from unshackle.core.config import config
from unshackle.core.api.stats import (
cdm_log_middleware,
key_id,
key_rate_limit,
rate_limit_error,
ring,
stats,
stats_middleware,
)
from unshackle.core.config import Config, config
from unshackle.core.console import console
from unshackle.core.constants import context_settings
from unshackle.core.downloaders import format_speed, parse_speed_limit, set_speed_limit
@@ -92,6 +105,189 @@ def _install_service_refresh(app: web.Application) -> None:
app.on_cleanup.append(stop)
RESTART_KEYS = ("remote_only", "services_refresh_interval")
"""``serve.*`` keys a reload cannot apply: they decide routes or tasks fixed at startup."""
log = logging.getLogger("serve")
def _prepare_serve(serve: dict, no_key: bool) -> Optional[str]:
"""Glob the device directories into ``serve``, validate it in place and return the api_secret.
Raises ValueError on a bad value. The api_secret is None under ``--no-key``.
"""
api_secret = None if no_key else serve.get("api_secret")
if not no_key and not api_secret:
raise ValueError(
"API secret key is not configured. Please add 'api_secret' to the 'serve' section in your config."
)
for key, directory, pattern in (
("devices", config.directories.wvds, "*.wvd"),
("playready_devices", config.directories.prds, "*.prd"),
):
if not serve.get(key):
serve[key] = []
serve[key].extend(directory.glob(pattern))
for device in serve[key]:
if not Path(device).is_file():
raise ValueError(f"serve.{key}: device file does not exist: {device}")
try:
parse_speed_limit(serve.get("global_speed_limit"))
except ValueError as e:
raise ValueError(f"serve.global_speed_limit: {e}")
users = serve.get("users", {})
if isinstance(users, dict):
users = {str(k): v for k, v in users.items()}
serve["users"] = users
tiers = serve.get("tiers") or {}
if not isinstance(tiers, dict):
raise ValueError("serve.tiers must be a mapping of tier name to settings")
for tier_name, tier_cfg in tiers.items():
if not isinstance(tier_cfg, dict):
raise ValueError(f"serve.tiers.{tier_name} must be a mapping of setting to value")
if problem := rate_limit_error(f"serve.tiers.{tier_name}", tier_cfg.get("rate_limit")):
raise ValueError(problem)
for user_key, user_cfg in users.items() if isinstance(users, dict) else []:
user_services = user_cfg.get("services") if isinstance(user_cfg, dict) else None
username = user_cfg.get("username", user_key[:8] + "...") if isinstance(user_cfg, dict) else user_key[:8]
if user_services:
log.info(f"User '{username}' restricted to services: {', '.join(user_services)}")
tier = user_cfg.get("tier") if isinstance(user_cfg, dict) else None
if tier and tier not in tiers:
raise ValueError(f"serve.users.{username}.tier: no such tier '{tier}' under serve.tiers")
if isinstance(user_cfg, dict):
if problem := rate_limit_error(f"serve.users.{username}", user_cfg.get("rate_limit")):
raise ValueError(problem)
if problem := server_cdm_max_height_error(f"serve.users.{username}", user_cfg.get("server_cdm_max_height")):
raise ValueError(problem)
limit = key_rate_limit(user_key, serve)
if limit:
log.info(f"User '{username}' rate limited to {limit} requests/hour")
return api_secret
def _build_cdm_config(
serve: dict, api_secret: Optional[str], serve_widevine: bool, serve_playready: bool
) -> tuple[dict, dict]:
"""The ``app["config"]`` for the main app and the PlayReady sub-app, derived from a prepared ``serve``.
The main config carries the api_secret user and a ``playready_devices`` list per user; the
sub-app config maps each user's ``playready_devices`` to ``devices`` for pyplayready.
"""
serve_config = dict(serve)
wvd_devices = serve_config.get("devices", []) if serve_widevine else []
prd_devices = serve_config.get("playready_devices", []) if serve_playready else []
wvd_device_names = [Path(d).stem for d in wvd_devices]
prd_device_names = [Path(d).stem for d in prd_devices]
users_cfg = serve_config.get("users")
serve_config["users"] = {k: dict(v) for k, v in users_cfg.items()} if isinstance(users_cfg, dict) else {}
if api_secret and api_secret not in serve_config["users"]:
serve_config["users"][api_secret] = {
"devices": wvd_device_names,
"playready_devices": prd_device_names,
"username": "api_user",
}
for user_key, user_config in serve_config["users"].items():
if "playready_devices" not in user_config:
user_config["playready_devices"] = []
if serve_playready:
log.warning(
f'User "{key_id(user_key)}" has no "playready_devices" configured; PlayReady access disabled '
f"for this user. Available PlayReady devices: {prd_device_names}"
)
serve_config["devices"] = wvd_devices
serve_config["playready_devices"] = prd_devices
playready_config = {
"devices": prd_devices,
"users": {
user_key: {
"devices": user_cfg.get("playready_devices", []),
"username": user_cfg.get("username", "user"),
}
for user_key, user_cfg in serve_config["users"].items()
}
if api_secret
else {},
}
return serve_config, playready_config
def _device_map(devices: list[Any]) -> dict[str, Path]:
"""The ``{stem: path}`` form pywidevine and pyplayready rewrite ``app["config"]["devices"]`` into at startup."""
return {Path(d).stem: Path(d) for d in devices}
def _device_stems(serve: dict) -> set[str]:
return {Path(d).stem for d in serve.get("devices", []) + serve.get("playready_devices", [])}
def reload_config(app: web.Application, *, no_key: bool, serve_widevine: bool, serve_playready: bool) -> bool:
"""Re-read the config file and swap it in; return True when the new file is in use.
The swap replaces every ``Config`` section, so services read new credentials and CDM mappings
too. Runs synchronously on the event loop so no request sees a half-updated config. A file
that fails to parse or validate, or that changes a startup-only setting, keeps the running
config. The reload leaves open CDM sessions alone: a revoked key fails authentication on
its next call.
"""
try:
if not config_module.config_path:
raise ValueError("no config file found")
fresh = Config.from_yaml(config_module.config_path)
serve = fresh.serve
api_secret = _prepare_serve(serve, no_key)
except Exception as e:
log.error(f"Config reload rejected, keeping the running config: {e}")
return False
changed = [k for k in RESTART_KEYS if serve.get(k) != config.serve.get(k)]
if serve_playready and bool(serve.get("playready_devices")) != ("playready_app" in app):
changed.append("playready_devices (first device added or last removed)")
if changed:
log.warning(f"Config reload skipped, restart to apply: {', '.join(changed)}")
return False
serve_config, playready_config = _build_cdm_config(serve, api_secret, serve_widevine, serve_playready)
old_users, new_users = set(app["config"]["users"]), set(serve_config["users"])
old_devices, new_devices = _device_stems(config.serve), _device_stems(serve)
vars(config).update(vars(fresh))
app["config"]["users"] = serve_config["users"]
if serve_widevine:
app["config"]["devices"] = _device_map(serve_config["devices"])
if playready_app := app.get("playready_app"):
playready_app["config"]["users"] = playready_config["users"]
playready_app["config"]["devices"] = _device_map(playready_config["devices"])
set_speed_limit(parse_speed_limit(serve.get("global_speed_limit")), lock=True)
log.info(
"Config reloaded: "
f"keys added {sorted(key_id(k) for k in new_users - old_users)}, "
f"removed {sorted(key_id(k) for k in old_users - new_users)}; "
f"devices added {sorted(new_devices - old_devices)}, removed {sorted(old_devices - new_devices)}"
)
return True
def _install_config_reload(app: web.Application, *, no_key: bool, serve_widevine: bool, serve_playready: bool) -> None:
"""Reload the config on SIGHUP, so ``systemctl reload`` applies a config edit without a restart."""
async def start(_app: web.Application) -> None:
reload = partial(
reload_config, _app, no_key=no_key, serve_widevine=serve_widevine, serve_playready=serve_playready
)
try:
asyncio.get_running_loop().add_signal_handler(signal.SIGHUP, reload)
except (AttributeError, NotImplementedError):
return
log.info("Config hot-reload: send SIGHUP (systemctl reload) after editing unshackle.yaml")
app.on_startup.append(start)
@click.command(
short_help="Serve your Local Widevine/PlayReady Devices and REST API for Remote Access.",
context_settings=context_settings,
@@ -173,8 +369,6 @@ def serve(
from pyplayready.remote import serve as pyplayready_serve
from pywidevine import serve as pywidevine_serve
log = logging.getLogger("serve")
if quiet:
logging.basicConfig(
force=True,
@@ -206,14 +400,7 @@ def serve(
logging.getLogger().addHandler(ring)
logging.getLogger("aiohttp.access").addFilter(_MaskAccessKey())
if not no_key:
api_secret = config.serve.get("api_secret")
if not api_secret:
raise click.ClickException(
"API secret key is not configured. Please add 'api_secret' to the 'serve' section in your config."
)
else:
api_secret = None
if no_key:
log.warning("Running with --no-key: Authentication is DISABLED for all API endpoints!")
if debug_api:
@@ -232,13 +419,10 @@ def serve(
caddy_p = None
try:
if not config.serve.get("devices"):
config.serve["devices"] = []
config.serve["devices"].extend(list(config.directories.wvds.glob("*.wvd")))
if not config.serve.get("playready_devices"):
config.serve["playready_devices"] = []
config.serve["playready_devices"].extend(list(config.directories.prds.glob("*.prd")))
try:
api_secret = _prepare_serve(config.serve, no_key)
except ValueError as e:
raise click.ClickException(str(e))
remote_only = remote_only or config.serve.get("remote_only", False)
if remote_only:
@@ -251,10 +435,7 @@ def serve(
elif quiet:
log.info("Developer dashboard disabled: set serve.dashboard.key in unshackle.yaml to enable it")
try:
global_speed_limit = parse_speed_limit(config.serve.get("global_speed_limit"))
except ValueError as e:
raise click.ClickException(f"serve.global_speed_limit: {e}")
global_speed_limit = parse_speed_limit(config.serve.get("global_speed_limit"))
if global_speed_limit:
set_speed_limit(global_speed_limit, lock=True)
log.info(f"Global speed limit: {format_speed(global_speed_limit)} (shared by all jobs)")
@@ -272,37 +453,11 @@ def serve(
log.info(f"Server accounts for {tag}: {', '.join(covered)}")
if config.key_vaults and not any(v.get("type") == "SQLite" for v in config.key_vaults):
log.warning("No SQLite key vault configured: a content key a remote client proves wrong cannot be flagged")
users = config.serve.get("users", {})
if isinstance(users, dict):
# yaml keys can parse as int; hmac.compare_digest and allowlist lookups need str
users = {str(k): v for k, v in users.items()}
config.serve["users"] = users
tiers = config.serve.get("tiers") or {}
if not isinstance(tiers, dict):
raise click.ClickException("serve.tiers must be a mapping of tier name to settings")
for tier_name, tier_cfg in tiers.items():
if not isinstance(tier_cfg, dict):
raise click.ClickException(f"serve.tiers.{tier_name} must be a mapping of setting to value")
if problem := rate_limit_error(f"serve.tiers.{tier_name}", tier_cfg.get("rate_limit")):
raise click.ClickException(problem)
for user_key, user_cfg in users.items() if isinstance(users, dict) else []:
user_services = user_cfg.get("services") if isinstance(user_cfg, dict) else None
username = user_cfg.get("username", user_key[:8] + "...") if isinstance(user_cfg, dict) else user_key[:8]
if user_services:
log.info(f"User '{username}' restricted to services: {', '.join(user_services)}")
tier = user_cfg.get("tier") if isinstance(user_cfg, dict) else None
if tier and tier not in tiers:
raise click.ClickException(f"serve.users.{username}.tier: no such tier '{tier}' under serve.tiers")
if isinstance(user_cfg, dict):
if problem := rate_limit_error(f"serve.users.{username}", user_cfg.get("rate_limit")):
raise click.ClickException(problem)
if problem := server_cdm_max_height_error(
f"serve.users.{username}", user_cfg.get("server_cdm_max_height")
):
raise click.ClickException(problem)
limit = key_rate_limit(user_key)
if limit:
log.info(f"User '{username}' rate limited to {limit} requests/hour")
serve_widevine = not api_only and not no_widevine
serve_playready = not api_only and not no_playready
serve_config, playready_config = _build_cdm_config(config.serve, api_secret, serve_widevine, serve_playready)
prd_devices = playready_config["devices"]
if api_only:
log.info("Starting REST API server (pywidevine/pyplayready CDM disabled)")
@@ -310,7 +465,6 @@ def serve(
app = web.Application(
middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
)
app["config"] = {"users": {}}
else:
app = web.Application(
middlewares=[
@@ -321,10 +475,7 @@ def serve(
compression_middleware,
]
)
api_users: dict = {api_secret: {"devices": [], "username": "api_user"}}
if isinstance(users, dict):
api_users.update(users)
app["config"] = {"users": api_users}
app["config"] = serve_config
app["debug_api"] = debug_api
from unshackle.core.api.session_store import get_session_store
@@ -341,6 +492,7 @@ def serve(
app.on_startup.append(start_session_cleanup)
app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app)
_install_config_reload(app, no_key=no_key, serve_widevine=serve_widevine, serve_playready=serve_playready)
setup_routes(app, remote_only=remote_only, dashboard=dashboard)
if not remote_only:
@@ -352,13 +504,6 @@ def serve(
log.info("(Press CTRL+C to quit)")
web.run_app(app, host=host, port=port, print=None)
else:
serve_widevine = not no_widevine
serve_playready = not no_playready
serve_config = dict(config.serve)
wvd_devices = serve_config.get("devices", []) if serve_widevine else []
prd_devices = serve_config.get("playready_devices", []) if serve_playready else []
cdm_parts = []
if serve_widevine:
cdm_parts.append("pywidevine CDM")
@@ -366,28 +511,6 @@ def serve(
cdm_parts.append("pyplayready CDM")
log.info(f"Starting integrated server ({' + '.join(cdm_parts)} + REST API)")
wvd_device_names = [d.stem if hasattr(d, "stem") else str(d) for d in wvd_devices]
prd_device_names = [d.stem if hasattr(d, "stem") else str(d) for d in prd_devices]
users_cfg = serve_config.get("users")
serve_config["users"] = dict(users_cfg) if isinstance(users_cfg, dict) else {}
if not no_key and api_secret not in serve_config["users"]:
serve_config["users"][api_secret] = {
"devices": wvd_device_names,
"playready_devices": prd_device_names,
"username": "api_user",
}
for user_key, user_config in serve_config["users"].items():
if "playready_devices" not in user_config:
# Require explicit PlayReady device access per user (default: no access).
user_config["playready_devices"] = []
log.warning(
f'User "{user_key}" has no "playready_devices" configured; PlayReady access disabled for this user. '
f"Available PlayReady devices: {prd_device_names}"
)
def create_serve_authentication(serve_playready_flag: bool):
@web.middleware
async def serve_authentication(request: web.Request, handler) -> web.StreamResponse:
@@ -424,7 +547,13 @@ def serve(
if no_key:
app = web.Application(
middlewares=[cors_middleware, stats_middleware, dashboard_authentication, compression_middleware]
middlewares=[
cors_middleware,
stats_middleware,
cdm_log_middleware,
dashboard_authentication,
compression_middleware,
]
)
else:
serve_auth = create_serve_authentication(serve_playready and bool(prd_devices))
@@ -432,6 +561,7 @@ def serve(
middlewares=[
cors_middleware,
stats_middleware,
cdm_log_middleware,
dashboard_authentication,
serve_auth,
compression_middleware,
@@ -455,6 +585,7 @@ def serve(
app.on_startup.append(start_session_cleanup)
app.on_cleanup.append(stop_session_cleanup)
_install_service_refresh(app)
_install_config_reload(app, no_key=no_key, serve_widevine=serve_widevine, serve_playready=serve_playready)
if serve_widevine:
app.on_startup.append(pywidevine_serve._startup)
@@ -466,20 +597,6 @@ def serve(
playready_app = web.Application()
else:
playready_app = web.Application(middlewares=[pyplayready_serve.authentication])
# PlayReady subapp config maps playready_devices to "devices" for pyplayready compatibility
playready_config = {
"devices": prd_devices,
"users": {
user_key: {
"devices": user_cfg.get("playready_devices", []),
"username": user_cfg.get("username", "user"),
}
for user_key, user_cfg in serve_config["users"].items()
}
if not no_key
else {},
}
playready_app["config"] = playready_config
playready_app.on_startup.append(pyplayready_serve._startup)
playready_app.on_cleanup.append(pyplayready_serve._cleanup)
@@ -495,6 +612,7 @@ def serve(
app.router.add_route("*", "/playready", playready_ping)
app.add_subapp("/playready", playready_app)
app["playready_app"] = playready_app
log.info(f"PlayReady CDM endpoints available at http://{host}:{port}/playready/")
elif serve_playready:
log.info("No PlayReady devices found, skipping PlayReady CDM endpoints")
+2
View File
@@ -601,6 +601,7 @@ def perform_download(
require_video=params.get("require_video", []),
require_subs=params.get("require_subs", []),
forced_subs=ctx.params["forced_subs"],
forced_subs_only=params.get("forced_subs_only", False),
forced_s_lang=params.get("forced_s_lang", []),
exact_lang=params.get("exact_lang", False),
sub_format=params.get("sub_format"),
@@ -615,6 +616,7 @@ def perform_download(
no_video=params.get("no_video", False),
no_attachments=params.get("no_attachments", False),
audio_description=params.get("audio_description", False),
audio_description_only=params.get("audio_description_only", False),
slow=params.get("slow", None),
list_=False,
list_titles=False,
+32 -9
View File
@@ -33,7 +33,7 @@ from unshackle.core.proxies.resolve import initialize_proxy_providers, resolve_p
from unshackle.core.services import Services
from unshackle.core.titles import Episode, Movie, Song, Title_T
from unshackle.core.tracks import Audio, Subtitle, Tracks, Video
from unshackle.core.utilities import declared_kwargs
from unshackle.core.utilities import declared_kwargs, kind_only_conflict
from unshackle.core.utils.click_types import AUDIO_CODEC_LIST, SUBTITLE_CODEC, VIDEO_CODEC_LIST
from unshackle.core.utils.collections import ci_get
from unshackle.core.utils.redact import REDACTED, URL_USERINFO_RE, redact_all, redact_secrets, redact_text
@@ -64,6 +64,7 @@ DEFAULT_DOWNLOAD_PARAMS = {
"require_video": [],
"require_subs": [],
"forced_subs": False,
"forced_subs_only": False,
"forced_s_lang": [],
"exact_lang": False,
"sub_format": None,
@@ -77,6 +78,7 @@ DEFAULT_DOWNLOAD_PARAMS = {
"no_video": False,
"no_attachments": False,
"audio_description": False,
"audio_description_only": False,
"slow": None,
"split_audio": None,
"skip_dl": False,
@@ -1925,6 +1927,13 @@ def validate_download_parameters(data: Dict[str, Any]) -> Optional[str]:
return "Cannot use both no_subs and subs_only"
if data.get("no_audio") and data.get("audio_only"):
return "Cannot use both no_audio and audio_only"
conflict = kind_only_conflict(data)
if conflict:
return f"Cannot use both {conflict[0]} and {conflict[1]}"
for key, default in DEFAULT_DOWNLOAD_PARAMS.items():
if isinstance(default, bool) and data.get(key) is not None and not isinstance(data[key], bool):
return f"{key} must be a boolean"
for key in ("require_audio", "require_video", "require_subs"):
value = data.get(key)
@@ -2333,19 +2342,36 @@ def _poll_cursor(before: int, items: List[Dict[str, Any]]) -> int:
return max(before, items[-1]["seq"]) if items else before
def _since(request: web.Request) -> int:
"""The ``since`` cursor of a poll; 0 (everything) when it is missing or not a number."""
try:
return int(request.query.get("since", 0))
except ValueError:
return 0
async def dashboard_logs_handler(request: web.Request) -> web.Response:
"""Recent log records; `since` (seq) and `level` filter the ring buffer."""
from unshackle.core.api.stats import ring
try:
since = int(request.query.get("since", 0))
except ValueError:
since = 0
since = _since(request)
before = ring.seq
records = ring.since(since, request.query.get("level"), request.query.get("logger"))
return web.json_response({"seq": _poll_cursor(before, records), "records": records})
async def dashboard_cdm_logs_handler(request: web.Request) -> web.Response:
"""CDM calls after `since` (seq), plus a snapshot of the open sessions of each live Cdm."""
from unshackle.core.api.stats import cdm_calls, cdm_sessions
since = _since(request)
before = cdm_calls.seq
records = cdm_calls.since(since)
return web.json_response(
{"seq": _poll_cursor(before, records), "records": records, "sessions": cdm_sessions(request.app)}
)
async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
"""One remote session's mirrored service log, for an operator watching a failing auth.
@@ -2359,10 +2385,7 @@ async def dashboard_session_logs_handler(request: web.Request) -> web.Response:
session = get_session_store().peek(session_id)
if session is None:
raise APIError(APIErrorCode.SESSION_NOT_FOUND, f"Remote session not found: {session_id}")
try:
since = int(request.query.get("since", 0))
except ValueError:
since = 0
since = _since(request)
buffer = session.log_buffer
before = buffer.last_seq if buffer else 0
records = buffer.since(since) if buffer else []
+93
View File
@@ -21,6 +21,7 @@ from unshackle.core.api.handlers import (
clear_cache_handler,
clear_finished_download_jobs_handler,
clear_temp_handler,
dashboard_cdm_logs_handler,
dashboard_events_handler,
dashboard_health_handler,
dashboard_jobs_handler,
@@ -895,6 +896,9 @@ async def download(request: web.Request) -> web.Response:
forced_subs:
type: boolean
description: Include forced subtitle tracks (default - false)
forced_subs_only:
type: boolean
description: Download forced subtitle tracks and no other subtitle tracks, implies forced_subs (default - false)
forced_s_lang:
type: array
items:
@@ -936,6 +940,9 @@ async def download(request: web.Request) -> web.Response:
audio_description:
type: boolean
description: Download audio description tracks (default - false)
audio_description_only:
type: boolean
description: Download audio description tracks and no standard audio tracks (default - false)
slow:
oneOf:
- type: boolean
@@ -2638,6 +2645,91 @@ async def dashboard_logs(request: web.Request) -> web.Response:
return await dashboard_logs_handler(request)
@api_handler
async def dashboard_cdm_logs(request: web.Request) -> web.Response:
"""
Dashboard: recent CDM calls and open CDM sessions.
---
summary: Dashboard CDM call log
description: >
The last 5000 calls to the Widevine and PlayReady CDM device routes, one record for each
call, with a snapshot of the open sessions of each live Cdm. A record never holds an API
key, `init_data`, a challenge, a certificate, a licence body or a content key.
Poll with `since` set to the `seq` of the last reply. The server does not publish these
records on the event stream.
tags: [Dashboard]
parameters:
- name: since
in: query
required: false
schema:
type: integer
description: Return only records with seq greater than this value
responses:
'200':
description: Records plus the current seq and the open sessions
content:
application/json:
schema:
type: object
properties:
seq:
type: integer
records:
type: array
items:
type: object
properties:
seq:
type: integer
ts:
type: number
drm:
type: string
enum: [widevine, playready]
op:
type: string
device:
type: string
key_id:
type: string
user:
type: string
session_id:
type: string
nullable: true
status:
type: integer
ok:
type: boolean
error:
type: string
nullable: true
ms:
type: number
sessions:
type: array
items:
type: object
properties:
drm:
type: string
device:
type: string
key_id:
type: string
user:
type: string
open:
type: integer
max:
type: integer
'401':
description: Dashboard key missing or invalid
"""
return await dashboard_cdm_logs_handler(request)
@api_handler
async def dashboard_session_logs(request: web.Request) -> web.Response:
"""
@@ -2980,6 +3072,7 @@ DASHBOARD_ROUTES: list[tuple[str, str, Handler, bool]] = [
("GET", DASHBOARD_PREFIX + "sessions", dashboard_sessions, True),
("GET", DASHBOARD_PREFIX + "jobs", dashboard_jobs, True),
("GET", DASHBOARD_PREFIX + "logs", dashboard_logs, True),
("GET", DASHBOARD_PREFIX + "cdm-logs", dashboard_cdm_logs, True),
("GET", DASHBOARD_PREFIX + "keys", dashboard_keys, True),
("GET", DASHBOARD_PREFIX + "services", dashboard_services, True),
("GET", DASHBOARD_PREFIX + "health", dashboard_health, True),
+226 -23
View File
@@ -2,12 +2,15 @@
from __future__ import annotations
import gzip
import hashlib
import json
import logging
import time
from collections import Counter, deque
from dataclasses import dataclass, field
from typing import Any, Deque, Dict, List, Optional
from functools import lru_cache
from typing import Any, Deque, Dict, List, Optional, Tuple
from aiohttp import web
@@ -17,6 +20,20 @@ from unshackle.core.config import config
from unshackle.core.utils.redact import redact_secrets
RATE_LIMIT_WINDOW = 3600.0
CDM_OPS = frozenset(
{
"open",
"close",
"set_service_certificate",
"get_service_certificate",
"get_license_challenge",
"parse_license",
"get_keys",
}
)
CDM_ERROR_LIMIT = 300
CDM_SESSION_ID_LIMIT = 64
CDM_DEVICE_LIMIT = 255
def _users() -> Dict[str, Any]:
@@ -70,15 +87,21 @@ def key_tier(key: Optional[str]) -> Optional[str]:
return str(tier) if tier else None
def key_rate_limit(key: Optional[str]) -> Optional[int]:
"""Requests per hour allowed for a key: its own override, else its tier's, else unlimited."""
user = _user_config(key)
if user is None:
def key_rate_limit(key: Optional[str], serve: Optional[Dict[str, Any]] = None) -> Optional[int]:
"""Requests per hour allowed for a key: its own override, else its tier's, else unlimited.
``serve`` reads a config block that is not yet the live one, as the reload validator does.
"""
if serve is None:
serve = config.serve or {}
users = serve.get("users")
user = users.get(key) if isinstance(users, dict) and key else None
if not isinstance(user, dict):
return None
own = user.get("rate_limit")
if isinstance(own, int) and own > 0:
return own
tiers = (config.serve or {}).get("tiers")
tiers = serve.get("tiers")
tier = tiers.get(user.get("tier")) if isinstance(tiers, dict) and user.get("tier") else None
limit = tier.get("rate_limit") if isinstance(tier, dict) else None
return limit if isinstance(limit, int) and limit > 0 else None
@@ -215,40 +238,220 @@ async def stats_middleware(request: web.Request, handler: Any) -> web.StreamResp
return response
class RingLogHandler(logging.Handler):
"""Keep the last N log records in memory and publish each one on the event bus."""
class SeqRing:
"""Keep the last N records in memory, each with a ``seq`` one above the record before it."""
def __init__(self, maxlen: int = 1000) -> None:
super().__init__()
def __init__(self, maxlen: int) -> None:
self.records: Deque[Dict[str, Any]] = deque(maxlen=maxlen)
self.seq = 0
def emit(self, record: logging.LogRecord) -> None:
"""Buffer and publish one record; ``seq`` moves only after the append, as in ``EventBus``."""
def append(self, record: Dict[str, Any]) -> Dict[str, Any]:
"""Buffer one record; ``seq`` moves only after the append, as in ``EventBus``."""
seq = self.seq + 1
item = {
"seq": seq,
"ts": record.created,
"level": record.levelname,
"logger": record.name,
"msg": redact_secrets(self.format(record)),
}
item = {"seq": seq, **record}
self.records.append(item)
self.seq = seq
return item
def since(self, seq: int = 0) -> List[Dict[str, Any]]:
"""The records after ``seq``, oldest first, from a copy: ``append()`` can run on a worker thread."""
return [r for r in self.records.copy() if r["seq"] > seq]
class RingLogHandler(SeqRing, logging.Handler):
"""Keep the last N log records in memory and publish each one on the event bus."""
def __init__(self, maxlen: int = 1000) -> None:
logging.Handler.__init__(self)
SeqRing.__init__(self, maxlen)
def emit(self, record: logging.LogRecord) -> None:
"""Buffer and publish one record."""
item = self.append(
{
"ts": record.created,
"level": record.levelname,
"logger": record.name,
"msg": redact_secrets(self.format(record)),
}
)
bus.publish("log", item)
def since(self, seq: int = 0, level: Optional[str] = None, logger: Optional[str] = None) -> List[Dict[str, Any]]:
"""Filter a copy of the records: ``emit()`` appends from worker threads."""
"""The records after ``seq`` at or above ``level``, from ``logger`` or one of its children."""
min_level = logging.getLevelName(level.upper()) if level else 0
if not isinstance(min_level, int):
min_level = 0
return [
r
for r in self.records.copy()
if r["seq"] > seq
and logging.getLevelName(r["level"]) >= min_level
for r in super().since(seq)
if logging.getLevelName(r["level"]) >= min_level
and (not logger or r["logger"] == logger or r["logger"].startswith(logger + "."))
]
ring = RingLogHandler()
cdm_calls = SeqRing(maxlen=5000)
@lru_cache(maxsize=None)
def _cdm_op(pattern: str) -> Optional[Tuple[str, str]]:
"""The ``(drm, op)`` of a route pattern, else None. Cached: the route table does not change."""
parts = pattern.strip("/").split("/")
drm = "widevine"
if parts[:1] == ["playready"]:
drm, parts = "playready", parts[1:]
if len(parts) < 2 or parts[0] != "{device}" or parts[1] not in CDM_OPS:
return None
return drm, parts[1]
def _cdm_route(request: web.Request) -> Optional[Tuple[str, str]]:
"""The ``(drm, op)`` of a CDM device route, else None.
Reads the matched route pattern, not the path, so a device name cannot pass for an operation
or for the ``/playready`` prefix.
"""
resource = request.match_info.route.resource
return _cdm_op(resource.canonical) if resource is not None else None
def _hex_id(value: Any) -> Optional[str]:
"""``value`` in lowercase when it is a hexadecimal string of a plausible length, else None."""
if not isinstance(value, str) or not value or len(value) > CDM_SESSION_ID_LIMIT:
return None
try:
bytes.fromhex(value)
except ValueError:
return None
return value.lower()
def _key_fields(secret_key: Optional[str]) -> Dict[str, str]:
"""The ``key_id`` and ``user`` of an API key: the ``id`` and ``label`` of its ``stats.keys`` entry."""
key = configured_key(secret_key)
return {"key_id": key_id(key), "user": mask_key(key)}
def _reply_json(response: Any) -> Dict[str, Any]:
"""The JSON object a handler answered, or ``{}``. ``compression_middleware`` runs inside the
CDM log middleware, so the body can arrive as gzip."""
try:
body = response.body
if response.headers.get("Content-Encoding") == "gzip":
body = gzip.decompress(body)
data = json.loads(body)
except Exception:
return {}
return data if isinstance(data, dict) else {}
async def _request_session_id(request: web.Request) -> Any:
"""The ``session_id`` of the JSON request body, when the handler consumed it: aiohttp caches that read.
A body nobody read belongs to a call refused before the handler, a 401 or a 429. It stays
unread: reading it would spend time and memory on an unauthenticated caller, and wait on a
body that never finishes arriving.
"""
if not request.content.at_eof():
return None
try:
body = await request.json()
except Exception:
return None
return body.get("session_id") if isinstance(body, dict) else None
async def _cdm_record(request: web.Request, drm: str, op: str, outcome: Any) -> Dict[str, Any]:
"""The record of one call. ``outcome`` is the response, or the exception the handler raised.
An exception that is not an HTTP one gives its class name only: its text can hold the
presented API key. The reply body is read only for a refusal and for the session id of an
``open``, so a challenge, a licence or a content key never passes through here.
"""
from unshackle.core.api.handlers import request_secret_key
reply: Dict[str, Any] = {}
if isinstance(outcome, web.StreamResponse):
status = outcome.status
if status >= 400 or op == "open":
reply = _reply_json(outcome)
message = reply.get("message")
error = message if isinstance(message, str) else outcome.reason
else:
status, error = 500, type(outcome).__name__
ok = status < 400
if op == "open":
data = reply.get("data") if ok else None
session_id = data.get("session_id") if isinstance(data, dict) else None
elif op == "close":
session_id = request.match_info.get("session_id")
else:
session_id = await _request_session_id(request)
return {
"drm": drm,
"op": op,
"device": request.match_info["device"][:CDM_DEVICE_LIMIT],
**_key_fields(request_secret_key(request)),
"session_id": _hex_id(session_id),
"status": status,
"ok": ok,
"error": None if ok else str(error)[:CDM_ERROR_LIMIT],
}
async def _record_cdm_call(
request: web.Request, route: Tuple[str, str], ts: float, started: float, outcome: Any
) -> None:
"""Append one call to ``cdm_calls``. Logs a failure and does not raise it: it must never fail the CDM call."""
ms = round((time.perf_counter() - started) * 1000, 1)
try:
cdm_calls.append({"ts": ts, **await _cdm_record(request, *route, outcome), "ms": ms})
except Exception:
logging.getLogger("serve").debug("CDM call not recorded", exc_info=True)
@web.middleware
async def cdm_log_middleware(request: web.Request, handler: Any) -> web.StreamResponse:
"""Record each call to a CDM device route in ``cdm_calls``.
Must run outside the serve authentication middleware, so the log holds a 401 too.
"""
route = _cdm_route(request)
if route is None:
return await handler(request)
ts, started = time.time(), time.perf_counter()
try:
response = await handler(request)
except Exception as e:
await _record_cdm_call(request, route, ts, started, e)
raise
await _record_cdm_call(request, route, ts, started, response)
return response
def cdm_sessions(app: web.Application) -> List[Dict[str, Any]]:
"""One row per live Cdm object: its open session count and the most it allows.
Both CDM libraries refuse an ``open`` only above ``MAX_NUM_OF_SESSIONS``, so the true limit
is one more. A Cdm with a ``SESSION_TIMEOUT`` drops older sessions on its next ``open``,
so those do not count as open.
"""
now = time.time()
rows: List[Dict[str, Any]] = []
for drm, owner in (("widevine", app), ("playready", app.get("playready_app"))):
cdms = owner.get("cdms") if owner is not None else None
for (secret_key, device), cdm in (cdms or {}).items():
sessions = getattr(cdm, "_Cdm__sessions", {}).values()
timeout = getattr(cdm, "SESSION_TIMEOUT", None)
rows.append(
{
"drm": drm,
"device": device,
**_key_fields(secret_key),
"open": len(sessions) if timeout is None else sum(now - s.opened_at <= timeout for s in sessions),
"max": cdm.MAX_NUM_OF_SESSIONS + 1,
}
)
return rows
+1 -1
View File
@@ -940,7 +940,7 @@ class CustomRemoteCDM:
raise ValueError("No challenge available - call get_license_challenge first")
if isinstance(license_message, str):
if self.is_playready and license_message.strip().startswith("<?xml"):
if self.is_playready and license_message.lstrip().startswith("<"):
license_message = license_message.encode("utf-8")
else:
try:
@@ -114,7 +114,7 @@ class DecryptLabsRemoteCDM:
Args:
secret: Decrypt Labs API key (matches config format)
host: Decrypt Labs API host URL (matches config format)
device_name: DRM system device name. Widevine takes ChromeCDM, L1 or L2, and PlayReady takes SL2 or SL3
device_name: DRM system device name. Widevine takes ChromeCDM, L1, L2 or L3, and PlayReady takes SL2 or SL3
service_name: Service tag for content key caching and vault operations
vaults: Vaults instance for local content key caching
device_type: Device type (CHROME, ANDROID, PLAYREADY) - for compatibility
@@ -585,7 +585,7 @@ class DecryptLabsRemoteCDM:
raise ValueError("No challenge available - call get_license_challenge first")
if isinstance(license_message, str):
if self.is_playready and license_message.strip().startswith("<?xml"):
if self.is_playready and license_message.lstrip().startswith("<"):
license_message = license_message.encode("utf-8")
else:
try:
+1 -1
View File
@@ -1209,7 +1209,7 @@ class DASH:
("urn:mpeg:dash:role:2011", "descriptive"),
("urn:tva:metadata:cs:AudioPurposeCS:2007", "1"),
)
for x in adaptation_set.findall("Accessibility")
for x in (*adaptation_set.findall("Accessibility"), *adaptation_set.findall("Role"))
)
@staticmethod
+24 -1
View File
@@ -17,7 +17,7 @@ from collections import defaultdict
from datetime import datetime, timezone
from pathlib import Path
from types import ModuleType
from typing import Any, Callable, Optional, Sequence, Union
from typing import Any, Callable, Mapping, Optional, Sequence, Union
from urllib.parse import ParseResult, urlparse
from uuid import uuid4
@@ -235,6 +235,29 @@ def keep_forced_subtitle(
return match_func(language, forced_s_lang)
def kind_only_conflict(params: Mapping[str, Any]) -> Optional[tuple[str, str]]:
"""
Return the first (kind-only flag, flag that drops its track type) pair that are both set.
The ``*_only`` track-type flags combine, so another one only drops the kind-only
flag's track type when its own ``*_only`` flag is not set as well.
"""
for flag, own_only, no_flag in (
("audio_description_only", "audio_only", "no_audio"),
("forced_subs_only", "subs_only", "no_subs"),
):
if not params.get(flag):
continue
if params.get(no_flag):
return flag, no_flag
if params.get(own_only):
continue
for other in ("video_only", "audio_only", "subs_only", "chapters_only"):
if params.get(other):
return flag, other
return None
def embedded_audio_langs(videos: Sequence[Any], keep_videos: bool) -> list[str]:
"""
Return the audio languages carried inside video tracks rather than beside them.