commit 08e66f21d7c5482b759ca0790c92e49ea3f5d105 Author: √(noham)² <100566912+NohamR@users.noreply.github.com> Date: Sat May 23 19:13:16 2026 +0200 Initial commit: dashdrm-mp4ff plugin Add a Streamlink plugin (plugin/dashdrm_mp4ff.py) to support DRM-protected MPEG-DASH streams with optional mp4ff-decrypt integration. Include README with usage and installation notes, an interactive helper (gen.py) that reads vault.json and prints Streamlink commands, a vault.json example, pyproject.toml (declares inquirerpy dependency), .python-version, and .gitignore. This provides the plugin, packaging metadata, and tooling needed to run and test DASH-DRM streams locally. diff --git a/.gitignore b/.gitignore new file mode 100644 index 0000000..f844245 --- /dev/null +++ b/.gitignore @@ -0,0 +1,3 @@ +vault.json +.DS_Store +*__pycache__ \ No newline at end of file diff --git a/.python-version b/.python-version new file mode 100644 index 0000000..2c07333 --- /dev/null +++ b/.python-version @@ -0,0 +1 @@ +3.11 diff --git a/README.md b/README.md new file mode 100644 index 0000000..0cc9744 --- /dev/null +++ b/README.md @@ -0,0 +1,75 @@ +# dashdrm-mp4ff + +Custom Streamlink plugin for DRM-protected MPEG-DASH streams. + +This project is based on [titus-au/streamlink-plugin-dashdrm](https://github.com/titus-au/streamlink-plugin-dashdrm), which itself is based on Streamlink's [`dash.py`](https://github.com/streamlink/streamlink/blob/master/src/streamlink/stream/dash/dash.py). + +## Install `mp4ff-decrypt` + +```bash +go install github.com/Eyevinn/mp4ff/cmd/mp4ff-decrypt@latest +``` + +Make sure the Go bin directory is in your `PATH`, then verify: +```bash +mp4ff-decrypt -h +``` + +If you do not want to use `mp4ff-decrypt`, you can still use the plugin without `--dashdrm-mp4ff-use-mp4ff`. + +## Usage + +Run Streamlink with the local plugin directory: +```bash +streamlink --plugin-dirs "plugin" "dashdrm://https://example.com/live.mpd" best +``` + +With `mp4ff-decrypt` and `kid:key` pairs: + +```bash +streamlink \ + --plugin-dirs "plugin" \ + --dashdrm-mp4ff-use-mp4ff \ + --dashdrm-mp4ff-decryption-key "kid1:key1,kid2:key2,kid3:key3" \ + "dashdrm://https://example.com/live.mpd" \ + best +``` + +Notes: + +- Keys can be provided as raw keys or as `kid:key`. +- When `kid:key` pairs are provided, the plugin parses the manifest and matches each track with the correct key. +- The plugin file is [`plugin/dashdrm_mp4ff.py`](./plugin/dashdrm_mp4ff.py). + +## `gen.py` + +[`gen.py`](./gen.py) is a small helper that reads channels from `vault.json`, lets you pick one interactively, and prints the full Streamlink command. + +Install the helper dependency: + +```bash +uv sync +``` + +Create your own `vault.json` from [`vault.json.example`](./vault.json.example), then run: +```bash +python gen.py +``` + +The expected structure is: +```json +{ + "channels": [ + { + "id": "channel-id", + "number": "1", + "name": "Channel name", + "mpd_url": "https://example.com/live.mpd", + "keys": [ + "kid1:key1", + "kid2:key2" + ] + } + ] +} +``` \ No newline at end of file diff --git a/gen.py b/gen.py new file mode 100644 index 0000000..0ccfcf8 --- /dev/null +++ b/gen.py @@ -0,0 +1,46 @@ +import json +from InquirerPy import prompt +from InquirerPy.base.control import Choice + +VAULT_FILE = "vault.json" + +def load_vault(): + try: + with open(VAULT_FILE, "r") as f: + return json.load(f)["channels"] + except FileNotFoundError: + return [] + +def main(): + channels = load_vault() + + if not channels: + print("No channels found in vault.json") + return + + choices = [ + Choice(value=channel, name=f"{channel.get('number', '?')} - {channel.get('name', 'Unknown')}") + for channel in channels + ] + + questions = [ + { + "type": "list", + "name": "channel", + "message": "Select a channel to generate the streamlink command for:", + "choices": choices, + } + ] + + result = prompt(questions) + selected_channel = result.get("channel") + + if selected_channel: + mpd_url = selected_channel.get("mpd_url", "") + keys_list = selected_channel.get("keys", []) + keys_str = ",".join(keys_list) + command = f'streamlink -l debug --plugin-dirs "./plugin" --dashdrm-mp4ff-use-mp4ff --dashdrm-mp4ff-decryption-key "{keys_str}" "dashdrm://{mpd_url}" best' + print(command) + +if __name__ == "__main__": + main() diff --git a/plugin/dashdrm_mp4ff.py b/plugin/dashdrm_mp4ff.py new file mode 100644 index 0000000..6a3d2dc --- /dev/null +++ b/plugin/dashdrm_mp4ff.py @@ -0,0 +1,1075 @@ +from __future__ import annotations + +import re +import itertools +import base64 +import queue +import subprocess +import tempfile +import os +from collections import defaultdict +from contextlib import suppress +from typing import List, Self +from datetime import timedelta + +from streamlink.exceptions import PluginError, FatalPluginError +from streamlink.plugin import pluginmatcher, pluginargument +from streamlink.plugin.plugin import (HIGH_PRIORITY, + parse_params, + stream_weight) +from streamlink.stream.dash import (DASHStream, + DASHStreamWorker, + DASHStreamWriter, + DASHStreamReader) +from streamlink.stream.dash.manifest import MPD, Representation +from streamlink.stream.dash.segment import DASHSegment +from streamlink.stream.ffmpegmux import FFMPEGMuxer +from streamlink.utils.url import update_scheme +from streamlink.session import Streamlink +from streamlink.utils.l10n import Language +from streamlink.utils.times import fromtimestamp, now +from streamlink.logger import getLogger +from streamlink.plugins.dash import MPEGDASH + +from streamlink.utils.parse import parse_xml +from typing import Any +from collections.abc import Mapping + + +period_sync_queue = queue.Queue() +rep_sync_queue = queue.Queue() + +log = getLogger(__name__) + + +class _DecryptedResponse: + """ + Wraps decrypted bytes to mimic the requests.Response interface + expected by DASHStreamWriter.write() (which calls .iter_content()). + """ + def __init__(self, data: bytes): + self._data = data + + def iter_content(self, chunk_size: int = 8192): + for i in range(0, len(self._data), chunk_size): + yield self._data[i:i + chunk_size] + + +DASHDRM_OPTIONS = [ + "decryption-key", + "presentation-delay", + "use-subtitles", + "ignore-location", + "ignore-availability", + "availability-grace", + "always-play-last-period", + "video-codec-preset", + "video-timescale", + "disable-multi-audio", + "language", + "use-mp4ff", +] + + +@pluginmatcher( + priority=HIGH_PRIORITY, + pattern=re.compile(r"dashdrm://(?P\S+)(?:\s(?P.+))?$"), +) +@pluginargument( + "decryption-key", + type="comma_list", + help="Decryption key(s) to be passed to ffmpeg." +) +@pluginargument( + "presentation-delay", + help="Override presentation delay value (in seconds). Similar to" + " --hls-live-edge." +) +@pluginargument( + "use-subtitles", + action="store_true", + help="Experiment with subtitles." +) +@pluginargument( + "ignore-location", + action="store_true", + help="Workaround to ignore Location tags that is not compliant resulting" + " in wrong segment URLs." +) +@pluginargument( + "ignore-availability", + action="store_true", + help="Workaround to ignore segment availability times where server has" + " wrong availability segment times (eg times that are in the future)." + " Use this with extreme caution as you are ignoring what the server is" + " telling you in relation to when a segment is available, thus you can" + " overwhelm a server with requests for segments that it already told" + " you is not (yet) available." +) +@pluginargument( + "availability-grace", + help="Workaround to delay getting segments even though the segment" + " availability time has been reached, as the segments are not" + " actually avaiable yet (resulting in 403/404 errors) possibly due to" + " mismatched server clock. A negative grace value can also be given," + " which can cover cases where the availability times are in the future." +) +@pluginargument( + "always-play-last-period", + action="store_true", + help="Always jump to the last period, even when multiple new periods are" + " found." +) +@pluginargument( + "video-codec-preset", + help="Use this to specific preset for transcode. This option is meant to be" + " paired with --ffmpeg-video-transcode. eg transcoding to libx264 with a" + " specific preset (eg ultrafast)" +) +@pluginargument( + "video-timescale", + help="Use this to only display videos with the specifie timescale (eg 50000)" + " filtering out any other. This can be used in multi-period mpds to display" + " the main content and filter out other content (eg ads) that has incompatible" + " timescale which can cause issues with client/players. Warning: Experimental" +) +@pluginargument( + "disable-multi-audio", + action="store_true", + help="Restore standard dash plugin function of single audio stream." +) +@pluginargument( + "language", + help="Allow filtering of audio to a specific language code." +) +@pluginargument( + "use-mp4ff", + action="store_true", + help="Use mp4ff-decrypt instead of ffmpeg for DRM decryption. " + "Requires mp4ff-decrypt to be installed and available in PATH. " +) +class MPEGDASHDRM(MPEGDASH): + + def _get_streams(self): + data = self.match.groupdict() + url = update_scheme("https://", str(data.get("url", "")), force=False) + params = parse_params(data.get("params")) + log.debug(f"URL={url}; params={params}") + + # process and store plugin options before passing streams back + for option in DASHDRM_OPTIONS: + if option == 'decryption-key': + if self.get_option('decryption-key'): + self.session.options[option] = self._process_keys() + elif option == 'always-play-last-period': + if not params.get('period'): + params['period'] = -1 + else: + self.session.options[option] = self.get_option(option) + + return DASHStreamDRM.parse_manifest(self.session, url, **params) + + @staticmethod + def _extract_kid(obj): + if not getattr(obj, 'contentProtections', None): + return None + + for cp in obj.contentProtections: + # cp is a streamlink ContentProtection (MPDNode subclass). + # cp.node is the underlying lxml _Element. + try: + raw_attribs = dict(cp.node.attrib) + except AttributeError: + continue + + for attr_key, attr_val in raw_attribs.items(): + # Strip namespace URI if present: "{urn:...}default_KID" → "default_KID" + local_name = attr_key.split('}')[-1] if '}' in attr_key else attr_key + if local_name.lower() in ('default_kid',): + kid = attr_val.replace('-', '').replace(' ', '').lower() + if len(kid) == 32: + log.debug('Found KID in ContentProtection attr %s: %s', attr_key, kid) + return kid + + return None + + @staticmethod + def _decode_key_hex(raw_key: str) -> str: + key_len = len(raw_key) + log.debug('Decryption Key %s has %s digits', raw_key, key_len) + if key_len in (21, 22, 23, 24): + # Looks like base64 – try to decode it + log.debug("Decryption key length is too short to be hex and looks like it might be base64, so we'll try and decode it..") + b64_string = raw_key + padding = 4 - (len(b64_string) % 4) + b64_string = b64_string + ("=" * padding) + decoded = base64.urlsafe_b64decode(b64_string).hex() + if decoded: + raw_key = decoded + key_len = len(raw_key) + log.debug( + 'Decryption Key (post base64 decode) is %s and has %s digits', + raw_key, key_len + ) + if key_len == 32: + try: + int(raw_key, 16) + except ValueError as err: + raise FatalPluginError("Expecting 128bit key in 32 hex digits, but the key contains invalid hex.") + elif key_len != 32: + raise FatalPluginError("Expecting 128bit key in 32 hex digits.") + return raw_key.lower() + + def _process_keys(self): + keys = self.get_option('decryption-key') + kid_key_map = {} # {kid_norm: key_hex} + ordered_keys = [] # positional fallback list + for k in keys: + parts = k.split(':') + if len(parts) == 2: + # Format is kid:key + kid_raw = parts[0] + key_raw = parts[1] + kid_norm = kid_raw.replace('-', '').lower() + else: + # Plain key (no kid) + kid_norm = None + key_raw = parts[-1] + + key_hex = self._decode_key_hex(key_raw) + if kid_norm: + kid_key_map[kid_norm] = key_hex + log.debug('KID %s -> Key %s', kid_norm, key_hex) + ordered_keys.append(key_hex) + + log.debug('kid_key_map: %s', kid_key_map) + log.debug('ordered_keys: %s', ordered_keys) + return {'kid_key_map': kid_key_map, 'ordered_keys': ordered_keys} + + +class FFMPEGMuxerDRM(FFMPEGMuxer): + + @classmethod + def _get_keys(cls, session): + keys = [] + if session.options.get("use-mp4ff"): + return keys + key_opt = session.options.get("decryption-key") + if key_opt: + if isinstance(key_opt, dict): + keys = list(key_opt.get("ordered_keys", [])) + else: + keys = list(key_opt) + if len(keys) == 1: + keys.extend(keys) + log.debug('Decryption Keys %s', keys) + return keys + + def __init__(self, session, *streams, **options): + # stream_kids is injected by DASHStreamDRM.open(); pop it before + # calling super().__init__() so FFMPEGMuxer doesn't choke on it. + super().__init__(session, *streams, **options) + keys = self._get_keys(session) + key = 0 + subtitles = self.session.options.get("use-subtitles") + vid_codec_preset = None + if session.options.get("video-codec-preset"): + vid_codec_preset = self.session.options.get("video-codec-preset") + + # Rebuild the ffmpeg command inserting -decryption_key before each -i + old_cmd = self._cmd.copy() + self._cmd = [] + + while len(old_cmd) > 0: + cmd = old_cmd.pop(0) + if cmd == "-i": + _ = old_cmd.pop(0) + if keys: + self._cmd.extend(["-decryption_key", keys[key]]) + key += 1 + if key == len(keys): + key = 1 + self._cmd.extend(['-thread_queue_size', '4096']) + self._cmd.extend([cmd, _]) + elif subtitles and cmd == "-c:a": + _ = old_cmd.pop(0) + self._cmd.extend([cmd, _]) + self._cmd.extend(["-c:s", "copy"]) + elif vid_codec_preset and cmd == "-c:v": + _ = old_cmd.pop(0) + self._cmd.extend([cmd, _]) + self._cmd.extend(["-preset:v", vid_codec_preset]) + else: + self._cmd.append(cmd) + log.debug("Updated ffmpeg command %s", self._cmd) + + +class DASHStreamWriterDRM(DASHStreamWriter): + reader: DASHStreamReaderDRM + stream: DASHStreamDRM + + def __init__(self, *args, **kwargs): + super().__init__(*args, **kwargs) + self._init_segment_path: str | None = None + + def __del__(self): + self._cleanup_init_segment() + + def _cleanup_init_segment(self): + if self._init_segment_path: + with suppress(OSError): + os.unlink(self._init_segment_path) + self._init_segment_path = None + + def _get_key_for_stream(self) -> str | None: + # Only set if no key was already configured (initial URL / CLI takes priority) + key_opt = self.session.options.get("decryption-key") + if not key_opt: + return None + + if isinstance(key_opt, dict): + kid_key_map = key_opt.get("kid_key_map", {}) + ordered_keys = key_opt.get("ordered_keys", []) + else: + # Parse the raw kid:key strings into the expected dict format + kid_key_map = {} + ordered_keys = key_opt + + if not ordered_keys: + return None + if len(ordered_keys) == 1: + return ordered_keys[0] + + stream_kid = getattr(self.reader, "_drm_kid", None) + if stream_kid and kid_key_map: + kid_norm = stream_kid.replace('-', '').lower() + if kid_norm in kid_key_map: + return kid_key_map[kid_norm] + + mime = getattr(self.reader, "mime_type", "") + if mime.startswith("video"): + return ordered_keys[0] + + reader_name = getattr(self.reader, "name", "") + try: + stream_index = int("".join(filter(str.isdigit, reader_name))) + except (ValueError, TypeError): + stream_index = 0 + + audio_key_count = len(ordered_keys) - 1 + key_index = 1 + (stream_index % audio_key_count) + return ordered_keys[key_index] + + def _decrypt_segment(self, data: bytes, is_init: bool = False) -> bytes: + # Only set if no key was already configured (initial URL / CLI takes priority) + key = self._get_key_for_stream() + if not key: + log.warning("No decryption key available for mp4ff-decrypt, returning raw data") + return data + + # Pour l'init segment : sauvegarder le brut AVANT déchiffrement + if is_init: + self._cleanup_init_segment() + fd, self._init_segment_path = tempfile.mkstemp(suffix="_init_raw.mp4") + os.close(fd) + with open(self._init_segment_path, "wb") as f: + f.write(data) + log.debug( + "%s raw init segment saved to %s", + self.reader.mime_type, + self._init_segment_path, + ) + + with tempfile.TemporaryDirectory() as tmpdir: + input_path = os.path.join(tmpdir, "segment.mp4") + output_path = os.path.join(tmpdir, "segment_dec.mp4") + + with open(input_path, "wb") as f: + f.write(data) + + if is_init or not self._init_segment_path: + cmd = ["mp4ff-decrypt", "-key", key, input_path, output_path] + else: + cmd = [ + "mp4ff-decrypt", + "-init", self._init_segment_path, + "-key", key, + input_path, + output_path, + ] + + log.debug("%s mp4ff-decrypt cmd: %s", self.reader.mime_type, cmd) + + try: + result = subprocess.run(cmd, capture_output=True, timeout=30) + except FileNotFoundError: + raise FatalPluginError( + "mp4ff-decrypt not found. Please install mp4ff and make sure " + "it is available in your PATH." + ) + except subprocess.TimeoutExpired: + log.error("mp4ff-decrypt timed out on segment, returning raw data") + return data + + if result.returncode != 0: + log.error( + "mp4ff-decrypt failed (exit %d): %s", + result.returncode, + result.stderr.decode(errors="replace"), + ) + return data + + with open(output_path, "rb") as f: + decrypted = f.read() + + return decrypted + + def fetch(self, segment: DASHSegment): + if self.closed: + return + + real_available_in = (segment.available_at - now()).total_seconds() + name = segment.name + log.debug( + f"{self.reader.mime_type} segment {name}: " + f"Available in {real_available_in:.01f}s ({segment.availability})" + ) + + if self.session.options.get("availability-grace"): + availability_grace = float(self.session.options.get("availability-grace")) + segment.available_at = segment.available_at + timedelta(seconds=availability_grace) + log.debug( + f"{self.reader.mime_type} segment {name}: " + f"Adding {availability_grace} seconds to segment availability" + ) + + if self.session.options.get("ignore-availability"): + segment.available_at = fromtimestamp(0) + log.debug(f"Ignoring availability timestamps. Now available at {segment.available_at}") + + response = super().fetch(segment) + + if not self.session.options.get("use-mp4ff"): + return response + + if response is None: + return response + + raw = b"".join(response.iter_content(8192)) + if not raw: + return response + + decrypted = self._decrypt_segment(raw, is_init=bool(segment.init)) + return _DecryptedResponse(decrypted) + + +class DASHStreamWorkerDRM(DASHStreamWorker): + reader: DASHStreamReaderDRM + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + def iter_segments(self): + init = True + back_off_factor = 1 + new_rep = None + queued = False + last_segment = None + first = True + filtered = False + video_is_filtered = False + repeated = 0 + while not self.closed: + # find the representation by ID + representation = self.mpd.get_representation(self.reader.ident) + + # check if a new representation is available + if not first and not new_rep: + new_rep, video_is_filtered = self.check_new_rep() + elif first: + _, filtered = self.check_new_rep(representation) + if filtered: + representation = None + log.debug("First run. Video filter %s.", filtered) + + first = False + if self.mpd.type == "static": + refresh_wait = 5 + else: + refresh_wait = ( + max( + self.mpd.minimumUpdatePeriod.total_seconds(), + # dont take the whole rep duration as wait time + # as some mpd will set a large number. we then + # end up staying in the sleeper loop too long + # and ffmpeg will timeout + min(representation.period.duration.total_seconds(), 5) + if representation else 0, + ) + or 5 + ) + + if new_rep and not queued: + # New rep available and no yield so we swap to the new one + self.reader.ident = new_rep.ident + representation = new_rep + new_rep = None + log.debug("New period and no yield. Swapping to next period.") + if video_is_filtered: + log.debug("New period %s marked to be filtered.", self.reader.ident) + filtered = True + representation = None + elif new_rep and queued: + # New rep available but we had yield so we dont swap yet. + # Set refresh to be very low since we know we actually have + # new content in the from of new_rep + refresh_wait = 1 + log.debug("New period and but we have yield. Not swapping to next period yet.") + + with self.sleeper(refresh_wait * back_off_factor): + if not representation: + if filtered: + log.debug("Period is marked to be filtered.") + if last_segment: + repeated += 1 + log.debug("Repeating last segment %s times.", repeated) + yield last_segment + else: + log.debug("No last segment to repeat.") + self.reload() + continue + queued = False + iter_segments = representation.segments( + sequence=self.sequence, + init=init, + # sync initial timeline generation between audio and video threads + timestamp=self.reader.timestamp if init else None, + ) + for segment in iter_segments: + if init and not segment.init: + self.sequence = segment.num + init = False + last_segment = segment + repeated = 0 + queued |= yield segment + + # close worker if type is not dynamic (all segments were put into writer queue) + if self.mpd.type != "dynamic": + self.close() + return + + # Implicit end of stream + + #if self.check_queue_deadline(queued): + # return + + if not self.reload(): + # use min instead of max to limit run-away backoff + back_off_factor = min(back_off_factor * 1.3, 10.0) + else: + back_off_factor = 1 + + +class DASHStreamWorkerDRMVideo(DASHStreamWorkerDRM): + reader: DASHStreamReaderDRMVideo + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + def rep_check_filtered(self, first_rep=None): + if not first_rep: + representation = self.mpd.get_representation(self.reader.ident) + else: + representation = first_rep + if representation.segmentTemplate: + current_timescale = representation.segmentTemplate.timescale + elif representation.segmentList: + current_timescale = representation.segmentList.timescale + else: + current_timescale = None + log.debug("Current rep (%s) timescale: %s", self.reader.mime_type, current_timescale) + + is_filtered = False + if self.session.options.get("video-timescale"): + wanted_timescale = int(self.session.options.get("video-timescale")) + if current_timescale and (current_timescale != wanted_timescale): + is_filtered = True + return is_filtered + + def next_period_available(self): + period_id = self.reader.ident[0] + current_period_ids = [p.id for p in self.mpd.periods] + current_period_idx = current_period_ids.index(period_id) + + log.debug("Current playing period: %s", current_period_idx + 1) + log.debug("Number of periods: %s", len(current_period_ids)) + + next_period = 0 + if len(current_period_ids) > current_period_idx + 1: + if self.session.options.get("always-play-last-period"): + next_period = current_period_ids[-1] + else: + next_period = current_period_ids[current_period_idx + 1] + + audio_count = len(self.stream.audio_representations) + if audio_count == 1 and self.stream.audio_representations == [None]: + audio_count = 0 + if audio_count > 0: + for _ in range(audio_count): + period_sync_queue.put(next_period) + + subtitles_count = len(self.stream.subtitles_representations) + if subtitles_count == 1 and self.stream.subtitles_representations == [None]: + subtitles_count = 0 + if subtitles_count > 0: + for _ in range(subtitles_count): + period_sync_queue.put(next_period) + + return next_period + + def check_new_rep(self, first_rep=None): + new_video_rep = None + is_filtered = False + audio_count = len(self.stream.audio_representations) + if audio_count == 1 and self.stream.audio_representations == [None]: + audio_count = 0 + subtitles_count = len(self.stream.subtitles_representations) + if subtitles_count == 1 and self.stream.subtitles_representations == [None]: + subtitles_count = 0 + log.debug("Audio stream count: %s", audio_count) + log.debug("Subtitles stream count: %s", subtitles_count) + + if first_rep: + is_filtered = self.rep_check_filtered(first_rep) + log.debug("Telling non-video streams first rep and filter") + if audio_count > 0: + for _ in range(audio_count): + rep_sync_queue.put((new_video_rep, is_filtered)) + if subtitles_count > 0: + for _ in range(subtitles_count): + rep_sync_queue.put((new_video_rep, is_filtered)) + log.debug("Current rep_sync queue size: %s", rep_sync_queue.qsize()) + return (None, is_filtered) + + log.debug("Checking for new period and representations in video stream") + next_period = self.next_period_available() + if next_period: + # reparse manifest to find the next stream + reloaded_streams = DASHStreamDRM.parse_manifest(self.session, self.mpd.url, next_period) + p, a, r = self.reader.ident + new_video_rep = self.mpd.get_representation((next_period, a, r)) + if new_video_rep: + log.debug("New video rep found. New ident: %s", new_video_rep.ident) + else: + log.debug("New period found, but can't find matching video rep, trying to find the stream the old way") + reload_stream = reloaded_streams[self.stream.stream_name] + new_video_rep = reload_stream.video_representation + if new_video_rep: + log.debug("New video representation found!") + + is_filtered = self.rep_check_filtered(new_video_rep) + + log.debug("Telling non-video streams new rep and filter") + if rep_sync_queue.qsize() > 0: + log.debug("Current rep_sync queue size is non-zero (%s), waiting 5 seconds", rep_sync_queue.qsize()) + self.sleeper(5) + if rep_sync_queue.qsize() > 0: + log.debug("Current rep_sync queue size is still non-zero (%s).", rep_sync_queue.qsize()) + if audio_count > 0: + for _ in range(audio_count): + rep_sync_queue.put((new_video_rep, is_filtered)) + if subtitles_count > 0: + for _ in range(subtitles_count): + rep_sync_queue.put((new_video_rep, is_filtered)) + log.debug("Current rep_sync queue size: %s", rep_sync_queue.qsize()) + return (new_video_rep, is_filtered) + + +class DASHStreamWorkerDRMNonVideo(DASHStreamWorkerDRM): + reader: DASHStreamReaderDRM + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + def next_period_available(self): + period_id = self.reader.ident[0] + current_period_ids = [p.id for p in self.mpd.periods] + current_period_idx = current_period_ids.index(period_id) + + log.debug("Current playing period: %s", current_period_idx + 1) + log.debug("Number of periods: %s", len(current_period_ids)) + next_period = period_sync_queue.get(block=True) + return next_period + + def check_new_rep(self, stream_type=None, first_rep=None): + new_stream_rep = None + is_filtered = False + if first_rep: + log.debug("Getting from video streams first rep and filter") + log.debug("Current rep_sync queue size just before get: %s", rep_sync_queue.qsize()) + _, is_filtered = rep_sync_queue.get(block=True) + return (None, is_filtered) + + log.debug("Checking new reps from video stream") + next_period = self.next_period_available() + if next_period: + log.debug("Got next period %s from video stream", next_period) + reloaded_streams = DASHStreamDRM.parse_manifest(self.session, self.mpd.url, next_period) + _, a, r = self.reader.ident + new_stream_rep = self.mpd.get_representation((next_period, a, r)) + if new_stream_rep: + log.debug("New %s rep found. New ident: %s", stream_type, new_stream_rep.ident) + else: + log.debug("New period found, but can't find matching %s rep, trying the old way", stream_type) + reload_stream = reloaded_streams[self.stream.stream_name] + if stream_type == "audio": + audio_num = int(self.name[-1]) + new_stream_rep = reload_stream.audio_representations[audio_num] + elif stream_type == "sub": + new_stream_rep = reload_stream.subtitles_representation + if new_stream_rep: + log.debug("New %s representation found!", stream_type) + + log.debug("Checking video rep and filter status from video stream") + log.debug("Current rep_sync queue size just before get: %s", rep_sync_queue.qsize()) + new_video_rep, is_filtered = rep_sync_queue.get(block=True) + log.debug("Video stream sent video rep %s and filter status %s", new_video_rep, is_filtered) + log.debug("New video rep: %s, isfiltered: %s", new_video_rep.ident, is_filtered) + return (new_stream_rep, is_filtered) + return (None, False) + + +class DASHStreamWorkerDRMAudio(DASHStreamWorkerDRMNonVideo): + reader: DASHStreamReaderDRMAudio + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + def check_new_rep(self, first_rep=None): + return super().check_new_rep(stream_type="audio", first_rep=first_rep) + + +class DASHStreamWorkerDRMSub(DASHStreamWorkerDRMNonVideo): + reader: DASHStreamReaderDRMSub + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + def check_new_rep(self, first_rep=None): + return super().check_new_rep(stream_type="sub", first_rep=first_rep) + + +class DASHStreamReaderDRM(DASHStreamReader): + __worker__ = DASHStreamWorkerDRM + __writer__ = DASHStreamWriterDRM + + worker: DASHStreamWorkerDRM + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + +class DASHStreamReaderDRMVideo(DASHStreamReaderDRM): + __worker__ = DASHStreamWorkerDRMVideo + __writer__ = DASHStreamWriterDRM + + worker: DASHStreamWorkerDRMVideo + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + +class DASHStreamReaderDRMAudio(DASHStreamReaderDRM): + __worker__ = DASHStreamWorkerDRMAudio + __writer__ = DASHStreamWriterDRM + + worker: DASHStreamWorkerDRMAudio + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + +class DASHStreamReaderDRMSub(DASHStreamReaderDRM): + __worker__ = DASHStreamWorkerDRMSub + __writer__ = DASHStreamWriterDRM + + worker: DASHStreamWorkerDRMSub + writer: DASHStreamWriterDRM + stream: DASHStreamDRM + + def read(self, size: int) -> bytes: + _ = self.buffer.read( + size, + block=self.writer.is_alive(), + timeout=self.timeout, + ) + log.debug("Subtitle stream segment: %s", _) + return _ + + +class DASHStreamDRM(DASHStream): + """ + Implementation of the MPEG-DASH protocol with DRM support. + """ + + __shortname__ = "dashdrm-mp4ff" + + def __init__( + self, + session: Streamlink, + mpd: MPD, + video_representation: Representation | None = None, + audio_representations: List[Representation] | None = None, + subtitles_representations: List[Representation] | None = None, + duration: float | None = None, + **kwargs, + ): + super().__init__( + session, + mpd, + video_representation, + audio_representations[0] if audio_representations[0] else None, + duration, + **kwargs, + ) + self.audio_representations = audio_representations + self.subtitles_representations = subtitles_representations + + @staticmethod + def parse_mpd(session, manifest: str, mpd_params: Mapping[str, Any]) -> MPD: + node = parse_xml(manifest, ignore_ns=True) + if session.options.get("ignore-location"): + location = node.find('Location') + if location is not None: + log.warning('Found Location tag: %s', location.text) + parent = location.getparent() + parent.remove(location) + return MPD(node, **mpd_params) + + @classmethod + def parse_manifest( + cls, + session: Streamlink, + url_or_manifest: str, + period: int | str = 0, + with_video_only: bool = False, + with_audio_only: bool = False, + **kwargs, + ) -> dict[str, DASHStreamDRM]: + + manifest, mpd_params = cls.fetch_manifest(session, url_or_manifest, **kwargs) + + # Case 2: decryption_key may appear in the final URL after an API redirect. + # fetch_manifest returns res.url (post-redirect), so we check it here. + # This covers the case where the initial URL is an API endpoint that + # redirects to the real MPD URL with ?decryption_key=... appended. + try: + mpd = cls.parse_mpd(session, manifest, mpd_params) + except Exception as err: + raise PluginError(f"Failed to parse MPD manifest: {err}") from err + + if session.options.get("presentation-delay"): + presentation_delay = session.options.get("presentation-delay") + mpd.suggestedPresentationDelay = timedelta(seconds=int(presentation_delay)) + + source = mpd_params.get("url", "MPD manifest") + video: list[Representation | None] = [None] if with_audio_only else [] + audio: list[Representation | None] = [None] if with_video_only else [] + subtitles: list[Representation | None] = [None] if with_audio_only else [] + + available_periods = [ + f"{idx}{f' (id={p.id!r})' if p.id is not None else ''}" + for idx, p in enumerate(mpd.periods) + ] + log.debug(f"Available DASH periods: {', '.join(available_periods)}") + + try: + if isinstance(period, int): + period_selection = mpd.periods[period] + else: + period_selection = mpd.periods_map[period] + except LookupError: + raise PluginError( + f"DASH period {period!r} not found. Select a valid period by index or by id attribute value.", + ) from None + + # Search for suitable video and audio representations + for aset in period_selection.adaptationSets: + # Extract the KID from the AdaptationSet-level ContentProtection. + # The default_KID attribute can sit either on the ContentProtection + # element itself (urn:mpeg:dash:mp4protection:2011) or as a + # child element. We try several common attribute names. + aset_kid = MPEGDASHDRM._extract_kid(aset) + if aset_kid: + log.debug('AdaptationSet KID: %s', aset_kid) + if aset.contentProtections: + if not session.options.get("decryption-key"): + raise PluginError(f"{source} is protected by DRM but no key given") + else: + log.debug(f"{source} is protected by DRM") + for rep in aset.representations: + # KID on the Representation overrides the AdaptationSet KID + rep_kid = MPEGDASHDRM._extract_kid(rep) or aset_kid + # Attach the KID to the representation so open() can pass it + # to FFMPEGMuxerDRM for key matching + rep._drm_kid = rep_kid + if rep_kid: + log.debug('Representation %s KID: %s', rep.id, rep_kid) + if rep.contentProtections: + if not session.options.get("decryption-key"): + raise PluginError(f"{source} is protected by DRM but no key given") + else: + log.debug(f"{source} is protected by DRM") + if rep.mimeType.startswith("video"): + video.append(rep) + elif rep.mimeType.startswith("audio"): # pragma: no branch + audio.append(rep) + elif (session.options.get("use-subtitles") and + rep.mimeType.startswith("application")): + subtitles.append(rep) + + if not video: + video.append(None) + if not audio: + audio.append(None) + if not subtitles: + subtitles.append(None) + + locale = session.localization + locale_lang = locale.language + lang = None + available_languages = set() + + # if the locale is explicitly set, prefer that language over others + for aud in audio: + if aud and aud.lang: + available_languages.add(aud.lang) + with suppress(LookupError): + if locale.explicit and aud.lang and Language.get(aud.lang) == locale_lang: + lang = aud.lang + + if not lang: + # filter by the first language that appears + lang = audio[0].lang if audio[0] else None + + if session.options.get("language"): + lang = session.options.get("language") + + log.debug( + f"Available languages for DASH audio streams: " + f"{', '.join(available_languages) or 'NONE'} (using: {lang or 'n/a'})", + ) + + if session.options.get("disable-multi-audio"): + if len(available_languages) > 1: + audio = [a for a in audio if a and (a.lang is None or a.lang == lang)] + + # if the language is given by the stream, filter out other languages that do not match + #if len(available_languages) > 1: + # audio = [a for a in audio if a and (a.lang is None or a.lang == lang)] + + ret = [] + for vid, aud in itertools.product(video, audio): + if not vid and not aud: + continue + + # Pass only the selected audio representation, not the full list. + # The full list was causing open() to mux ALL audio tracks regardless + # of which stream the user selected (e.g. 540p+fr_96k would still + # open fr_250k, qaa_96k, qad_96k etc.) + + if session.options.get("disable-multi-audio"): + stream = DASHStreamDRM(session, mpd, vid, [aud], subtitles, **kwargs) + else: + stream = DASHStreamDRM(session, mpd, vid, audio, subtitles, **kwargs) + stream_name = [] + if vid: + stream_name.append(f"{vid.height or vid.bandwidth_rounded:0.0f}{'p' if vid.height else 'k'}") + if aud and len(audio) > 1: + stream_name.append(f"a{aud.bandwidth:0.0f}k") + ret.append(("+".join(stream_name), stream)) + + # rename duplicate streams + dict_value_list = defaultdict(list) + for k, v in ret: + dict_value_list[k].append(v) + + def sortby_bandwidth(dash_stream: DASHStreamDRM) -> float: + if dash_stream.video_representation: + return dash_stream.video_representation.bandwidth + if session.options.get("disable-multi-audio"): + if dash_stream.audio_representation: + return dash_stream.audio_representation.bandwidth + return 0 + + ret_new = {} + for q in dict_value_list: + items = dict_value_list[q] + with suppress(AttributeError): + items = sorted(items, key=sortby_bandwidth, reverse=True) + for n in range(len(items)): + if n == 0: + ret_new[q] = items[n] + elif n == 1: + ret_new[f"{q}_alt"] = items[n] + else: + ret_new[f"{q}_alt{n}"] = items[n] + + # add stream_name to the returned streams so we can find it again + for stream_name in ret_new: + ret_new[stream_name].stream_name = stream_name + + return ret_new + + def open(self): + video, audio, audio1 = None, None, None + rep_video = self.video_representation + rep_audios = self.audio_representations + rep_subtitles = self.subtitles_representations + + timestamp = now() + fds = [] + # stream_kids tracks the KID for each ffmpeg -i input, in order. + # This list is passed to FFMPEGMuxerDRM so it can match kid→key. + maps = ["0:v?", "0:a?"] + metadata = {} + + if rep_video: + video = DASHStreamReaderDRMVideo(self, rep_video, timestamp, name="video") + video._drm_kid = getattr(rep_video, "_drm_kid", None) + log.debug(f"Opening DASH reader for: {rep_video.ident!r} - {rep_video.mimeType}") + video.open() + fds.append(video) + + next_map = 1 + if rep_audios: + for i, rep_audio in enumerate(rep_audios): + audio = DASHStreamReaderDRMAudio(self, rep_audio, timestamp, name="audio" + str(i)) + audio._drm_kid = getattr(rep_audio, "_drm_kid", None) + if not audio1: + audio1 = audio + log.debug(f"Opening DASH reader for: {rep_audio.ident!r} - {rep_audio.mimeType}") + audio.open() + fds.append(audio) + metadata["s:a:{0}".format(i)] = [ + "language={0}".format(rep_audio.lang), + "title=\"{0}\"".format(rep_audio.lang), + ] + maps.extend(f"{i}:a" for i in range(next_map, next_map + len(rep_audios))) + next_map = len(rep_audios) + 1 + + # only do subtitles if we have video + if rep_subtitles and rep_subtitles[0] and rep_video: + for _, rep_subtitle in enumerate(rep_subtitles): + subtitle = DASHStreamReaderDRMSub(self, rep_subtitle, timestamp, name="subtitle" + str(_)) + subtitle._drm_kid = getattr(rep_subtitle, "_drm_kid", None) + log.debug(f"Opening DASH reader for: {rep_subtitle.ident!r} - {rep_subtitle.mimeType}") + subtitle.open() + fds.append(subtitle) + metadata["s:s:{0}".format(_)] = [ + "language={0}".format(rep_subtitle.lang), + "title=\"{0}\"".format(rep_subtitle.lang), + ] + maps.extend(f"{_}:s" for _ in range(next_map, next_map + len(rep_subtitles))) + + if video and audio and FFMPEGMuxerDRM.is_usable(self.session): + return FFMPEGMuxerDRM(self.session, *fds, copyts=True, maps=maps, metadata=metadata).open() + elif video: + return video + elif audio: + return audio1 + + +__plugin__ = MPEGDASHDRM diff --git a/pyproject.toml b/pyproject.toml new file mode 100644 index 0000000..836e3cb --- /dev/null +++ b/pyproject.toml @@ -0,0 +1,9 @@ +[project] +name = "dashdrm-mp4ff" +version = "0.1.0" +description = "Plugin for Streamlink to support DASH-DRM protected streams using mp4ff in option." +readme = "README.md" +requires-python = ">=3.11" +dependencies = [ + "inquirerpy>=0.3.4", +] diff --git a/uv.lock b/uv.lock new file mode 100644 index 0000000..3704a3a --- /dev/null +++ b/uv.lock @@ -0,0 +1,57 @@ +version = 1 +revision = 3 +requires-python = ">=3.11" + +[[package]] +name = "dashdrm-mp4ff" +version = "0.1.0" +source = { virtual = "." } +dependencies = [ + { name = "inquirerpy" }, +] + +[package.metadata] +requires-dist = [{ name = "inquirerpy", specifier = ">=0.3.4" }] + +[[package]] +name = "inquirerpy" +version = "0.3.4" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "pfzy" }, + { name = "prompt-toolkit" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/64/73/7570847b9da026e07053da3bbe2ac7ea6cde6bb2cbd3c7a5a950fa0ae40b/InquirerPy-0.3.4.tar.gz", hash = "sha256:89d2ada0111f337483cb41ae31073108b2ec1e618a49d7110b0d7ade89fc197e", size = 44431, upload-time = "2022-06-27T23:11:20.598Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/ce/ff/3b59672c47c6284e8005b42e84ceba13864aa0f39f067c973d1af02f5d91/InquirerPy-0.3.4-py3-none-any.whl", hash = "sha256:c65fdfbac1fa00e3ee4fb10679f4d3ed7a012abf4833910e63c295827fe2a7d4", size = 67677, upload-time = "2022-06-27T23:11:17.723Z" }, +] + +[[package]] +name = "pfzy" +version = "0.3.4" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/d9/5a/32b50c077c86bfccc7bed4881c5a2b823518f5450a30e639db5d3711952e/pfzy-0.3.4.tar.gz", hash = "sha256:717ea765dd10b63618e7298b2d98efd819e0b30cd5905c9707223dceeb94b3f1", size = 8396, upload-time = "2022-01-28T02:26:17.946Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/8c/d7/8ff98376b1acc4503253b685ea09981697385ce344d4e3935c2af49e044d/pfzy-0.3.4-py3-none-any.whl", hash = "sha256:5f50d5b2b3207fa72e7ec0ef08372ef652685470974a107d0d4999fc5a903a96", size = 8537, upload-time = "2022-01-28T02:26:16.047Z" }, +] + +[[package]] +name = "prompt-toolkit" +version = "3.0.52" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "wcwidth" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/a1/96/06e01a7b38dce6fe1db213e061a4602dd6032a8a97ef6c1a862537732421/prompt_toolkit-3.0.52.tar.gz", hash = "sha256:28cde192929c8e7321de85de1ddbe736f1375148b02f2e17edd840042b1be855", size = 434198, upload-time = "2025-08-27T15:24:02.057Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/84/03/0d3ce49e2505ae70cf43bc5bb3033955d2fc9f932163e84dc0779cc47f48/prompt_toolkit-3.0.52-py3-none-any.whl", hash = "sha256:9aac639a3bbd33284347de5ad8d68ecc044b91a762dc39b7c21095fcd6a19955", size = 391431, upload-time = "2025-08-27T15:23:59.498Z" }, +] + +[[package]] +name = "wcwidth" +version = "0.7.0" +source = { registry = "https://pypi.org/simple" } +sdist = { url = "https://files.pythonhosted.org/packages/2c/ee/afaf0f85a9a18fe47a67f1e4422ed6cf1fe642f0ae0a2f81166231303c52/wcwidth-0.7.0.tar.gz", hash = "sha256:90e3a7ea092341c44b99562e75d09e4d5160fe7a3974c6fb842a101a95e7eed0", size = 182132, upload-time = "2026-05-02T16:04:12.653Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/41/52/e465037f5375f43533d1a80b6923955201596a99142ed524d77b571a1418/wcwidth-0.7.0-py3-none-any.whl", hash = "sha256:5d69154c429a82910e241c738cd0e2976fac8a2dd47a1a805f4afed1c0f136f2", size = 110825, upload-time = "2026-05-02T16:04:11.033Z" }, +] diff --git a/vault.json.example b/vault.json.example new file mode 100644 index 0000000..5ea98cb --- /dev/null +++ b/vault.json.example @@ -0,0 +1,24 @@ +{ + "channels": [ + { + "id": "example-channel-1", + "number": "1", + "name": "Example Channel 1", + "mpd_url": "https://example.com/playlist/v1/live/123/1/live.mpd", + "keys": [ + "11111111111111111111111111111111:aaaaaaaaaaaaaaaaaaaaaaaaaaaaaaaa", + "22222222222222222222222222222222:bbbbbbbbbbbbbbbbbbbbbbbbbbbbbbbb" + ] + }, + { + "id": "example-channel-2", + "number": "2", + "name": "Example Channel 2", + "mpd_url": "https://example.com/playlist/v1/live/456/1/live.mpd", + "keys": [ + "33333333333333333333333333333333:cccccccccccccccccccccccccccccccc", + "44444444444444444444444444444444:dddddddddddddddddddddddddddddddd" + ] + } + ] +}