diff --git a/record_dynamic.py b/record_dynamic.py
index 8301006..2d8e1f7 100644
--- a/record_dynamic.py
+++ b/record_dynamic.py
@@ -6045,14 +6045,22 @@ def resolve_nm3u8dl_license_url(
separators=(",", ":"),
).encode("utf-8")
+ clean_license_url, license_headers = (
+ source_playlist_headers.split_stream_url_metadata(
+ license_url,
+ policy=source_playlist_headers.MATURE_PLAYLIST_HEADER_POLICY,
+ )
+ )
+ request_headers = {"User-Agent": "curl/8.21.0"}
+ request_headers.update(license_headers)
+ request_headers.update({
+ "Content-Type": "application/json",
+ "Accept": "application/json",
+ })
request = Request(
- license_url,
+ clean_license_url,
data=payload,
- headers={
- "Content-Type": "application/json",
- "Accept": "application/json",
- "User-Agent": "curl/8.21.0",
- },
+ headers=request_headers,
method="POST",
)
diff --git a/recorder_source/discovery.py b/recorder_source/discovery.py
index c795f41..a88a887 100644
--- a/recorder_source/discovery.py
+++ b/recorder_source/discovery.py
@@ -190,7 +190,15 @@ def normalize_playlist_license_key(value: str) -> Tuple[str, ...]:
if not text:
return ()
if not text.startswith("{"):
- return (text,)
+ if text.lower().startswith(("http://", "https://")) or "," not in text:
+ return (text,)
+ pairs = [pair.strip() for pair in text.split(",")]
+ if not all(
+ re.fullmatch(r"[0-9a-fA-F]{32}:[0-9a-fA-F]{32}", pair)
+ for pair in pairs
+ ):
+ raise RuntimeError("Invalid comma-separated ClearKey KID:KEY list")
+ return tuple(dict.fromkeys(pairs))
try:
data = json.loads(text)
except json.JSONDecodeError:
diff --git a/recorder_source/json_playlist.py b/recorder_source/json_playlist.py
index f297dcd..0ffc9c8 100644
--- a/recorder_source/json_playlist.py
+++ b/recorder_source/json_playlist.py
@@ -232,6 +232,19 @@ def _entry_lines(
if key_id and key_value:
license_key = f"{key_id}:{key_value}"
+ if not license_key:
+ sonuxs = _alias_value(
+ record, ("sonuxs",), first_wins=policy.first_record_alias_wins,
+ )
+ if isinstance(sonuxs, (list, tuple)):
+ pairs = [
+ value.strip()
+ for value in sonuxs
+ if isinstance(value, str) and value.strip()
+ ]
+ if pairs:
+ license_key = ",".join(pairs)
+
if license_key:
lines.extend([
"#KODIPROP:inputstream.adaptive.license_type=clearkey",
diff --git a/tests/test_drm_license_compat.py b/tests/test_drm_license_compat.py
new file mode 100644
index 0000000..7dd9749
--- /dev/null
+++ b/tests/test_drm_license_compat.py
@@ -0,0 +1,129 @@
+"""Regression tests for license-request-only headers without importing the recorder entrypoint."""
+
+from __future__ import annotations
+
+import ast
+import base64
+import io
+import json
+from pathlib import Path
+import re
+import unittest
+from typing import List
+from urllib.request import Request
+import xml.etree.ElementTree as ET
+
+from recorder_source import playlist_headers
+
+
+KID = "00112233445566778899aabbccddeeff"
+KEY = "ffeeddccbbaa99887766554433221100"
+STREAM_URL = "https://cdn.test/starsports1hd.mpd"
+LICENSE_URL = "https://license.test/starsports1hd.mpd?license=1"
+MANIFEST = (
+ ''
+ ''
+ ''
+)
+
+
+def _encode(hex_string: str) -> str:
+ return base64.urlsafe_b64encode(bytes.fromhex(hex_string)).decode("ascii").rstrip("=")
+
+
+def _decode(value: str) -> bytes:
+ return base64.urlsafe_b64decode(value + "=" * (-len(value) % 4))
+
+
+class LicenseRequestHeaderTests(unittest.TestCase):
+ def _run_resolver(self, license_key: str, *, through_selection: bool = True):
+ source = (Path(__file__).resolve().parents[1] / "record_dynamic.py").read_text(
+ encoding="utf-8"
+ )
+ parsed = ast.parse(source)
+ selected = [
+ node for node in parsed.body
+ if isinstance(node, ast.FunctionDef)
+ and node.name in {"resolve_nm3u8dl_license_url", "resolve_nm3u8dl_source_keys"}
+ ]
+ self.assertEqual(len(selected), 2)
+
+ captured_requests = []
+ manifest_calls = []
+
+ def fake_manifest(stream_url, headers):
+ manifest_calls.append((stream_url, dict(headers)))
+ return MANIFEST
+
+ def fake_urlopen(request, timeout):
+ captured_requests.append((request, timeout))
+ response = {"keys": [{"kid": _encode(KID), "k": _encode(KEY)}]}
+ return io.BytesIO(json.dumps(response).encode("utf-8"))
+
+ namespace = {
+ "List": List,
+ "ET": ET,
+ "json": json,
+ "re": re,
+ "Request": Request,
+ "urlopen": fake_urlopen,
+ "source_playlist_headers": playlist_headers,
+ "_fetch_nm3u8dl_stream_manifest_text": fake_manifest,
+ "_is_nm3u8dl_drmlive_host": lambda url: False,
+ "_nm3u8dl_b64url_encode": lambda raw: base64.urlsafe_b64encode(raw).decode("ascii").rstrip("="),
+ "_nm3u8dl_b64url_decode": _decode,
+ "NM3U8DL_QUALITY_HTTP_TIMEOUT_SEC": 15,
+ "log_timeout_exception": lambda *a, **kw: None,
+ "log": lambda message: None,
+ }
+ exec(compile(ast.Module(body=selected, type_ignores=[]), "", "exec"), namespace)
+ stream_headers = {
+ "User-Agent": "Stream-only UA",
+ "Referer": "https://stream-only.test/",
+ "Cookie": "stream-only-cookie",
+ }
+ if through_selection:
+ resolved = namespace["resolve_nm3u8dl_source_keys"](
+ {"keys": [license_key], "stream_url": STREAM_URL},
+ stream_headers,
+ )
+ else:
+ resolved = namespace["resolve_nm3u8dl_license_url"](
+ license_key, STREAM_URL, stream_headers,
+ )
+ self.assertEqual(resolved, [f"{KID}:{KEY}"])
+ self.assertEqual(manifest_calls, [(STREAM_URL, stream_headers)])
+ self.assertEqual(len(captured_requests), 1)
+ request, timeout = captured_requests[0]
+ self.assertEqual(timeout, 20)
+ self.assertEqual(request.get_method(), "POST")
+ self.assertEqual(json.loads(request.data), {
+ "kids": [_encode(KID)],
+ "type": "temporary",
+ })
+ return request, {name.casefold(): value for name, value in request.header_items()}
+
+ def test_license_pipe_headers_apply_only_to_license_request(self):
+ request, headers = self._run_resolver(
+ LICENSE_URL
+ + "|User-Agent=plaYtv/7.1.5 (Linux; Android 13) ExoPlayerLib/2.11.6"
+ + "&Referer=https://www.jiotv.co/&Origin=https://www.jiotv.co/"
+ )
+ self.assertEqual(request.full_url, LICENSE_URL)
+ self.assertEqual(headers["user-agent"], "plaYtv/7.1.5 (Linux; Android 13) ExoPlayerLib/2.11.6")
+ self.assertEqual(headers["referer"], "https://www.jiotv.co/")
+ self.assertEqual(headers["origin"], "https://www.jiotv.co/")
+ self.assertNotIn("cookie", headers)
+ self.assertEqual(headers["content-type"], "application/json")
+ self.assertEqual(headers["accept"], "application/json")
+
+ def test_plain_license_url_keeps_existing_request_defaults(self):
+ request, headers = self._run_resolver(LICENSE_URL, through_selection=False)
+ self.assertEqual(request.full_url, LICENSE_URL)
+ self.assertEqual(headers["user-agent"], "curl/8.21.0")
+ self.assertNotIn("referer", headers)
+ self.assertNotIn("cookie", headers)
+
+
+if __name__ == "__main__":
+ unittest.main()
diff --git a/tests/test_source_core.py b/tests/test_source_core.py
index 6222003..3a5c8dc 100644
--- a/tests/test_source_core.py
+++ b/tests/test_source_core.py
@@ -470,6 +470,57 @@ def test_clearkey_jwk_normalization_matches_mature_internal_format(self):
("ba8896d605246871ac424878491d86a1:8600d4153034b3cbc852f13ea4b7482c",),
)
+ def test_clearkey_comma_separated_pairs_keep_every_key(self):
+ pairs = tuple(f"{index:032x}:{index + 30:032x}" for index in range(1, 11))
+ self.assertEqual(
+ normalize_playlist_license_key(",".join(pairs)),
+ pairs,
+ )
+ self.assertEqual(
+ normalize_playlist_license_key(" " + pairs[0] + ", " + pairs[1] + " "),
+ pairs[:2],
+ )
+ self.assertEqual(
+ normalize_playlist_license_key(pairs[0]),
+ (pairs[0],),
+ )
+ self.assertEqual(
+ normalize_playlist_license_key(
+ "https://license.test/get?ids=one,two|Referer=https://site.test/"
+ ),
+ ("https://license.test/get?ids=one,two|Referer=https://site.test/",),
+ )
+ with self.assertRaisesRegex(RuntimeError, "Invalid comma-separated ClearKey"):
+ normalize_playlist_license_key(pairs[0] + ",not-a-key")
+
+ def test_sonuxs_json_keys_are_preserved_when_legacy_fields_null(self):
+ pairs = [f"{index:032x}:{index + 30:032x}" for index in range(1, 6)]
+ source = {
+ "channels": [{
+ "id": "E1HD",
+ "name": "Star Sports 1 HD",
+ "stream_url": "https://cdn.test/star.mpd",
+ "key_id": None,
+ "key": None,
+ "sonuxs": pairs,
+ "cookie": "__hdnea__=test",
+ }],
+ }
+ candidate = parse_playlist_text(json.dumps(source)).candidates[0]
+ self.assertEqual(candidate.keys, tuple(pairs))
+ self.assertEqual(candidate.license_type, "clearkey")
+ self.assertEqual(candidate.headers["Cookie"], "__hdnea__=test")
+
+ def test_sonuxs_does_not_replace_existing_explicit_license_key(self):
+ source = {"channels": [{
+ "name": "A",
+ "stream_url": "https://cdn.test/a.mpd",
+ "license_key": "abc:def",
+ "sonuxs": ["11111111111111111111111111111111:22222222222222222222222222222222"],
+ }]}
+ candidate = parse_playlist_text(json.dumps(source)).candidates[0]
+ self.assertEqual(candidate.keys, ("abc:def",))
+
def test_effective_probe_headers_use_mature_provider_defaults(self):
headers = build_effective_probe_headers(
"SONYLIV",