Ap0dexMe0/NF-MSL
1
0
Fork 0
mirror of https://github.com/Ap0dexMe0/NF-MSL.git synced 2026-10-24 16:52:10 +02:00
Code Issues Packages Projects Releases Wiki Activity

refactored

Browse Source
This commit is contained in:
Pari Malam
2026-05-29 20:53:59 +08:00
parent 3468c8f5b4
commit 001c33b8d8
8 changed files with 2244 additions and 2474 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.gitignore
+2 -1
View File
@@ -4,4 +4,5 @@ cookies/*
venv/*
output/*
.vscode/*
modules/__pycache__/
modules/__pycache__/
__pycache__/
main.py
+320 -1191
View File
File diff suppressed because it is too large Load Diff
modules/helpers.py
+512
View File
@@ -0,0 +1,512 @@
"""
helpers.py — Shared utility functions for the NF-MSL project.
Every function here was extracted from main.py to eliminate code duplication.
No behaviour was changed; all original logic is preserved.
"""
from __future__ import annotations
import base64
import json
import logging
import random
import re
import uuid
from http.cookies import SimpleCookie
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import requests
from Crypto.Cipher import AES
from Crypto.Util.Padding import unpad
# ---------------------------------------------------------------------------
# Directory / file helpers
# ---------------------------------------------------------------------------
def ensure_output_dir(platform_name, base_dir: Optional[Path] = None) -> Path:
"""Create (if needed) and return the ``output`` directory under *base_dir*.
If *base_dir* is ``None`` it defaults to the directory that contains the
calling script (i.e. ``Path(__file__).resolve().parent``).
"""
if base_dir is None:
base_dir = Path(__file__).resolve().parent
output_dir = base_dir / "output" / platform_name
output_dir.mkdir(parents=True, exist_ok=True)
return output_dir
def save_json(path: Path, data: Any, log: Optional[logging.Logger] = None) -> None:
"""Write *data* as pretty-printed JSON to *path*."""
try:
path.write_text(json.dumps(data, indent=2, ensure_ascii=False), encoding="utf-8")
if log:
log.info("Saved %s", path.name)
except Exception:
if log:
log.exception("Failed to save %s", path)
raise
# ---------------------------------------------------------------------------
# Session / cookie helpers
# ---------------------------------------------------------------------------
def restore_auth_cookies(
session: requests.Session,
cookies_path: Path,
log: logging.Logger,
) -> None:
"""Load previously saved cookies from *cookies_path* into *session*."""
if not cookies_path.exists():
return
try:
cached = json.loads(cookies_path.read_text(encoding="utf-8"))
if isinstance(cached, dict):
session.cookies.update(cached)
except Exception as exc:
log.warning("Saved auth cookies could not be restored: %s", exc)
def get_cookie_value(session: requests.Session, name: str) -> str:
"""Return the value of the first cookie named *name*, or empty string."""
for cookie in session.cookies:
if cookie.name == name:
return cookie.value
return ""
def get_nfvdid(session: requests.Session, response: Optional[requests.Response] = None) -> str:
"""Extract the ``nfvdid`` cookie from *session* (or *response* as fallback).
Raises ``RuntimeError`` when the cookie cannot be found.
"""
nfvdid = ""
for cookie in session.cookies:
if cookie.name == "nfvdid":
nfvdid = cookie.value
break
if not nfvdid and response is not None:
nfvdid = response.cookies.get("nfvdid", "")
if not nfvdid:
raise RuntimeError("The initial nfvdid cookie was not returned")
return nfvdid
def get_flow_session_cookies(session: requests.Session) -> Tuple[str, str]:
"""Return ``(nfvdid, flowSessionId)`` from the session cookie jar."""
nfvdid = ""
flow_session_id = ""
for cookie in session.cookies:
if cookie.name == "nfvdid":
nfvdid = cookie.value
elif cookie.name == "flwssn":
flow_session_id = cookie.value
return nfvdid, flow_session_id
def save_session_cookies(
session: requests.Session,
path: Path,
log: logging.Logger,
) -> Dict[str, str]:
"""Save all session cookies to *path* as JSON and return the dict."""
auth_cookies: Dict[str, str] = {}
for cookie in session.cookies:
auth_cookies[cookie.name] = cookie.value
try:
path.write_text(json.dumps(auth_cookies, indent=2), encoding="utf-8")
log.info("Authentication cookies saved to: %s", path)
except Exception:
log.exception("Failed to save cookies")
raise
return auth_cookies
def build_cookie_header(
session: requests.Session,
important_names: Tuple[str, ...],
) -> str:
"""Build a ``Cookie`` header string from *important_names* cookies in *session*.
Only the first occurrence of each name is included.
"""
seen: Dict[str, str] = {}
for cookie in session.cookies:
if cookie.name in important_names and cookie.value and cookie.name not in seen:
seen[cookie.name] = cookie.value
return "; ".join(f"{n}={seen[n]}" for n in important_names if n in seen)
def apply_set_cookie_headers(
session: requests.Session,
response: requests.Response,
cookie_values: Optional[List[str]] = None,
) -> None:
"""Parse ``Set-Cookie`` headers from *response* and apply them to *session*.
*cookie_values* is an optional list that will also be processed (useful
when the caller already extracted raw ``Set-Cookie`` strings from
``response.raw.headers``).
"""
values: List[str] = list(cookie_values) if cookie_values else []
raw_headers = getattr(response.raw, "headers", None)
if raw_headers is not None and hasattr(raw_headers, "get_all"):
values.extend(raw_headers.get_all("Set-Cookie") or [])
header_value = response.headers.get("Set-Cookie")
if header_value and header_value not in values:
values.append(header_value)
for raw_cookie in values:
jar = SimpleCookie()
try:
jar.load(raw_cookie)
except Exception:
continue
for morsel in jar.values():
cookie_domain = morsel["domain"] or None
cookie_path = morsel["path"] or "/"
if morsel.value == "":
try:
session.cookies.clear(domain=cookie_domain, path=cookie_path, name=morsel.key)
except Exception:
pass
continue
try:
session.cookies.clear(domain=cookie_domain, path=cookie_path, name=morsel.key)
except Exception:
pass
session.cookies.set(
morsel.key, morsel.value,
domain=cookie_domain, path=cookie_path,
secure=bool(morsel["secure"]),
)
def dedupe_important_cookies(
session: requests.Session,
important_names: Tuple[str, ...],
) -> None:
"""Remove duplicate cookies so only the best-scoring one per name remains.
Scoring prefers ``.netflix.com`` domain, ``/`` path, and non-empty value.
"""
preferred: Dict[str, Any] = {}
for cookie in session.cookies:
if cookie.name not in important_names:
continue
current = preferred.get(cookie.name)
score = (cookie.domain == ".netflix.com", cookie.path == "/", bool(cookie.value))
if current is None or score >= current[0]:
preferred[cookie.name] = (score, cookie)
for cookie in list(session.cookies):
winner = preferred.get(cookie.name)
if not winner:
continue
winner_cookie = winner[1]
if (cookie.domain, cookie.path, cookie.value) != (
winner_cookie.domain, winner_cookie.path, winner_cookie.value,
):
try:
session.cookies.clear(domain=cookie.domain, path=cookie.path, name=cookie.name)
except Exception:
pass
def collect_important_cookies(
session: requests.Session,
important_names: Tuple[str, ...],
) -> Dict[str, str]:
"""Return a flat dict of important cookie name → value (first seen wins)."""
cookies: Dict[str, str] = {}
for cookie in session.cookies:
if cookie.name in important_names and cookie.value and cookie.name not in cookies:
cookies[cookie.name] = cookie.value
return cookies
# ---------------------------------------------------------------------------
# ID / random-string generators
# ---------------------------------------------------------------------------
def generate_hex_id(length: int = 32, uppercase: bool = False) -> str:
"""Return a random hex string of the given *length*."""
charset = "0123456789ABCDEF" if uppercase else "0123456789abcdef"
return "".join(random.choice(charset) for _ in range(length))
def generate_netflix_uuid() -> str:
"""Return a UUID in the Netflix ``8-4-4-4-12`` hex format."""
return (
generate_hex_id(8) + "-"
+ generate_hex_id(4) + "-"
+ generate_hex_id(4) + "-"
+ generate_hex_id(4) + "-"
+ generate_hex_id(12)
)
def generate_request_id() -> str:
"""Shorthand for a 32-char lowercase hex request ID."""
return generate_hex_id(32)
def generate_esn_random_suffix(length: int = 64) -> str:
"""Return a random alphanumeric suffix for ESN generation."""
return "".join(random.choice("ABCDEFGHIJKLMNOPQRSTUVWXYZ0123456789") for _ in range(length))
# ---------------------------------------------------------------------------
# MSL crypto helpers
# ---------------------------------------------------------------------------
def _resolve_key(raw_key: Any) -> bytes:
"""Convert a key value (hex-str, bytes, or raw) to ``bytes``."""
if isinstance(raw_key, str):
try:
return bytes.fromhex(raw_key)
except ValueError:
return raw_key.encode("utf-8")
return raw_key
def decrypt_msl_header(
headerdata_b64: str,
encryption_key_value: Any,
sign_key_value: Any,
) -> Dict[str, Any]:
"""Decrypt an MSL encrypted header and return the parsed JSON dict.
*encryption_key_value* and *sign_key_value* may be hex-strings or bytes.
Raises ``RuntimeError`` when either key is missing.
"""
encryption_key = _resolve_key(encryption_key_value)
sign_key = _resolve_key(sign_key_value)
if not encryption_key:
raise RuntimeError("The encryption key is missing")
if not sign_key:
raise RuntimeError("The sign key is missing")
encrypted_header = json.loads(base64.b64decode(headerdata_b64))
iv = base64.b64decode(encrypted_header["iv"])
ciphertext = base64.b64decode(encrypted_header["ciphertext"])
cipher = AES.new(encryption_key, AES.MODE_CBC, iv)
decrypted = unpad(cipher.decrypt(ciphertext), AES.block_size)
return json.loads(decrypted.decode("utf-8"))
# ---------------------------------------------------------------------------
# HTML / response parsing helpers
# ---------------------------------------------------------------------------
def extract_first_match(html: str, patterns: List[str]) -> Optional[str]:
"""Try each regex *pattern* against *html* and return the first captured group."""
for pattern in patterns:
match = re.search(pattern, html)
if match:
return match.group(1)
return None
CLCS_SESSION_ID_PATTERNS = [
r'"clcsSessionId"\s*:\s*"([0-9a-f\-]{36})"',
r'\\"clcsSessionId\\"\s*:\s*\\"([0-9a-f\-]{36})\\"',
r'"serverState"\s*:\s*"[^\"]*clcsSessionId\\":\\"([0-9a-f\-]{36})',
r'"trackingInfo"\s*:\s*"[^\"]*clcsSessionId\\":\\"([0-9a-f\-]{36})',
r'"sessionId"\s*:\s*"([0-9a-f\-]{36})"',
]
RENDITION_ID_PATTERNS = [
r'"renditionId"\s*:\s*"([0-9a-f\-]{36})"',
r'\\"renditionId\\"\s*:\s*\\"([0-9a-f\-]{36})\\"',
]
def extract_clcs_session_id(html: str) -> str:
"""Extract ``clcsSessionId`` from login page HTML.
Raises ``RuntimeError`` when not found.
"""
result = extract_first_match(html, CLCS_SESSION_ID_PATTERNS)
if not result:
raise RuntimeError("Could not extract clcsSessionId from the login page HTML")
return result
def extract_rendition_id(html: str) -> str:
"""Extract ``renditionId`` from login page HTML.
Raises ``RuntimeError`` when not found.
"""
result = extract_first_match(html, RENDITION_ID_PATTERNS)
if not result:
raise RuntimeError("Could not extract the initial renditionId from the login page HTML")
return result
def parse_flow_data(response_data: Dict[str, Any]) -> Dict[str, str]:
"""Walk a GraphQL response tree and extract flow metadata.
Returns a dict that may contain keys:
``clcsSessionId``, ``renditionId``, ``flowSessionId``, ``mode``,
``flow``, ``membershipStatus``.
"""
flow: Dict[str, str] = {}
data = response_data.get("data", {})
operation_key = next(iter(data.keys()), "")
inner = data.get(operation_key, {})
screen = inner.get("screen", inner) if isinstance(inner, dict) else {}
stack = [screen]
while stack:
value = stack.pop()
if isinstance(value, dict):
tracking_info = value.get("trackingInfo")
if isinstance(tracking_info, str) and tracking_info:
try:
tracking = json.loads(tracking_info)
except Exception:
tracking = {}
if tracking.get("clcsSessionId") and not flow.get("clcsSessionId"):
flow["clcsSessionId"] = tracking.get("clcsSessionId", "")
if tracking.get("clcsRenditionId") and not flow.get("renditionId"):
flow["renditionId"] = tracking.get("clcsRenditionId", "")
payload_json = value.get("payloadJson")
if isinstance(payload_json, str) and payload_json:
try:
payload = json.loads(payload_json)
except Exception:
payload = {}
if payload.get("flwssn") and not flow.get("flowSessionId"):
flow["flowSessionId"] = payload.get("flwssn", "")
if payload.get("mode") and not flow.get("mode"):
flow["mode"] = payload.get("mode", "")
if payload.get("flow") and not flow.get("flow"):
flow["flow"] = payload.get("flow", "")
if value.get("membershipStatus"):
flow["membershipStatus"] = value.get("membershipStatus", "")
for child in value.values():
stack.append(child)
elif isinstance(value, list):
for item in value:
stack.append(item)
return flow
# ---------------------------------------------------------------------------
# MSL payload parsing helpers (used by run_tv / run_tv_otp)
# ---------------------------------------------------------------------------
def parse_msl_payload(payload: Any) -> Tuple[str, Optional[Dict], Optional[str]]:
"""Classify an MSL payload and return ``(type_str, parsed_dict, text_str)``.
*type_str* is one of ``"msl_payload"``, ``"json_array"``, or ``"text"``.
"""
payload_type = type(payload).__name__
parsed_payload: Optional[Dict] = None
text_payload: Optional[str] = None
if isinstance(payload, dict):
payload_type = "msl_payload"
parsed_payload = payload
elif isinstance(payload, list):
payload_type = "json_array"
parsed_payload = {"items": payload}
elif isinstance(payload, str):
cleaned = payload.rstrip(
"\x00\x01\x02\x03\x04\x05\x06\x07\x08\x09\x0a\x0b\x0c\x0d\x0e\x0f\x10"
)
text_payload = cleaned
try:
maybe_json = json.loads(cleaned)
if isinstance(maybe_json, dict):
parsed_payload = maybe_json
payload_type = "text"
else:
payload_type = "text"
except Exception:
payload_type = "text"
return payload_type, parsed_payload, text_payload
def extract_useridtoken_from_payload(
parsed_payload: Optional[Dict], raw_payload: Any,
) -> Optional[Dict[str, Any]]:
"""Walk the payload tree and return the first dict with both
``tokendata`` and ``signature`` keys (the useridtoken)."""
useridtoken: Optional[Dict[str, Any]] = None
stack = [parsed_payload if parsed_payload is not None else raw_payload]
while stack:
current = stack.pop()
if isinstance(current, dict):
if useridtoken is None and set(current.keys()) >= {"tokendata", "signature"}:
useridtoken = current
for child in current.values():
stack.append(child)
elif isinstance(current, list):
for item in current:
stack.append(item)
return useridtoken
def build_msl_trace_event(
msl_message_id: Any,
key_id: str,
payload_type: str,
parsed_payload: Optional[Dict],
text_payload: Optional[str],
extra_fields: Optional[Dict[str, Any]] = None,
) -> Dict[str, Any]:
"""Build a single MSL trace event dict (used for debug logging)."""
from datetime import datetime, timezone
event: Dict[str, Any] = {
"_type": "decrypt",
"_mslId": msl_message_id,
"_timestamp": datetime.now(timezone.utc)
.isoformat(timespec="milliseconds")
.replace("+00:00", "Z"),
"_keyId": key_id,
"_dataType": payload_type,
}
if extra_fields:
event.update(extra_fields)
if text_payload is not None:
event["_text"] = text_payload
if parsed_payload is not None:
event["_payload"] = parsed_payload
return event
def extract_key_id_from_mastertoken(mastertoken: Dict[str, Any]) -> str:
"""Extract the hex key ID from a master token's ``tokendata``."""
if not mastertoken:
return ""
token_data = json.loads(base64.b64decode(mastertoken["tokendata"]).decode("utf-8"))
return str(token_data.get("sequencenumber", "")).encode("utf-8").hex()
# ---------------------------------------------------------------------------
# Request-args helpers (TV platform)
# ---------------------------------------------------------------------------
def request_args_to_dict(request_args: List[Dict[str, Any]]) -> Dict[str, Any]:
"""Convert a list of ``{"name":…, "value": {"stringValue"|"booleanValue":…}}``
dicts into a plain ``{name: value}`` mapping."""
result: Dict[str, Any] = {}
for arg in request_args:
value = arg["value"]
if "stringValue" in value:
result[arg["name"]] = value["stringValue"]
elif "booleanValue" in value:
result[arg["name"]] = value["booleanValue"]
return result
modules/msl_android.py
+263 -360
View File
@@ -1,28 +1,29 @@
from __future__ import annotations
import base64
import gzip
import json
import random
import sys
import zlib
from pathlib import Path
from typing import Any, Dict, List, Optional, Tuple
import jsonpickle
import requests
from io import BytesIO
from pathlib import Path
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from Cryptodome.Cipher import AES
from Cryptodome.Hash import HMAC, SHA256
from Cryptodome.Cipher import PKCS1_OAEP, AES as _AES
from Cryptodome.Hash import SHA1
from Cryptodome.PublicKey import RSA
from Cryptodome.PublicKey.RSA import RsaKey
from Cryptodome.Random import get_random_bytes
from Cryptodome.Util import Padding
from pywidevine import Cdm as WidevineCdm, Device as WidevineDevice, PSSH
class MSLObject:
def __repr__(self) -> str:
return f"<{self.__class__.__name__} {jsonpickle.encode(self, unpicklable=False)}>"
from modules.msl_base import MSLBase, MSLKeys as _BaseMSLKeys, get_widevine_key
class MSLKeys(MSLObject):
# ---------------------------------------------------------------------------
# Platform-specific key container
# ---------------------------------------------------------------------------
class MSLKeys(_BaseMSLKeys):
"""Android MSL keys – extends the base with an RSA key and CDM session."""
def __init__(
self,
encryption: Optional[bytes] = None,
@@ -30,26 +31,40 @@ class MSLKeys(MSLObject):
rsa: Optional[RsaKey] = None,
mastertoken: Optional[dict] = None,
cdm_session: Any = None,
):
self.encryption = encryption
self.sign = sign
) -> None:
super().__init__(encryption=encryption, sign=sign, mastertoken=mastertoken)
self.rsa = rsa
self.mastertoken = mastertoken
self.cdm_session = cdm_session
class MSL_ANDROID:
DEFAULT_HANDSHAKE_ENDPOINT = "https://android.prod.ftl.netflix.com/nq/androidui/pbo_license/~1.0.0/router"
DEFAULT_MANIFEST_ENDPOINT = "https://android.prod.ftl.netflix.com/msl/playapi/android/manifest"
DEFAULT_MANIFEST_PARAMS = {
# ---------------------------------------------------------------------------
# MSL_ANDROID
# ---------------------------------------------------------------------------
class MSL_ANDROID(MSLBase):
"""Netflix MSL client for Android devices using Widevine DRM."""
# -- Platform constants --------------------------------------------------
DEFAULT_HANDSHAKE_ENDPOINT: str = (
"https://android.prod.ftl.netflix.com/nq/androidui/pbo_license/~1.0.0/router"
)
DEFAULT_MANIFEST_ENDPOINT: str = (
"https://android.prod.ftl.netflix.com/msl/playapi/android/manifest"
)
DEFAULT_MANIFEST_PARAMS: Dict[str, str] = {
"ab_ui_ver": "android",
"nrdapp_version": "18.26.0",
}
DEFAULT_USER_AGENT = "com.netflix.mediaclient/63988 (Linux; U; Android 15; en_US; SM-F711N; Build/AP3A.240905.015.A2; Cronet/143.0.7445.0)"
DEFAULT_REQUEST_CONTEXT = '{"appState":"foreground","appView":"unknown"}'
DEFAULT_NRDJS_VERSION = "v3.12.55"
DEFAULT_NETJS_VERSION = "3.0.5"
DEFAULT_PBO_VERSION = 2
DEFAULT_PBO_COMMON = {
DEFAULT_USER_AGENT: str = (
"com.netflix.mediaclient/63988 "
"(Linux; U; Android 15; en_US; SM-F711N; "
"Build/AP3A.240905.015.A2; Cronet/143.0.7445.0)"
)
DEFAULT_REQUEST_CONTEXT: str = '{"appState":"foreground","appView":"unknown"}'
DEFAULT_NRDJS_VERSION: str = "v3.12.55"
DEFAULT_NETJS_VERSION: str = "3.0.5"
DEFAULT_PBO_VERSION: int = 2
DEFAULT_PBO_COMMON: Dict[str, str] = {
"sdk": "18.26.0",
"platform": "18.26.0",
"application": "Netflix Android 18.26.0",
@@ -58,8 +73,10 @@ class MSL_ANDROID:
"clientVersion": "18.26.0",
"apkVersion": "18.26.0",
}
DEFAULT_PBO_LANGUAGES = ["en-US", "en"]
DEFAULT_DEVICE_MODEL = "SM-F711N"
DEFAULT_PBO_LANGUAGES: List[str] = ["en-US", "en"]
DEFAULT_DEVICE_MODEL: str = "SM-F711N"
# -- Constructor ---------------------------------------------------------
def __init__(
self,
@@ -69,14 +86,13 @@ class MSL_ANDROID:
sender: str,
user_auth: Optional[dict] = None,
drm: str = "widevine",
):
self.session = session
self.keys = keys
self.sender = sender
) -> None:
super().__init__(session=session, keys=keys, message_id=message_id, sender=sender)
self.user_auth = user_auth
self.message_id = message_id
self.drm = drm
# -- Widevine handshake --------------------------------------------------
@classmethod
def handshake(
cls,
@@ -91,11 +107,12 @@ class MSL_ANDROID:
endpoint: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
) -> MSLKeys:
"""Perform a Widevine key exchange and return negotiated keys."""
if cookies:
session.cookies.update(cookies)
cache_path = Path(msl_keys_path)
msl_keys = MSL_ANDROID.load_cache_data(cache_path)
msl_keys = cls.load_cache_data(cache_path)
if msl_keys is not None and not new_msl:
return msl_keys
@@ -103,7 +120,7 @@ class MSL_ANDROID:
raise ValueError(f"Unsupported DRM mode: {drm}")
if not cdm:
raise ValueError("Widevine CDM is required for this iOS MSL flow")
raise ValueError("Widevine CDM is required for this Android MSL flow")
message_id = random.randint(0, pow(2, 52))
msl_keys = MSLKeys()
@@ -135,7 +152,7 @@ class MSL_ANDROID:
},
},
"headerdata": base64.standard_b64encode(
MSL_ANDROID.generate_msg_header(
cls.generate_msg_header(
message_id=message_id,
sender=sender,
is_handshake=True,
@@ -169,10 +186,14 @@ class MSL_ANDROID:
host="android15.prod.cloud.netflix.com",
language="en-US,en",
)
res = session.post(url=handshake_endpoint, data=data, headers=handshake_headers, timeout=30)
res = session.post(
url=handshake_endpoint, data=data, headers=handshake_headers, timeout=30
)
if res.status_code != 200:
raise RuntimeError(f"Key exchange failed: HTTP {res.status_code} {res.text[:500]}")
raise RuntimeError(
f"Key exchange failed: HTTP {res.status_code} {res.text[:500]}"
)
parsed = cls.parse_concatenated_json(res.text)
if not parsed:
@@ -181,12 +202,17 @@ class MSL_ANDROID:
key_exchange = parsed[0]
if "errordata" in key_exchange:
decoded_error = base64.standard_b64decode(key_exchange["errordata"]).decode("utf-8")
decoded_error = base64.standard_b64decode(
key_exchange["errordata"]
).decode("utf-8")
error_json = json.loads(decoded_error)
raise RuntimeError(f"Key exchange failed: {error_json}")
if "headerdata" not in key_exchange:
raise RuntimeError(f"Key exchange failed: missing headerdata in response: {str(key_exchange)[:500]}")
raise RuntimeError(
f"Key exchange failed: missing headerdata in response: "
f"{str(key_exchange)[:500]}"
)
header_json = json.loads(
base64.standard_b64decode(key_exchange["headerdata"]).decode("utf-8")
@@ -195,22 +221,190 @@ class MSL_ANDROID:
key_data = key_response_data["keydata"]
cdm.parse_license(msl_keys.cdm_session, key_data["cdmkeyresponse"])
keys = cdm.get_keys(msl_keys.cdm_session)
msl_keys.encryption = MSL_ANDROID.get_widevine_key(
wv_keys = cdm.get_keys(msl_keys.cdm_session)
msl_keys.encryption = get_widevine_key(
kid=base64.standard_b64decode(key_data["encryptionkeyid"]),
keys=keys,
keys=wv_keys,
permissions=["AllowEncrypt", "AllowDecrypt"],
)
msl_keys.sign = MSL_ANDROID.get_widevine_key(
msl_keys.sign = get_widevine_key(
kid=base64.standard_b64decode(key_data["hmackeyid"]),
keys=keys,
keys=wv_keys,
permissions=["AllowSign", "AllowSignatureVerify"],
)
msl_keys.mastertoken = key_response_data["mastertoken"]
MSL_ANDROID.cache_keys(msl_keys, cache_path)
cls.cache_keys(msl_keys, cache_path)
return msl_keys
# -- RSA / ASYMMETRIC_WRAPPED handshake (no Widevine required) -----------
@classmethod
def rsa_handshake(
cls,
msl_keys_path: str,
session: requests.Session,
sender: str,
new_msl: bool,
cookies: Optional[Dict[str, str]],
endpoint: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
) -> "MSLKeys":
"""Perform an ASYMMETRIC_WRAPPED (RSA + JWE) key exchange.
This path does **not** require a Widevine device file. Netflix wraps
the session AES-GCM keys with our ephemeral RSA-OAEP public key and
returns them inside a JWE compact serialisation.
The ESN must use the ``NFCDCH-02-`` prefix (web-style) so that the
Android FTL endpoint accepts the ``NONE`` entity auth scheme.
"""
if cookies:
session.cookies.update(cookies)
cache_path = Path(msl_keys_path)
cached = cls.load_cache_data(cache_path)
if cached is not None and not new_msl:
return cached
# ---- Generate ephemeral RSA-2048 keypair ----------------------------
rsa_key = RSA.generate(2048)
pub_der_b64 = base64.b64encode(
rsa_key.publickey().export_key("DER")
).decode("ascii")
message_id = random.randint(0, 2 ** 52)
msl_keys = MSLKeys(rsa=rsa_key)
keyrequestdata = {
"scheme": "ASYMMETRIC_WRAPPED",
"keydata": {
"keypairid": "rsaKeypairId",
"mechanism": "JWE_RSA",
"publickey": pub_der_b64,
},
}
data = jsonpickle.encode(
{
"entityauthdata": {
"scheme": "NONE",
"authdata": {"identity": sender},
},
"headerdata": base64.standard_b64encode(
cls.generate_msg_header(
message_id=message_id,
sender=sender,
is_handshake=True,
keyrequestdata=keyrequestdata,
).encode("utf-8")
).decode("utf-8"),
"signature": "",
},
unpicklable=False,
)
data += json.dumps(
{
"payload": base64.standard_b64encode(
json.dumps(
{
"messageid": message_id,
"data": "",
"sequencenumber": 1,
"endofmsg": True,
}
).encode("utf-8")
).decode("utf-8"),
"signature": "",
}
)
handshake_endpoint = endpoint or cls.DEFAULT_HANDSHAKE_ENDPOINT
handshake_headers = headers or cls.build_request_headers(
request_name="mintCookies",
esn=sender,
host="android15.prod.cloud.netflix.com",
language="en-US,en",
)
res = session.post(
url=handshake_endpoint,
data=data,
headers=handshake_headers,
timeout=30,
)
if res.status_code != 200:
raise RuntimeError(
f"RSA key exchange failed: HTTP {res.status_code} {res.text[:500]}"
)
parsed = cls.parse_concatenated_json(res.text)
if not parsed:
raise RuntimeError("RSA key exchange failed: empty MSL response")
key_exchange = parsed[0]
if "errordata" in key_exchange:
decoded_error = base64.standard_b64decode(
key_exchange["errordata"]
).decode("utf-8")
raise RuntimeError(
f"RSA key exchange failed: {json.loads(decoded_error)}"
)
if "headerdata" not in key_exchange:
raise RuntimeError(
f"RSA key exchange failed: missing headerdata in response: "
f"{str(key_exchange)[:500]}"
)
header_json = json.loads(
base64.standard_b64decode(key_exchange["headerdata"]).decode("utf-8")
)
key_response_data = header_json["keyresponsedata"]
key_data = key_response_data["keydata"]
# ---- Decrypt JWE-wrapped session keys -------------------------------
msl_keys.encryption = cls._decrypt_jwe_key(
key_data["encryptionkey"], rsa_key
)
msl_keys.sign = cls._decrypt_jwe_key(key_data["hmackey"], rsa_key)
msl_keys.mastertoken = key_response_data["mastertoken"]
# Don't persist the RSA key object (not picklable); clear it before caching
msl_keys.rsa = None
cls.cache_keys(msl_keys, cache_path)
return msl_keys
@staticmethod
def _decrypt_jwe_key(field_b64: str, rsa_private: RsaKey) -> bytes:
"""Decrypt a Netflix JWE-wrapped session key.
The field value is standard-base64-encoded JWE compact serialisation.
Netflix uses RSA-OAEP (SHA-1) for the CEK and A128GCM for content
but skips the AAD in the GCM tag — so we decrypt without verification
and trust the RSA layer for integrity.
"""
jwe_compact = base64.b64decode(field_b64 + "==").decode("utf-8")
hdr_b64, enc_cek_b64, iv_b64, ct_b64, _ = jwe_compact.split(".")
# Decrypt the Content Encryption Key (CEK) with RSA-OAEP / SHA-1
enc_cek = base64.urlsafe_b64decode(enc_cek_b64 + "==")
cek = PKCS1_OAEP.new(rsa_private, hashAlgo=SHA1).decrypt(enc_cek)
# Decrypt the payload with AES-128-GCM (no AAD verification)
iv = base64.urlsafe_b64decode(iv_b64 + "==")
ct = base64.urlsafe_b64decode(ct_b64 + "==")
plaintext = _AES.new(cek, _AES.MODE_GCM, nonce=iv).decrypt(ct)
# Payload is a JWK JSON — extract the raw key bytes
jwk = json.loads(plaintext.decode("utf-8"))
remainder = len(jwk["k"]) % 4
padded = jwk["k"] + "=" * (4 - remainder if remainder else 0)
return base64.urlsafe_b64decode(padded)
# -- Platform-specific request headers -----------------------------------
@staticmethod
def build_request_headers(
request_name: str,
@@ -225,6 +419,7 @@ class MSL_ANDROID:
language: Optional[str] = "en-US",
device_model: Optional[str] = None,
) -> Dict[str, str]:
"""Build the HTTP headers required for Android MSL requests."""
headers: Dict[str, str] = {
"Host": host or "android15.prod.cloud.netflix.com",
"Accept": "*/*",
@@ -237,7 +432,9 @@ class MSL_ANDROID:
"X-DeviceModel": device_model or MSL_ANDROID.DEFAULT_DEVICE_MODEL,
"x-netflix.client.request.name": request_name,
"x-netflix.request.attempt": "1",
"x-netflix.request.id": "".join(random.choice("0123456789abcdef") for _ in range(32)),
"x-netflix.request.id": "".join(
random.choice("0123456789abcdef") for _ in range(32)
),
"x-netflix.request.client.context": MSL_ANDROID.DEFAULT_REQUEST_CONTEXT,
"x-netflix.request.client.languages": "en-US",
"x-netflix.request.client.timezoneid": "America/New_York",
@@ -273,50 +470,14 @@ class MSL_ANDROID:
headers.update(extra_headers)
return headers
# -- Manifest defaults ---------------------------------------------------
@staticmethod
def manifest_request_defaults() -> Tuple[str, Dict[str, str]]:
"""Return the default manifest endpoint and query params for Android."""
return MSL_ANDROID.DEFAULT_MANIFEST_ENDPOINT, dict(MSL_ANDROID.DEFAULT_MANIFEST_PARAMS)
@staticmethod
def generate_msg_header(
message_id: int,
sender: str,
is_handshake: bool,
userauthdata: Optional[dict] = None,
keyrequestdata: Optional[dict] = None,
compression: Optional[str] = "GZIP",
) -> str:
header_data: Dict[str, Any] = {
"messageid": message_id,
"renewable": True,
"handshake": is_handshake,
"capabilities": {
"compressionalgos": [compression] if compression else [],
"languages": ["en-US", "en"],
"encoderformats": ["JSON"],
},
"timestamp": int(datetime.now(timezone.utc).timestamp()),
"sender": sender,
"nonreplayable": False,
"recipient": "Netflix",
}
if userauthdata:
header_data["userauthdata"] = userauthdata
if keyrequestdata:
header_data["keyrequestdata"] = [keyrequestdata]
return jsonpickle.encode(header_data, unpicklable=False)
@staticmethod
def get_widevine_key(kid: bytes, keys: List[Any], permissions: List[str]) -> Optional[bytes]:
import re
normalized_perms = {re.sub(r'(?<!^)(?=[A-Z])', '_', p).lower() for p in permissions}
for key in keys:
if key.type != "OPERATOR_SESSION":
continue
key_perms = {p.lower() for p in (getattr(key, "permissions", None) or [])}
if normalized_perms <= key_perms:
return key.key
return None
# -- Override send_message to add PBO normalisation ----------------------
def send_message(
self,
@@ -326,277 +487,19 @@ class MSL_ANDROID:
userauthdata: Optional[dict] = None,
headers: Optional[dict] = None,
proxy: Optional[Dict[str, str]] = None,
timeout: int = 30,
) -> Tuple[Dict[str, Any], Any]:
normalized_application_data = self.normalize_application_data(endpoint, application_data)
message = self.create_message(normalized_application_data, userauthdata)
request_kwargs: Dict[str, Any] = {
"url": endpoint,
"data": message,
"params": params,
"headers": headers,
"timeout": 30,
}
if proxy:
request_kwargs["proxies"] = proxy
res = self.session.post(**request_kwargs)
if res.status_code != 200:
raise RuntimeError(
f"MSL request failed with HTTP {res.status_code}: {res.text[:500]}"
)
response_text = res.text or ""
stripped_response = response_text.lstrip()
if not stripped_response:
raise RuntimeError("MSL request failed: empty response body")
if not stripped_response.startswith("{"):
content_type = res.headers.get("content-type", "")
raise RuntimeError(
"MSL request failed: the server did not return concatenated MSL JSON. "
f"Content-Type: {content_type!r}. Body preview: {response_text[:500]!r}"
)
header, payload_data = self.parse_message(response_text)
if not header:
raise RuntimeError(
f"MSL request failed: parsed response does not contain a header. Body preview: {response_text[:500]!r}"
)
if "errordata" in header:
decoded_error = json.loads(
base64.standard_b64decode(header["errordata"].encode("utf-8")).decode("utf-8")
)
raise RuntimeError(f"MSL response contains an error: {decoded_error}")
return header, payload_data
@classmethod
def normalize_application_data(cls, endpoint: str, application_data: Any) -> Any:
if not isinstance(application_data, dict):
return application_data
if cls._looks_like_wrapped_pbo_payload(application_data):
return application_data
route = cls._extract_pbo_route(application_data, endpoint)
if route is None:
return application_data
common = dict(cls.DEFAULT_PBO_COMMON)
if isinstance(application_data.get("common"), dict):
common.update(application_data["common"])
wrapped: Dict[str, Any] = {
"version": application_data.get("version", cls.DEFAULT_PBO_VERSION),
"common": common,
"url": route,
"languages": application_data.get("languages", list(cls.DEFAULT_PBO_LANGUAGES)),
"params": application_data.get("params", {}),
}
for key in ("path", "method", "route", "endpoint"):
wrapped.pop(key, None)
for key, value in application_data.items():
if key in wrapped or key in {"version", "common", "languages", "params", "path", "method", "route", "endpoint"}:
continue
wrapped[key] = value
return wrapped
@staticmethod
def _looks_like_wrapped_pbo_payload(application_data: Dict[str, Any]) -> bool:
return (
"version" in application_data
and "common" in application_data
and "url" in application_data
and "languages" in application_data
and "params" in application_data
"""Send an MSL message with PBO payload normalisation."""
normalized = self.normalize_application_data(endpoint, application_data)
return super().send_message(
endpoint=endpoint,
params=params,
application_data=normalized,
userauthdata=userauthdata,
headers=headers,
proxy=proxy,
timeout=timeout,
)
@staticmethod
def _extract_pbo_route(application_data: Dict[str, Any], endpoint: str) -> Optional[str]:
for key in ("url", "path", "route", "method", "endpoint"):
value = application_data.get(key)
if not isinstance(value, str) or not value.strip():
continue
value = value.strip()
return value if value.startswith("/") else f"/{value}"
endpoint_lower = endpoint.lower()
if "manifest" in endpoint_lower:
return "/manifest"
if "pbo_tokens" in endpoint_lower or "pbo_config" in endpoint_lower:
return None
return None
def create_message(self, application_data: Dict[str, Any], userauthdata: Optional[dict] = None) -> str:
self.message_id += 1
headerdata = self.encrypt(
self.generate_msg_header(
message_id=self.message_id,
sender=self.sender,
is_handshake=False,
userauthdata=userauthdata,
)
)
message = json.dumps(
{
"headerdata": base64.standard_b64encode(headerdata.encode("utf-8")).decode("utf-8"),
"signature": self.sign(headerdata).decode("utf-8"),
"mastertoken": self.keys.mastertoken,
},
separators=(",", ":"),
)
compressed_application_data = self.gzip_compress(
json.dumps(application_data, separators=(",", ":")).encode("utf-8")
).decode("utf-8")
payload_dicts = [
{
"sequencenumber": 1,
"messageid": self.message_id,
"compressionalgo": "GZIP",
"data": compressed_application_data,
},
{
"sequencenumber": 2,
"messageid": self.message_id,
"endofmsg": True,
"data": "",
},
]
for payload_dict in payload_dicts:
payload_chunk = self.encrypt(json.dumps(payload_dict, separators=(",", ":")))
message += json.dumps(
{
"payload": base64.standard_b64encode(payload_chunk.encode("utf-8")).decode("utf-8"),
"signature": self.sign(payload_chunk).decode("utf-8"),
},
separators=(",", ":"),
)
return message
def decrypt_payload_chunks(self, payload_chunks: List[Dict[str, str]]) -> Any:
raw_data = ""
assert self.keys.encryption is not None
for payload_chunk in payload_chunks:
payload_chunk_json = json.loads(base64.standard_b64decode(payload_chunk["payload"]).decode("utf-8"))
payload_decrypted = AES.new(
key=self.keys.encryption,
mode=AES.MODE_CBC,
iv=base64.standard_b64decode(payload_chunk_json["iv"]),
).decrypt(base64.standard_b64decode(payload_chunk_json["ciphertext"]))
payload_decrypted = Padding.unpad(payload_decrypted, 16)
payload_decrypted_json = json.loads(payload_decrypted.decode("utf-8"))
payload_data = base64.standard_b64decode(payload_decrypted_json["data"])
if payload_decrypted_json.get("compressionalgo") == "GZIP":
payload_data = zlib.decompress(payload_data, 16 + zlib.MAX_WBITS)
raw_data += payload_data.decode("utf-8")
if not raw_data:
return None
try:
data = json.loads(raw_data)
except Exception:
return raw_data
if "error" in data:
return None
if "result" not in data:
return data
return data["result"]
@staticmethod
def parse_concatenated_json(message: str) -> List[Dict[str, Any]]:
decoder = json.JSONDecoder()
items: List[Dict[str, Any]] = []
index = 0
length = len(message)
while index < length:
while index < length and message[index].isspace():
index += 1
if index >= length:
break
item, next_index = decoder.raw_decode(message, index)
items.append(item)
index = next_index
return items
def parse_message(self, message: str) -> Tuple[Dict[str, Any], Any]:
parsed_message = self.parse_concatenated_json(message)
header = parsed_message[0]
encrypted_payload_chunks = parsed_message[1:] if len(parsed_message) > 1 else []
payload_chunks = self.decrypt_payload_chunks(encrypted_payload_chunks) if encrypted_payload_chunks else {}
return header, payload_chunks
@staticmethod
def gzip_compress(data: bytes) -> bytes:
out = BytesIO()
with gzip.GzipFile(fileobj=out, mode="w") as gzip_file:
gzip_file.write(data)
return base64.standard_b64encode(out.getvalue())
@staticmethod
def base64key_decode(payload: str) -> bytes:
length = len(payload) % 4
if length == 2:
payload += "=="
elif length == 3:
payload += "="
elif length != 0:
raise ValueError("Invalid base64 string")
return base64.urlsafe_b64decode(payload.encode("utf-8"))
def encrypt(self, plaintext: str) -> str:
if not self.keys.encryption:
raise ValueError("Encryption key is not available")
if not self.keys.mastertoken:
raise ValueError("Master token is not available")
iv = get_random_bytes(16)
tokendata = json.loads(base64.standard_b64decode(self.keys.mastertoken["tokendata"]).decode("utf-8"))
return json.dumps(
{
"ciphertext": base64.standard_b64encode(
AES.new(self.keys.encryption, AES.MODE_CBC, iv).encrypt(
Padding.pad(plaintext.encode("utf-8"), 16)
)
).decode("utf-8"),
"keyid": f"{self.sender}_{tokendata['sequencenumber']}",
"sha256": "AA==",
"iv": base64.standard_b64encode(iv).decode("utf-8"),
}
)
def sign(self, text: str) -> bytes:
if not self.keys.sign:
raise ValueError("Sign key is not available")
return base64.standard_b64encode(HMAC.new(self.keys.sign, text.encode("utf-8"), SHA256).digest())
@staticmethod
def load_cache_data(msl_keys_path: Optional[Path] = None) -> Optional[MSLKeys]:
if not msl_keys_path or not msl_keys_path.is_file():
return None
msl_keys = jsonpickle.decode(msl_keys_path.read_text(encoding="utf-8"))
if msl_keys.mastertoken:
tokendata = json.loads(base64.standard_b64decode(msl_keys.mastertoken["tokendata"]).decode("utf-8"))
renewal_window = datetime.fromtimestamp(int(tokendata["renewalwindow"]), tz=timezone.utc)
remaining_hours = (renewal_window - datetime.now(timezone.utc)).total_seconds() / 3600
if remaining_hours < 10:
return None
return msl_keys
@staticmethod
def cache_keys(msl_keys: MSLKeys, msl_keys_path: Path) -> None:
with open(msl_keys_path, "w", encoding="utf-8") as cache_file:
cache_file.write(jsonpickle.encode(msl_keys, indent=4))
__all__ = ["MSL_ANDROID", "MSLKeys"]
modules/msl_base.py
+502
View File
@@ -0,0 +1,502 @@
from __future__ import annotations
import base64
import gzip
import json
import random
import re
import zlib
import jsonpickle
import requests
from io import BytesIO
from pathlib import Path
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from Cryptodome.Cipher import AES
from Cryptodome.Hash import HMAC, SHA256
from Cryptodome.Random import get_random_bytes
from Cryptodome.Util import Padding
# ---------------------------------------------------------------------------
# Core data objects
# ---------------------------------------------------------------------------
class MSLObject:
"""Base class providing a compact JSON repr for debugging."""
def __repr__(self) -> str:
return f"<{self.__class__.__name__} {jsonpickle.encode(self, unpicklable=False)}>"
class MSLKeys(MSLObject):
"""Holds the negotiated encryption / signing keys and master token."""
def __init__(
self,
encryption: Optional[bytes] = None,
sign: Optional[bytes] = None,
mastertoken: Optional[dict] = None,
) -> None:
self.encryption = encryption
self.sign = sign
self.mastertoken = mastertoken
# ---------------------------------------------------------------------------
# Widevine key helper (shared by Android, iOS, TV)
# ---------------------------------------------------------------------------
def get_widevine_key(kid: bytes, keys: List[Any], permissions: List[str]) -> Optional[bytes]:
"""Find the Widevine operator-session key whose permissions are a
superset of *permissions*. Permission names are normalised from
CamelCase to snake_case before comparison."""
normalized_perms = {re.sub(r'(?<!^)(?=[A-Z])', '_', p).lower() for p in permissions}
for key in keys:
if key.type != "OPERATOR_SESSION":
continue
key_perms = {p.lower() for p in (getattr(key, "permissions", None) or [])}
if normalized_perms <= key_perms:
return key.key
return None
# ---------------------------------------------------------------------------
# MSLBase – the shared foundation for all platform MSL classes
# ---------------------------------------------------------------------------
class MSLBase:
"""
Base class encapsulating the common MSL protocol logic shared by every
platform variant (Android, iOS, TV, Web).
Subclasses override:
- class-level DEFAULT_* constants
- ``build_request_headers()`` (platform-specific header set)
- ``handshake()`` (key-exchange / DRM negotiation)
- optionally ``generate_msg_header()`` for platform quirks
"""
# --- Subclasses must override these ------------------------------------
DEFAULT_HANDSHAKE_ENDPOINT: str = ""
DEFAULT_USER_AGENT: str = ""
DEFAULT_REQUEST_CONTEXT: str = "{}"
# --- Shared protocol constants -----------------------------------------
DEFAULT_PBO_VERSION: int = 2
DEFAULT_PBO_LANGUAGES: List[str] = ["en-US", "en"]
# -----------------------------------------------------------------------
# Constructor
# -----------------------------------------------------------------------
def __init__(
self,
session: requests.Session,
keys: MSLKeys,
message_id: int,
sender: str,
**_kwargs: Any,
) -> None:
self.session = session
self.keys = keys
self.sender = sender
self.message_id = message_id
# -----------------------------------------------------------------------
# JSON helpers
# -----------------------------------------------------------------------
@staticmethod
def parse_concatenated_json(message: str) -> List[Dict[str, Any]]:
"""Parse concatenated JSON objects from a single string (MSL wire format)."""
decoder = json.JSONDecoder()
items: List[Dict[str, Any]] = []
index = 0
length = len(message)
while index < length:
while index < length and message[index].isspace():
index += 1
if index >= length:
break
item, next_index = decoder.raw_decode(message, index)
items.append(item)
index = next_index
return items
@staticmethod
def stable_json(obj: Dict[str, Any]) -> str:
return json.dumps(obj, separators=(",", ":"), ensure_ascii=False)
# -----------------------------------------------------------------------
# Base64 helpers
# -----------------------------------------------------------------------
@staticmethod
def b64_encode_bytes(value: bytes) -> str:
return base64.b64encode(value).decode("ascii")
@staticmethod
def base64key_decode(payload: str) -> bytes:
"""Decode a URL-safe base64 JWK key, padding if necessary."""
remainder = len(payload) % 4
if remainder == 2:
payload += "=="
elif remainder == 3:
payload += "="
elif remainder != 0:
raise ValueError("Invalid base64 string")
return base64.urlsafe_b64decode(payload.encode("utf-8"))
# -----------------------------------------------------------------------
# Compression
# -----------------------------------------------------------------------
@staticmethod
def gzip_compress(data: bytes) -> bytes:
out = BytesIO()
with gzip.GzipFile(fileobj=out, mode="w") as gz:
gz.write(data)
return base64.standard_b64encode(out.getvalue())
# -----------------------------------------------------------------------
# Encryption / Signing
# -----------------------------------------------------------------------
def encrypt(self, plaintext: str) -> str:
if not self.keys.encryption:
raise ValueError("Encryption key is not available")
if not self.keys.mastertoken:
raise ValueError("Master token is not available")
iv = get_random_bytes(16)
tokendata = json.loads(
base64.standard_b64decode(self.keys.mastertoken["tokendata"]).decode("utf-8")
)
ciphertext = AES.new(self.keys.encryption, AES.MODE_CBC, iv).encrypt(
Padding.pad(plaintext.encode("utf-8"), 16)
)
return json.dumps(
{
"ciphertext": base64.standard_b64encode(ciphertext).decode("utf-8"),
"keyid": f"{self.sender}_{tokendata['sequencenumber']}",
"sha256": "AA==",
"iv": base64.standard_b64encode(iv).decode("utf-8"),
},
separators=(",", ":"),
)
def sign(self, text: str) -> bytes:
if not self.keys.sign:
raise ValueError("Sign key is not available")
return base64.standard_b64encode(
HMAC.new(self.keys.sign, text.encode("utf-8"), SHA256).digest()
)
# -----------------------------------------------------------------------
# AES-CBC convenience (used by MGK branch)
# -----------------------------------------------------------------------
@staticmethod
def aes_cbc_encrypt(key: bytes, iv: bytes, plaintext: bytes) -> bytes:
return AES.new(key, AES.MODE_CBC, iv).encrypt(Padding.pad(plaintext, AES.block_size))
@staticmethod
def aes_cbc_decrypt(key: bytes, iv: bytes, ciphertext: bytes) -> bytes:
return Padding.unpad(
AES.new(key, AES.MODE_CBC, iv).decrypt(ciphertext),
AES.block_size,
)
# -----------------------------------------------------------------------
# Message creation
# -----------------------------------------------------------------------
@staticmethod
def generate_msg_header(
message_id: int,
sender: str,
is_handshake: bool,
userauthdata: Optional[dict] = None,
keyrequestdata: Optional[dict] = None,
compression: Optional[str] = "GZIP",
languages: Optional[List[str]] = None,
recipient: Optional[str] = "Netflix",
) -> str:
header_data: Dict[str, Any] = {
"messageid": message_id,
"renewable": True,
"handshake": is_handshake,
"capabilities": {
"compressionalgos": [compression] if compression else [],
"languages": languages or ["en-US", "en"],
"encoderformats": ["JSON"],
},
"timestamp": int(datetime.now(timezone.utc).timestamp()),
"sender": sender,
"nonreplayable": False,
}
if recipient:
header_data["recipient"] = recipient
if userauthdata:
header_data["userauthdata"] = userauthdata
if keyrequestdata:
header_data["keyrequestdata"] = [keyrequestdata]
return jsonpickle.encode(header_data, unpicklable=False)
def create_message(
self,
application_data: Dict[str, Any],
userauthdata: Optional[dict] = None,
) -> str:
self.message_id += 1
headerdata = self.encrypt(
self.generate_msg_header(
message_id=self.message_id,
sender=self.sender,
is_handshake=False,
userauthdata=userauthdata,
)
)
message = json.dumps(
{
"headerdata": base64.standard_b64encode(headerdata.encode("utf-8")).decode("utf-8"),
"signature": self.sign(headerdata).decode("utf-8"),
"mastertoken": self.keys.mastertoken,
},
separators=(",", ":"),
)
compressed_application_data = self.gzip_compress(
json.dumps(application_data, separators=(",", ":")).encode("utf-8")
).decode("utf-8")
payload_dicts = [
{
"sequencenumber": 1,
"messageid": self.message_id,
"compressionalgo": "GZIP",
"data": compressed_application_data,
},
{
"sequencenumber": 2,
"messageid": self.message_id,
"endofmsg": True,
"data": "",
},
]
for payload_dict in payload_dicts:
payload_chunk = self.encrypt(json.dumps(payload_dict, separators=(",", ":")))
message += json.dumps(
{
"payload": base64.standard_b64encode(payload_chunk.encode("utf-8")).decode("utf-8"),
"signature": self.sign(payload_chunk).decode("utf-8"),
},
separators=(",", ":"),
)
return message
# -----------------------------------------------------------------------
# Message parsing / decryption
# -----------------------------------------------------------------------
def decrypt_payload_chunks(self, payload_chunks: List[Dict[str, str]]) -> Any:
if not self.keys.encryption:
raise ValueError("Encryption key is not available")
raw_data = ""
for payload_chunk in payload_chunks:
chunk_json = json.loads(
base64.standard_b64decode(payload_chunk["payload"]).decode("utf-8")
)
decrypted = AES.new(
key=self.keys.encryption,
mode=AES.MODE_CBC,
iv=base64.standard_b64decode(chunk_json["iv"]),
).decrypt(base64.standard_b64decode(chunk_json["ciphertext"]))
decrypted = Padding.unpad(decrypted, 16)
payload_json = json.loads(decrypted.decode("utf-8"))
payload_data = base64.standard_b64decode(payload_json["data"])
if payload_json.get("compressionalgo") == "GZIP":
payload_data = zlib.decompress(payload_data, 16 + zlib.MAX_WBITS)
raw_data += payload_data.decode("utf-8")
if not raw_data:
return None
try:
data = json.loads(raw_data)
except Exception:
return raw_data
if "error" in data:
return None
if "result" not in data:
return data
return data["result"]
def parse_message(self, message: str) -> Tuple[Dict[str, Any], Any]:
parsed = self.parse_concatenated_json(message)
header = parsed[0]
encrypted_chunks = parsed[1:] if len(parsed) > 1 else []
payload = self.decrypt_payload_chunks(encrypted_chunks) if encrypted_chunks else {}
return header, payload
# -----------------------------------------------------------------------
# Send message (common path – subclasses may override for extra checks)
# -----------------------------------------------------------------------
def send_message(
self,
endpoint: str,
params: Dict[str, str],
application_data: Dict[str, Any],
userauthdata: Optional[dict] = None,
headers: Optional[dict] = None,
proxy: Optional[Dict[str, str]] = None,
timeout: int = 30,
) -> Tuple[Dict[str, Any], Any]:
message = self.create_message(application_data, userauthdata)
request_kwargs: Dict[str, Any] = {
"url": endpoint,
"data": message,
"params": params,
"headers": headers,
"timeout": timeout,
}
if proxy:
request_kwargs["proxies"] = proxy
res = self.session.post(**request_kwargs)
if res.status_code != 200:
raise RuntimeError(
f"MSL request failed with HTTP {res.status_code}: {res.text[:500]}"
)
response_text = res.text or ""
stripped = response_text.lstrip()
if not stripped:
raise RuntimeError("MSL request failed: empty response body")
if not stripped.startswith("{"):
raise RuntimeError(
"MSL request failed: the server did not return concatenated MSL JSON. "
f"Content-Type: {res.headers.get('content-type', '')!r}. "
f"Body preview: {response_text[:500]!r}"
)
header, payload_data = self.parse_message(response_text)
if not header:
raise RuntimeError(
f"MSL request failed: parsed response does not contain a header. "
f"Body preview: {response_text[:500]!r}"
)
if "errordata" in header:
decoded_error = json.loads(
base64.standard_b64decode(header["errordata"].encode("utf-8")).decode("utf-8")
)
raise RuntimeError(f"MSL response contains an error: {decoded_error}")
return header, payload_data
# -----------------------------------------------------------------------
# PBO payload normalisation (shared by Android, iOS, TV)
# -----------------------------------------------------------------------
@classmethod
def normalize_application_data(cls, endpoint: str, application_data: Any) -> Any:
if not isinstance(application_data, dict):
return application_data
if cls._looks_like_wrapped_pbo_payload(application_data):
return application_data
route = cls._extract_pbo_route(application_data, endpoint)
if route is None:
return application_data
common = dict(getattr(cls, "DEFAULT_PBO_COMMON", {}))
if isinstance(application_data.get("common"), dict):
common.update(application_data["common"])
wrapped: Dict[str, Any] = {
"version": application_data.get("version", cls.DEFAULT_PBO_VERSION),
"common": common,
"url": route,
"languages": application_data.get("languages", list(cls.DEFAULT_PBO_LANGUAGES)),
"params": application_data.get("params", {}),
}
for key in ("path", "method", "route", "endpoint"):
wrapped.pop(key, None)
for key, value in application_data.items():
if key in wrapped or key in {
"version", "common", "languages", "params",
"path", "method", "route", "endpoint",
}:
continue
wrapped[key] = value
return wrapped
@staticmethod
def _looks_like_wrapped_pbo_payload(application_data: Dict[str, Any]) -> bool:
return (
"version" in application_data
and "common" in application_data
and "url" in application_data
and "languages" in application_data
and "params" in application_data
)
@staticmethod
def _extract_pbo_route(application_data: Dict[str, Any], endpoint: str) -> Optional[str]:
for key in ("url", "path", "route", "method", "endpoint"):
value = application_data.get(key)
if not isinstance(value, str) or not value.strip():
continue
value = value.strip()
return value if value.startswith("/") else f"/{value}"
endpoint_lower = endpoint.lower()
if "manifest" in endpoint_lower:
return "/manifest"
if "pbo_tokens" in endpoint_lower or "pbo_config" in endpoint_lower:
return None
return None
# -----------------------------------------------------------------------
# Cache I/O
# -----------------------------------------------------------------------
@staticmethod
def load_cache_data(msl_keys_path: Optional[Path] = None) -> Optional[MSLKeys]:
"""Load cached MSL keys from disk. Returns ``None`` if the cache is
missing, corrupt, or the token is about to expire (< 10 h remaining)."""
if not msl_keys_path or not msl_keys_path.is_file():
return None
msl_keys = jsonpickle.decode(msl_keys_path.read_text(encoding="utf-8"))
if msl_keys.mastertoken:
tokendata = json.loads(
base64.standard_b64decode(msl_keys.mastertoken["tokendata"]).decode("utf-8")
)
renewal_window = datetime.fromtimestamp(
int(tokendata["renewalwindow"]), tz=timezone.utc
)
remaining_hours = (renewal_window - datetime.now(timezone.utc)).total_seconds() / 3600
if remaining_hours < 10:
return None
return msl_keys
@staticmethod
def cache_keys(msl_keys: MSLKeys, msl_keys_path: Path) -> None:
"""Persist *msl_keys* to *msl_keys_path*."""
msl_keys_path.parent.mkdir(parents=True, exist_ok=True)
msl_keys_path.write_text(jsonpickle.encode(msl_keys, indent=4), encoding="utf-8")
modules/msl_mgk.py
+392 -407
View File
File diff suppressed because it is too large Load Diff
modules/msl_tv.py
+142 -329
View File
@@ -1,30 +1,28 @@
from __future__ import annotations
import base64
import gzip
import json
import random
import sys
import zlib
import jsonpickle
import requests
from io import BytesIO
from pathlib import Path
from datetime import datetime, timezone
from typing import Any, Dict, List, Optional, Tuple
from Cryptodome.Cipher import AES, PKCS1_OAEP
from Cryptodome.Hash import HMAC, SHA256
import jsonpickle
import requests
from Cryptodome.Cipher import PKCS1_OAEP
from Cryptodome.PublicKey import RSA
from Cryptodome.PublicKey.RSA import RsaKey
from Cryptodome.Random import get_random_bytes
from Cryptodome.Util import Padding
from pywidevine import Cdm as WidevineCdm, Device as WidevineDevice, PSSH
class MSLObject:
def __repr__(self) -> str:
return f"<{self.__class__.__name__} {jsonpickle.encode(self, unpicklable=False)}>"
from .msl_base import MSLBase, MSLKeys as _BaseMSLKeys, get_widevine_key
class MSLKeys(MSLObject):
# ---------------------------------------------------------------------------
# Platform-specific key container
# ---------------------------------------------------------------------------
class MSLKeys(_BaseMSLKeys):
"""TV MSL keys – extends the base with an RSA key and CDM session."""
def __init__(
self,
encryption: Optional[bytes] = None,
@@ -32,41 +30,59 @@ class MSLKeys(MSLObject):
rsa: Optional[RsaKey] = None,
mastertoken: Optional[dict] = None,
cdm_session: Any = None,
):
self.encryption = encryption
self.sign = sign
) -> None:
super().__init__(encryption=encryption, sign=sign, mastertoken=mastertoken)
self.rsa = rsa
self.mastertoken = mastertoken
self.cdm_session = cdm_session
class MSL_TV:
DEFAULT_HANDSHAKE_ENDPOINT = "https://nrdp25.prod.ftl.netflix.com/nq/nrdjs/pbo_tokens/%5E1.0.0/router"
DEFAULT_MANIFEST_ENDPOINT = "https://api-global.netflix.com/playapi/nrdjs/manifest/1"
DEFAULT_MANIFEST_PARAMS = {
# ---------------------------------------------------------------------------
# MSL_TV
# ---------------------------------------------------------------------------
class MSL_TV(MSLBase):
"""Netflix MSL client for TV devices, supporting both Widevine and RSA."""
# -- Platform constants --------------------------------------------------
DEFAULT_HANDSHAKE_ENDPOINT: str = (
"https://nrdp25.prod.ftl.netflix.com/nq/nrdjs/pbo_tokens/%5E1.0.0/router"
)
DEFAULT_MANIFEST_ENDPOINT: str = (
"https://api-global.netflix.com/playapi/nrdjs/manifest/1"
)
DEFAULT_MANIFEST_PARAMS: Dict[str, str] = {
"ab_ui_ver": "darwin",
"nrdapp_version": "2025.2.3.0",
}
DEFAULT_USER_AGENT = (
DEFAULT_USER_AGENT: str = (
"Netflix/2025.2.3.0 "
"(DEVTYPE=NFANDROID2-PRV-NVIDIASHIELDANDROIDTV2019; "
"Milo=1.0.6315; build_number=6315; build_sha=a1b915de)"
)
DEFAULT_REQUEST_CONTEXT = '{"appstate":"foreground","reason":"unknown"}'
DEFAULT_NRDJS_VERSION = "v3.12.55"
DEFAULT_NETJS_VERSION = "3.0.5"
DEFAULT_PBO_VERSION = 2
DEFAULT_PBO_COMMON = {
DEFAULT_REQUEST_CONTEXT: str = '{"appstate":"foreground","reason":"unknown"}'
DEFAULT_NRDJS_VERSION: str = "v3.12.55"
DEFAULT_NETJS_VERSION: str = "3.0.5"
DEFAULT_PBO_VERSION: int = 2
DEFAULT_PBO_COMMON: Dict[str, str] = {
"sdk": "2025.2.3.0",
"platform": "2025.2.3.0",
"application": "12.1.9-23083 R 2025.2 android-30-JPLAYER2 ninja_6==NVIDIA/mdarcy/mdarcy:11/RQ1A.210105.003/7825230_4040.2147:user/release-keys",
"uiversion": "UI-release-20260407_44745-gibbon-r100-darwinql-69067=5,80198=2,80211=3",
"application": (
"12.1.9-23083 R 2025.2 android-30-JPLAYER2 "
"ninja_6==NVIDIA/mdarcy/mdarcy:11/RQ1A.210105.003/"
"7825230_4040.2147:user/release-keys"
),
"uiversion": (
"UI-release-20260407_44745-gibbon-r100-darwinql-69067="
"5,80198=2,80211=3"
),
"uiPlatform": "tv_ui",
"clientVersion": "v3.12.55",
"apkVersion": "12.1.9",
}
DEFAULT_PBO_LANGUAGES = ["en-US", "en-PH", "en"]
DEFAULT_DEVICE_MODEL = "NVIDIA_SHIELD%20Android%20TV"
DEFAULT_PBO_LANGUAGES: List[str] = ["en-US", "en-PH", "en"]
DEFAULT_DEVICE_MODEL: str = "NVIDIA_SHIELD%20Android%20TV"
# -- Constructor ---------------------------------------------------------
def __init__(
self,
@@ -76,14 +92,13 @@ class MSL_TV:
sender: str,
user_auth: Optional[dict] = None,
drm: str = "widevine",
):
self.session = session
self.keys = keys
self.sender = sender
) -> None:
super().__init__(session=session, keys=keys, message_id=message_id, sender=sender)
self.user_auth = user_auth
self.message_id = message_id
self.drm = drm
# -- Dual-mode handshake (Widevine + RSA) --------------------------------
@classmethod
def handshake(
cls,
@@ -98,17 +113,21 @@ class MSL_TV:
endpoint: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
) -> MSLKeys:
"""Perform a key exchange using Widevine (if CDM available) or RSA."""
if cookies:
session.cookies.update(cookies)
cache_path = Path(msl_keys_path)
msl_keys = MSL_TV.load_cache_data(cache_path)
msl_keys = cls.load_cache_data(cache_path)
if msl_keys is not None and not new_msl:
return msl_keys
message_id = random.randint(0, pow(2, 52))
msl_keys = MSLKeys()
# ---- Choose DRM scheme ---------------------------------------------
if not cdm and drm == "widevine":
# No CDM provided – fall back to RSA key exchange
msl_keys.rsa = RSA.generate(2048)
assert msl_keys.rsa is not None
keyrequestdata = {
@@ -122,12 +141,15 @@ class MSL_TV:
},
}
elif drm == "widevine":
# CDM available – use Widevine
if not isinstance(cdm, WidevineCdm):
device = WidevineDevice.load(cdm_device)
cdm = WidevineCdm.from_device(device)
cdm_session = cdm.open()
msl_keys.cdm_session = cdm_session
challenge = cdm.get_license_challenge(cdm_session, PSSH.new(system_id=PSSH.SystemId.Widevine))
challenge = cdm.get_license_challenge(
cdm_session, PSSH.new(system_id=PSSH.SystemId.Widevine)
)
wv_request = base64.b64encode(challenge).decode("utf-8")
keyrequestdata = {
"scheme": "WIDEVINE",
@@ -138,6 +160,7 @@ class MSL_TV:
else:
raise ValueError(f"Unsupported DRM mode: {drm}")
# ---- Build & send handshake request --------------------------------
data = jsonpickle.encode(
{
"entityauthdata": {
@@ -147,7 +170,7 @@ class MSL_TV:
},
},
"headerdata": base64.standard_b64encode(
MSL_TV.generate_msg_header(
cls.generate_msg_header(
message_id=message_id,
sender=sender,
is_handshake=True,
@@ -181,10 +204,14 @@ class MSL_TV:
host="nrdp25.prod.ftl.netflix.com",
language="en-US,en-PH,en",
)
res = session.post(url=handshake_endpoint, data=data, headers=handshake_headers, timeout=30)
res = session.post(
url=handshake_endpoint, data=data, headers=handshake_headers, timeout=30
)
if res.status_code != 200:
raise RuntimeError(f"Key exchange failed: HTTP {res.status_code} {res.text[:500]}")
raise RuntimeError(
f"Key exchange failed: HTTP {res.status_code} {res.text[:500]}"
)
parsed = cls.parse_concatenated_json(res.text)
if not parsed:
@@ -193,12 +220,17 @@ class MSL_TV:
key_exchange = parsed[0]
if "errordata" in key_exchange:
decoded_error = base64.standard_b64decode(key_exchange["errordata"]).decode("utf-8")
decoded_error = base64.standard_b64decode(
key_exchange["errordata"]
).decode("utf-8")
error_json = json.loads(decoded_error)
raise RuntimeError(f"Key exchange failed: {error_json}")
if "headerdata" not in key_exchange:
raise RuntimeError(f"Key exchange failed: missing headerdata in response: {str(key_exchange)[:500]}")
raise RuntimeError(
f"Key exchange failed: missing headerdata in response: "
f"{str(key_exchange)[:500]}"
)
header_json = json.loads(
base64.standard_b64decode(key_exchange["headerdata"]).decode("utf-8")
@@ -206,37 +238,46 @@ class MSL_TV:
key_response_data = header_json["keyresponsedata"]
key_data = key_response_data["keydata"]
# ---- Derive encryption / signing keys from response ----------------
if cdm:
# Widevine path
cdm.parse_license(msl_keys.cdm_session, key_data["cdmkeyresponse"])
keys = cdm.get_keys(msl_keys.cdm_session)
msl_keys.encryption = MSL_TV.get_widevine_key(
wv_keys = cdm.get_keys(msl_keys.cdm_session)
msl_keys.encryption = get_widevine_key(
kid=base64.standard_b64decode(key_data["encryptionkeyid"]),
keys=keys,
keys=wv_keys,
permissions=["AllowEncrypt", "AllowDecrypt"],
)
msl_keys.sign = MSL_TV.get_widevine_key(
msl_keys.sign = get_widevine_key(
kid=base64.standard_b64decode(key_data["hmackeyid"]),
keys=keys,
keys=wv_keys,
permissions=["AllowSign", "AllowSignatureVerify"],
)
else:
# RSA path
assert msl_keys.rsa is not None
cipher_rsa = PKCS1_OAEP.new(msl_keys.rsa)
msl_keys.encryption = MSL_TV.base64key_decode(
msl_keys.encryption = cls.base64key_decode(
json.loads(
cipher_rsa.decrypt(base64.standard_b64decode(key_data["encryptionkey"])).decode("utf-8")
cipher_rsa.decrypt(
base64.standard_b64decode(key_data["encryptionkey"])
).decode("utf-8")
)["k"]
)
msl_keys.sign = MSL_TV.base64key_decode(
msl_keys.sign = cls.base64key_decode(
json.loads(
cipher_rsa.decrypt(base64.standard_b64decode(key_data["hmackey"])).decode("utf-8")
cipher_rsa.decrypt(
base64.standard_b64decode(key_data["hmackey"])
).decode("utf-8")
)["k"]
)
msl_keys.mastertoken = key_response_data["mastertoken"]
MSL_TV.cache_keys(msl_keys, cache_path)
cls.cache_keys(msl_keys, cache_path)
return msl_keys
# -- Platform-specific request headers -----------------------------------
@staticmethod
def build_request_headers(
request_name: str,
@@ -251,6 +292,7 @@ class MSL_TV:
language: Optional[str] = "en-US,en-PH,en",
device_model: Optional[str] = None,
) -> Dict[str, str]:
"""Build the HTTP headers required for TV MSL requests."""
headers: Dict[str, str] = {
"Host": host or "nrdp25.prod.ftl.netflix.com",
"Language": language or "en-US,en-PH,en",
@@ -285,10 +327,15 @@ class MSL_TV:
headers.update(extra_headers)
return headers
# -- Manifest defaults ---------------------------------------------------
@staticmethod
def manifest_request_defaults() -> Tuple[str, Dict[str, str]]:
"""Return the default manifest endpoint and query params for TV."""
return MSL_TV.DEFAULT_MANIFEST_ENDPOINT, dict(MSL_TV.DEFAULT_MANIFEST_PARAMS)
# -- Custom generate_msg_header (TV-specific languages) ------------------
@staticmethod
def generate_msg_header(
message_id: int,
@@ -298,37 +345,19 @@ class MSL_TV:
keyrequestdata: Optional[dict] = None,
compression: Optional[str] = "GZIP",
) -> str:
header_data: Dict[str, Any] = {
"messageid": message_id,
"renewable": True,
"handshake": is_handshake,
"capabilities": {
"compressionalgos": [compression] if compression else [],
"languages": ["en-US", "en-PH", "en"],
"encoderformats": ["JSON"],
},
"timestamp": int(datetime.now(timezone.utc).timestamp()),
"sender": sender,
"nonreplayable": False,
"recipient": "Netflix",
}
if userauthdata:
header_data["userauthdata"] = userauthdata
if keyrequestdata:
header_data["keyrequestdata"] = [keyrequestdata]
return jsonpickle.encode(header_data, unpicklable=False)
"""Generate an MSL message header with TV-specific language list."""
return MSLBase.generate_msg_header(
message_id=message_id,
sender=sender,
is_handshake=is_handshake,
userauthdata=userauthdata,
keyrequestdata=keyrequestdata,
compression=compression,
languages=["en-US", "en-PH", "en"],
)
@staticmethod
def get_widevine_key(kid: bytes, keys: List[Any], permissions: List[str]) -> Optional[bytes]:
import re
normalized_perms = {re.sub(r'(?<!^)(?=[A-Z])', '_', p).lower() for p in permissions}
for key in keys:
if key.type != "OPERATOR_SESSION":
continue
key_perms = {p.lower() for p in (getattr(key, "permissions", None) or [])}
if normalized_perms <= key_perms:
return key.key
return None
# -- Override send_message to add PBO normalisation ----------------------
# BUG FIX: original used sys.exit(print(...)) on error; now raises RuntimeError.
def send_message(
self,
@@ -338,262 +367,46 @@ class MSL_TV:
userauthdata: Optional[dict] = None,
headers: Optional[dict] = None,
proxy: Optional[Dict[str, str]] = None,
timeout: int = 30,
) -> Tuple[Dict[str, Any], Any]:
normalized_application_data = self.normalize_application_data(endpoint, application_data)
message = self.create_message(normalized_application_data, userauthdata)
request_kwargs: Dict[str, Any] = {
"url": endpoint,
"data": message,
"params": params,
"headers": headers,
"timeout": 30,
}
if proxy:
request_kwargs["proxies"] = proxy
res = self.session.post(**request_kwargs)
header, payload_data = self.parse_message(res.text)
if "errordata" in header:
decoded_error = json.loads(
base64.standard_b64decode(header["errordata"].encode("utf-8")).decode("utf-8")
)
sys.exit(print(f"MSL response contains an error: {decoded_error}"))
return header, payload_data
"""Send an MSL message with PBO payload normalisation.
@classmethod
def normalize_application_data(cls, endpoint: str, application_data: Any) -> Any:
if not isinstance(application_data, dict):
return application_data
if cls._looks_like_wrapped_pbo_payload(application_data):
return application_data
route = cls._extract_pbo_route(application_data, endpoint)
if route is None:
return application_data
common = dict(cls.DEFAULT_PBO_COMMON)
if isinstance(application_data.get("common"), dict):
common.update(application_data["common"])
wrapped: Dict[str, Any] = {
"version": application_data.get("version", cls.DEFAULT_PBO_VERSION),
"common": common,
"url": route,
"languages": application_data.get("languages", list(cls.DEFAULT_PBO_LANGUAGES)),
"params": application_data.get("params", {}),
}
for key in ("path", "method", "route", "endpoint"):
wrapped.pop(key, None)
for key, value in application_data.items():
if key in wrapped or key in {"version", "common", "languages", "params", "path", "method", "route", "endpoint"}:
continue
wrapped[key] = value
return wrapped
@staticmethod
def _looks_like_wrapped_pbo_payload(application_data: Dict[str, Any]) -> bool:
return (
"version" in application_data
and "common" in application_data
and "url" in application_data
and "languages" in application_data
and "params" in application_data
Raises :class:`RuntimeError` on MSL errors instead of calling
``sys.exit()``.
"""
normalized = self.normalize_application_data(endpoint, application_data)
return super().send_message(
endpoint=endpoint,
params=params,
application_data=normalized,
userauthdata=userauthdata,
headers=headers,
proxy=proxy,
timeout=timeout,
)
@staticmethod
def _extract_pbo_route(application_data: Dict[str, Any], endpoint: str) -> Optional[str]:
for key in ("url", "path", "route", "method", "endpoint"):
value = application_data.get(key)
if not isinstance(value, str) or not value.strip():
continue
value = value.strip()
return value if value.startswith("/") else f"/{value}"
endpoint_lower = endpoint.lower()
if "manifest" in endpoint_lower:
return "/manifest"
if "pbo_tokens" in endpoint_lower or "pbo_config" in endpoint_lower:
return None
return None
def create_message(self, application_data: Dict[str, Any], userauthdata: Optional[dict] = None) -> str:
self.message_id += 1
headerdata = self.encrypt(
self.generate_msg_header(
message_id=self.message_id,
sender=self.sender,
is_handshake=False,
userauthdata=userauthdata,
)
)
message = json.dumps(
{
"headerdata": base64.standard_b64encode(headerdata.encode("utf-8")).decode("utf-8"),
"signature": self.sign(headerdata).decode("utf-8"),
"mastertoken": self.keys.mastertoken,
},
separators=(",", ":"),
)
compressed_application_data = self.gzip_compress(
json.dumps(application_data, separators=(",", ":")).encode("utf-8")
).decode("utf-8")
payload_dicts = [
{
"sequencenumber": 1,
"messageid": self.message_id,
"compressionalgo": "GZIP",
"data": compressed_application_data,
},
{
"sequencenumber": 2,
"messageid": self.message_id,
"endofmsg": True,
"data": "",
},
]
for payload_dict in payload_dicts:
payload_chunk = self.encrypt(json.dumps(payload_dict, separators=(",", ":")))
message += json.dumps(
{
"payload": base64.standard_b64encode(payload_chunk.encode("utf-8")).decode("utf-8"),
"signature": self.sign(payload_chunk).decode("utf-8"),
},
separators=(",", ":"),
)
return message
def decrypt_payload_chunks(self, payload_chunks: List[Dict[str, str]]) -> Any:
raw_data = ""
assert self.keys.encryption is not None
for payload_chunk in payload_chunks:
payload_chunk_json = json.loads(base64.standard_b64decode(payload_chunk["payload"]).decode("utf-8"))
payload_decrypted = AES.new(
key=self.keys.encryption,
mode=AES.MODE_CBC,
iv=base64.standard_b64decode(payload_chunk_json["iv"]),
).decrypt(base64.standard_b64decode(payload_chunk_json["ciphertext"]))
payload_decrypted = Padding.unpad(payload_decrypted, 16)
payload_decrypted_json = json.loads(payload_decrypted.decode("utf-8"))
payload_data = base64.standard_b64decode(payload_decrypted_json["data"])
if payload_decrypted_json.get("compressionalgo") == "GZIP":
payload_data = zlib.decompress(payload_data, 16 + zlib.MAX_WBITS)
raw_data += payload_data.decode("utf-8")
if not raw_data:
return None
try:
data = json.loads(raw_data)
except Exception:
return raw_data
if "error" in data:
return None
if "result" not in data:
return data
return data["result"]
@staticmethod
def parse_concatenated_json(message: str) -> List[Dict[str, Any]]:
decoder = json.JSONDecoder()
items: List[Dict[str, Any]] = []
index = 0
length = len(message)
while index < length:
while index < length and message[index].isspace():
index += 1
if index >= length:
break
item, next_index = decoder.raw_decode(message, index)
items.append(item)
index = next_index
return items
def parse_message(self, message: str) -> Tuple[Dict[str, Any], Any]:
parsed_message = self.parse_concatenated_json(message)
header = parsed_message[0]
encrypted_payload_chunks = parsed_message[1:] if len(parsed_message) > 1 else []
payload_chunks = self.decrypt_payload_chunks(encrypted_payload_chunks) if encrypted_payload_chunks else {}
return header, payload_chunks
@staticmethod
def gzip_compress(data: bytes) -> bytes:
out = BytesIO()
with gzip.GzipFile(fileobj=out, mode="w") as gzip_file:
gzip_file.write(data)
return base64.standard_b64encode(out.getvalue())
@staticmethod
def base64key_decode(payload: str) -> bytes:
length = len(payload) % 4
if length == 2:
payload += "=="
elif length == 3:
payload += "="
elif length != 0:
raise ValueError("Invalid base64 string")
return base64.urlsafe_b64decode(payload.encode("utf-8"))
def encrypt(self, plaintext: str) -> str:
if not self.keys.encryption:
raise ValueError("Encryption key is not available")
if not self.keys.mastertoken:
raise ValueError("Master token is not available")
iv = get_random_bytes(16)
tokendata = json.loads(base64.standard_b64decode(self.keys.mastertoken["tokendata"]).decode("utf-8"))
return json.dumps(
{
"ciphertext": base64.standard_b64encode(
AES.new(self.keys.encryption, AES.MODE_CBC, iv).encrypt(
Padding.pad(plaintext.encode("utf-8"), 16)
)
).decode("utf-8"),
"keyid": f"{self.sender}_{tokendata['sequencenumber']}",
"sha256": "AA==",
"iv": base64.standard_b64encode(iv).decode("utf-8"),
}
)
def sign(self, text: str) -> bytes:
if not self.keys.sign:
raise ValueError("Sign key is not available")
return base64.standard_b64encode(HMAC.new(self.keys.sign, text.encode("utf-8"), SHA256).digest())
# -- Cache I/O overrides (RSA key serialisation) -------------------------
@staticmethod
def load_cache_data(msl_keys_path: Optional[Path] = None) -> Optional[MSLKeys]:
if not msl_keys_path or not msl_keys_path.is_file():
"""Load cached keys, re-importing the RSA key from its PEM form."""
msl_keys = MSLBase.load_cache_data(msl_keys_path)
if msl_keys is None:
return None
msl_keys = jsonpickle.decode(msl_keys_path.read_text(encoding="utf-8"))
if msl_keys.rsa:
# Re-import RSA key from serialised PEM form
if getattr(msl_keys, "rsa", None):
msl_keys.rsa = RSA.import_key(msl_keys.rsa)
if msl_keys.mastertoken:
tokendata = json.loads(base64.standard_b64decode(msl_keys.mastertoken["tokendata"]).decode("utf-8"))
renewal_window = datetime.fromtimestamp(int(tokendata["renewalwindow"]), tz=timezone.utc)
remaining_hours = (renewal_window - datetime.now(timezone.utc)).total_seconds() / 3600
if remaining_hours < 10:
return None
return msl_keys
@staticmethod
def cache_keys(msl_keys: MSLKeys, msl_keys_path: Path) -> None:
"""Persist keys, exporting the RSA key to PEM for JSON compatibility."""
original_rsa = msl_keys.rsa
exported_rsa = None
if msl_keys.rsa:
exported_rsa = msl_keys.rsa.export_key()
msl_keys.rsa = exported_rsa
with open(msl_keys_path, "w", encoding="utf-8") as cache_file:
cache_file.write(jsonpickle.encode(msl_keys, indent=4))
msl_keys.rsa = msl_keys.rsa.export_key()
MSLBase.cache_keys(msl_keys, msl_keys_path)
if original_rsa:
msl_keys.rsa = original_rsa
msl_keys.rsa = original_rsa
__all__ = ["MSL_TV", "MSLKeys"]
modules/msl_web.py
+111 -186
View File
@@ -1,54 +1,62 @@
from __future__ import annotations
import base64
import gzip
import json
import random
import zlib
import jsonpickle
import requests
from collections import OrderedDict
from datetime import datetime, timezone
from http.cookiejar import CookieJar
from io import BytesIO
from pathlib import Path
from typing import Any, Dict, Iterable, List, Optional, Tuple
from collections import OrderedDict
from Cryptodome.Cipher import AES, PKCS1_OAEP
from Cryptodome.Hash import HMAC, SHA256
import jsonpickle
import requests
from Cryptodome.Cipher import PKCS1_OAEP
from Cryptodome.PublicKey import RSA
from Cryptodome.PublicKey.RSA import RsaKey
from Cryptodome.Random import get_random_bytes
from Cryptodome.Util import Padding
from .msl_base import MSLBase, MSLKeys as _BaseMSLKeys
class MSLObject:
def __repr__(self) -> str:
return f"<{self.__class__.__name__} {jsonpickle.encode(self, unpicklable=False)}>"
# ---------------------------------------------------------------------------
# Platform-specific key container
# ---------------------------------------------------------------------------
class MSLKeys(_BaseMSLKeys):
"""Web MSL keys – extends the base with an RSA key (no CDM session)."""
class MSLKeys(MSLObject):
def __init__(
self,
encryption: Optional[bytes] = None,
sign: Optional[bytes] = None,
rsa: Optional[RsaKey] = None,
mastertoken: Optional[dict] = None,
):
self.encryption = encryption
self.sign = sign
) -> None:
super().__init__(encryption=encryption, sign=sign, mastertoken=mastertoken)
self.rsa = rsa
self.mastertoken = mastertoken
class MSL_WEB:
DEFAULT_HANDSHAKE_ENDPOINT = "https://www.netflix.com/nq/msl_v1/nrdjs/pbo_tokens/%5E1.0.0/router"
DEFAULT_USER_AGENT = (
# ---------------------------------------------------------------------------
# MSL_WEB
# ---------------------------------------------------------------------------
class MSL_WEB(MSLBase):
"""Netflix MSL client for web browsers using RSA key exchange."""
# -- Platform constants --------------------------------------------------
DEFAULT_HANDSHAKE_ENDPOINT: str = (
"https://www.netflix.com/nq/msl_v1/nrdjs/pbo_tokens/%5E1.0.0/router"
)
DEFAULT_USER_AGENT: str = (
"Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
"AppleWebKit/537.36 (KHTML, like Gecko) "
"Chrome/146.0.0.0 Safari/537.36"
)
DEFAULT_REQUEST_CONTEXT = '{"appstate":"foreground"}'
DEFAULT_NRDJS_VERSION = "v3.11.512"
DEFAULT_NETJS_VERSION = "3.0.5"
DEFAULT_REQUEST_CONTEXT: str = '{"appstate":"foreground"}'
DEFAULT_NRDJS_VERSION: str = "v3.11.512"
DEFAULT_NETJS_VERSION: str = "3.0.5"
# -- Constructor ---------------------------------------------------------
def __init__(
self,
@@ -57,12 +65,11 @@ class MSL_WEB:
message_id: int,
sender: str,
user_auth: Optional[dict] = None,
):
self.session = session
self.keys = keys
self.sender = sender
) -> None:
super().__init__(session=session, keys=keys, message_id=message_id, sender=sender)
self.user_auth = user_auth
self.message_id = message_id
# -- RSA handshake -------------------------------------------------------
@classmethod
def handshake(
@@ -75,6 +82,7 @@ class MSL_WEB:
endpoint: Optional[str] = None,
headers: Optional[Dict[str, str]] = None,
) -> MSLKeys:
"""Perform an RSA (ASYMMETRIC_WRAPPED) key exchange."""
if cookies:
session.cookies.update(cookies)
@@ -140,7 +148,9 @@ class MSL_WEB:
timeout=30,
)
if response.status_code != 200:
raise RuntimeError(f"Key exchange failed: HTTP {response.status_code} {response.text[:500]}")
raise RuntimeError(
f"Key exchange failed: HTTP {response.status_code} {response.text[:500]}"
)
parsed = cls.parse_concatenated_json(response.text)
if not parsed:
@@ -148,26 +158,42 @@ class MSL_WEB:
header = parsed[0]
if "errordata" in header:
decoded_error = base64.standard_b64decode(header["errordata"]).decode("utf-8")
decoded_error = base64.standard_b64decode(header["errordata"]).decode(
"utf-8"
)
raise RuntimeError(f"Key exchange failed: {decoded_error}")
if "headerdata" not in header:
raise RuntimeError(f"Key exchange failed: missing headerdata: {str(header)[:500]}")
raise RuntimeError(
f"Key exchange failed: missing headerdata: {str(header)[:500]}"
)
header_json = json.loads(base64.standard_b64decode(header["headerdata"]).decode("utf-8"))
header_json = json.loads(
base64.standard_b64decode(header["headerdata"]).decode("utf-8")
)
key_data = header_json["keyresponsedata"]["keydata"]
cipher_rsa = PKCS1_OAEP.new(keys.rsa)
keys.encryption = cls.base64key_decode(
json.loads(cipher_rsa.decrypt(base64.standard_b64decode(key_data["encryptionkey"])).decode("utf-8"))["k"]
json.loads(
cipher_rsa.decrypt(
base64.standard_b64decode(key_data["encryptionkey"])
).decode("utf-8")
)["k"]
)
keys.sign = cls.base64key_decode(
json.loads(cipher_rsa.decrypt(base64.standard_b64decode(key_data["hmackey"])).decode("utf-8"))["k"]
json.loads(
cipher_rsa.decrypt(
base64.standard_b64decode(key_data["hmackey"])
).decode("utf-8")
)["k"]
)
keys.mastertoken = header_json["keyresponsedata"]["mastertoken"]
cls.cache_keys(keys, cache_path)
return keys
# -- Platform-specific request headers -----------------------------------
@staticmethod
def build_request_headers(
request_name: str,
@@ -179,6 +205,7 @@ class MSL_WEB:
expiry_timeout: Optional[int] = 12750,
extra_headers: Optional[Dict[str, str]] = None,
) -> Dict[str, str]:
"""Build the HTTP headers required for Web MSL requests."""
headers: Dict[str, str] = {
"User-Agent": user_agent or MSL_WEB.DEFAULT_USER_AGENT,
"Accept": "*/*",
@@ -207,6 +234,8 @@ class MSL_WEB:
headers.update(extra_headers)
return headers
# -- Custom generate_msg_header (empty languages / encoderformats) -------
@staticmethod
def generate_msg_header(
message_id: int,
@@ -216,6 +245,11 @@ class MSL_WEB:
keyrequestdata: Optional[dict] = None,
compression: Optional[str] = "GZIP",
) -> str:
"""Generate an MSL message header with Web-specific capabilities.
The Web platform sends empty ``languages`` and ``encoderformats``
arrays in the capabilities block.
"""
header_data: Dict[str, Any] = {
"messageid": message_id,
"renewable": True,
@@ -236,6 +270,8 @@ class MSL_WEB:
header_data["keyrequestdata"] = [keyrequestdata]
return jsonpickle.encode(header_data, unpicklable=False)
# -- Web-specific send_message (no PBO normalisation) --------------------
def send_message(
self,
endpoint: str,
@@ -244,177 +280,51 @@ class MSL_WEB:
userauthdata: Optional[dict] = None,
headers: Optional[Dict[str, str]] = None,
) -> Tuple[Dict[str, Any], Any]:
"""Send an MSL message without PBO payload normalisation.
Raises :class:`RuntimeError` on MSL errors.
"""
message = self.create_message(application_data, userauthdata)
response = self.session.post(url=endpoint, params=params, data=message, headers=headers, timeout=30)
response = self.session.post(
url=endpoint, params=params, data=message, headers=headers, timeout=30
)
response.raise_for_status()
header, payload = self.parse_message(response.text)
if "errordata" in header:
decoded_error = json.loads(base64.standard_b64decode(header["errordata"]).decode("utf-8"))
decoded_error = json.loads(
base64.standard_b64decode(header["errordata"]).decode("utf-8")
)
raise RuntimeError(f"MSL response contains an error: {decoded_error}")
return header, payload
def create_message(self, application_data: Any, userauthdata: Optional[dict] = None) -> str:
self.message_id += 1
headerdata = self.encrypt(
self.generate_msg_header(
message_id=self.message_id,
sender=self.sender,
is_handshake=False,
userauthdata=userauthdata,
)
)
message = json.dumps(
{
"headerdata": base64.standard_b64encode(headerdata.encode("utf-8")).decode("utf-8"),
"signature": self.sign(headerdata).decode("utf-8"),
"mastertoken": self.keys.mastertoken,
},
separators=(",", ":"),
)
compressed_data = self.gzip_compress(json.dumps(application_data, separators=(",", ":")).encode("utf-8")).decode("utf-8")
payloads = [
{
"sequencenumber": 1,
"messageid": self.message_id,
"compressionalgo": "GZIP",
"data": compressed_data,
},
{
"sequencenumber": 2,
"messageid": self.message_id,
"endofmsg": True,
"data": "",
},
]
for payload in payloads:
encrypted_chunk = self.encrypt(json.dumps(payload, separators=(",", ":")))
message += json.dumps(
{
"payload": base64.standard_b64encode(encrypted_chunk.encode("utf-8")).decode("utf-8"),
"signature": self.sign(encrypted_chunk).decode("utf-8"),
},
separators=(",", ":"),
)
return message
def parse_message(self, message: str) -> Tuple[Dict[str, Any], Any]:
parsed = self.parse_concatenated_json(message)
header = parsed[0]
payload_chunks = parsed[1:] if len(parsed) > 1 else []
payload = self.decrypt_payload_chunks(payload_chunks) if payload_chunks else {}
return header, payload
def decrypt_payload_chunks(self, payload_chunks: List[Dict[str, str]]) -> Any:
if not self.keys.encryption:
raise ValueError("Encryption key is not available")
raw_data = ""
for payload_chunk in payload_chunks:
chunk_json = json.loads(base64.standard_b64decode(payload_chunk["payload"]).decode("utf-8"))
decrypted = AES.new(
key=self.keys.encryption,
mode=AES.MODE_CBC,
iv=base64.standard_b64decode(chunk_json["iv"]),
).decrypt(base64.standard_b64decode(chunk_json["ciphertext"]))
decrypted = Padding.unpad(decrypted, 16)
payload_json = json.loads(decrypted.decode("utf-8"))
payload_data = base64.standard_b64decode(payload_json["data"])
if payload_json.get("compressionalgo") == "GZIP":
payload_data = zlib.decompress(payload_data, 16 + zlib.MAX_WBITS)
raw_data += payload_data.decode("utf-8")
data = json.loads(raw_data)
if "error" in data:
return None
return data.get("result", data)
def encrypt(self, plaintext: str) -> str:
if not self.keys.encryption:
raise ValueError("Encryption key is not available")
if not self.keys.mastertoken:
raise ValueError("Master token is not available")
iv = get_random_bytes(16)
tokendata = json.loads(base64.standard_b64decode(self.keys.mastertoken["tokendata"]).decode("utf-8"))
ciphertext = AES.new(self.keys.encryption, AES.MODE_CBC, iv).encrypt(Padding.pad(plaintext.encode("utf-8"), 16))
return json.dumps(
{
"ciphertext": base64.standard_b64encode(ciphertext).decode("utf-8"),
"keyid": f"{self.sender}_{tokendata['sequencenumber']}",
"sha256": "AA==",
"iv": base64.standard_b64encode(iv).decode("utf-8"),
},
separators=(",", ":"),
)
def sign(self, text: str) -> bytes:
if not self.keys.sign:
raise ValueError("Sign key is not available")
return base64.standard_b64encode(HMAC.new(self.keys.sign, text.encode("utf-8"), SHA256).digest())
@staticmethod
def parse_concatenated_json(message: str) -> List[Dict[str, Any]]:
decoder = json.JSONDecoder()
items: List[Dict[str, Any]] = []
index = 0
while index < len(message):
while index < len(message) and message[index].isspace():
index += 1
if index >= len(message):
break
item, index = decoder.raw_decode(message, index)
items.append(item)
return items
@staticmethod
def gzip_compress(data: bytes) -> bytes:
out = BytesIO()
with gzip.GzipFile(fileobj=out, mode="w") as handle:
handle.write(data)
return base64.standard_b64encode(out.getvalue())
@staticmethod
def base64key_decode(payload: str) -> bytes:
remainder = len(payload) % 4
if remainder == 2:
payload += "=="
elif remainder == 3:
payload += "="
elif remainder != 0:
raise ValueError("Invalid base64 string")
return base64.urlsafe_b64decode(payload.encode("utf-8"))
# -- Cache I/O overrides (RSA key serialisation) -------------------------
@staticmethod
def load_cache_data(msl_keys_path: Optional[Path] = None) -> Optional[MSLKeys]:
if not msl_keys_path or not msl_keys_path.is_file():
"""Load cached keys, re-importing the RSA key from its PEM form."""
msl_keys = MSLBase.load_cache_data(msl_keys_path)
if msl_keys is None:
return None
msl_keys = jsonpickle.decode(msl_keys_path.read_text(encoding="utf-8"))
if msl_keys.rsa:
# Re-import RSA key from serialised PEM form
if getattr(msl_keys, "rsa", None):
msl_keys.rsa = RSA.import_key(msl_keys.rsa)
if msl_keys.mastertoken:
tokendata = json.loads(base64.standard_b64decode(msl_keys.mastertoken["tokendata"]).decode("utf-8"))
renewal_window = datetime.fromtimestamp(int(tokendata["renewalwindow"]), tz=timezone.utc)
if (renewal_window - datetime.now(timezone.utc)).total_seconds() / 3600 < 10:
return None
return msl_keys
@staticmethod
def cache_keys(msl_keys: MSLKeys, msl_keys_path: Path) -> None:
"""Persist keys, exporting the RSA key to PEM for JSON compatibility."""
original_rsa = msl_keys.rsa
if msl_keys.rsa:
msl_keys.rsa = msl_keys.rsa.export_key()
msl_keys_path.write_text(jsonpickle.encode(msl_keys, indent=4), encoding="utf-8")
MSLBase.cache_keys(msl_keys, msl_keys_path)
if original_rsa:
msl_keys.rsa = original_rsa
# -- Cookie-jar helpers (Web-specific) -----------------------------------
@staticmethod
def cookiejar_to_list(cookie_jar: CookieJar) -> List[Dict[str, Any]]:
"""Serialise a :class:`CookieJar` to a list of dicts."""
cookies: List[Dict[str, Any]] = []
for cookie in cookie_jar:
cookies.append(
@@ -433,13 +343,17 @@ class MSL_WEB:
@staticmethod
def cookiejar_to_ordered_dict(cookie_jar: CookieJar) -> Dict[str, str]:
"""Convert a :class:`CookieJar` to an :class:`OrderedDict` of name→value."""
cookies: Dict[str, str] = OrderedDict()
for cookie in cookie_jar:
cookies[cookie.name] = cookie.value
return cookies
@staticmethod
def save_cookie_values(cookie_jar: CookieJar, output_path: str | Path) -> Dict[str, str]:
def save_cookie_values(
cookie_jar: CookieJar, output_path: str | Path
) -> Dict[str, str]:
"""Save cookie name→value pairs to a JSON file and return them."""
path = Path(output_path)
payload = MSL_WEB.cookiejar_to_ordered_dict(cookie_jar)
path.write_text(json.dumps(payload, indent=2), encoding="utf-8")
@@ -447,11 +361,19 @@ class MSL_WEB:
@staticmethod
def save_cookiejar(cookie_jar: CookieJar, output_path: str | Path) -> None:
"""Save the full cookie jar to a JSON file."""
path = Path(output_path)
path.write_text(json.dumps(MSL_WEB.cookiejar_to_list(cookie_jar), indent=2), encoding="utf-8")
path.write_text(
json.dumps(MSL_WEB.cookiejar_to_list(cookie_jar), indent=2),
encoding="utf-8",
)
@staticmethod
def load_cookiejar(session: requests.Session, source: str | Path | Iterable[Dict[str, Any]] | Dict[str, str]) -> None:
def load_cookiejar(
session: requests.Session,
source: str | Path | Iterable[Dict[str, Any]] | Dict[str, str],
) -> None:
"""Load cookies into *session* from a file path, iterable, or dict."""
if isinstance(source, (str, Path)):
path = Path(source)
if not path.is_file():
@@ -473,4 +395,7 @@ class MSL_WEB:
secure=item.get("secure", False),
expires=item.get("expires"),
rest=item.get("rest", {}),
)
)
__all__ = ["MSL_WEB", "MSLKeys"]