From c21869782083ec5aa63e94279cad315bd623cf8d Mon Sep 17 00:00:00 2001 From: Hugoved Date: Sun, 17 May 2026 23:59:51 +0200 Subject: [PATCH] Create msl_mgk.py --- msl_mgk.py | 1008 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 1008 insertions(+) create mode 100644 msl_mgk.py diff --git a/msl_mgk.py b/msl_mgk.py new file mode 100644 index 0000000..3485d91 --- /dev/null +++ b/msl_mgk.py @@ -0,0 +1,1008 @@ +from __future__ import annotations +import base64 +import gzip +import json +import os +import random +import re +import zlib +from datetime import datetime, timezone +from enum import Enum +from io import BytesIO +from pathlib import Path +from typing import Any, Dict, List, Optional, Tuple, Union +import jsonpickle +import requests +from Cryptodome.Cipher import AES +from Cryptodome.Hash import HMAC, SHA256, SHA384 +from Cryptodome.Random import get_random_bytes +from Cryptodome.Util.Padding import pad, unpad + + +class MSLObject: + def __repr__(self) -> str: + return f"<{self.__class__.__name__} {self.__dict__}>" + + +class MSLKeys(MSLObject): + def __init__( + self, + encryption: Optional[bytes] = None, + sign: Optional[bytes] = None, + mastertoken: Optional[dict] = None, + wrapdata: Optional[bytes] = None, + derivation_key: Optional[bytes] = None, + ) -> None: + self.encryption = encryption + self.sign = sign + self.mastertoken = mastertoken + self.wrapdata = wrapdata + self.derivation_key = derivation_key + + +class Scheme(Enum): + def __str__(self) -> str: + return str(self.value) + + +class EntityAuthenticationSchemes(Scheme): + ModelGroup = "MGK" + + +class UserAuthenticationSchemes(Scheme): + EmailPassword = "EMAIL_PASSWORD" + NetflixIDCookies = "NETFLIXID" + UserIDToken = "USER_ID_TOKEN" + + +class EntityAuthentication(MSLObject): + def __init__(self, scheme: EntityAuthenticationSchemes, authdata: Dict[str, Any]) -> None: + self.scheme = str(scheme) + self.authdata = authdata + + @classmethod + def ModelGroup(cls, identity: str) -> "EntityAuthentication": + return cls(EntityAuthenticationSchemes.ModelGroup, {"identity": identity}) + + +class UserAuthentication(MSLObject): + def __init__(self, scheme: UserAuthenticationSchemes, authdata: Dict[str, Any]) -> None: + self.scheme = str(scheme) + self.authdata = authdata + + @classmethod + def EmailPassword(cls, email: str, password: str) -> "UserAuthentication": + return cls( + UserAuthenticationSchemes.EmailPassword, + {"email": email, "password": password}, + ) + + @classmethod + def UserIDToken( + cls, + token_data: str, + signature: str, + master_token: dict, + ) -> "UserAuthentication": + return cls( + UserAuthenticationSchemes.UserIDToken, + { + "useridtoken": { + "tokendata": token_data, + "signature": signature, + }, + "mastertoken": master_token, + }, + ) + + @classmethod + def NetflixIDCookies( + cls, + netflixid: Optional[str], + securenetflixid: Optional[str], + ) -> "UserAuthentication": + return cls( + UserAuthenticationSchemes.NetflixIDCookies, + { + "netflixid": netflixid, + "securenetflixid": securenetflixid, + }, + ) + + +class MSL: + DEFAULT_MANIFEST_ENDPOINT = "https://api-global.netflix.com/playapi/nrdjs/manifest/1" + DEFAULT_MANIFEST_PARAMS = { + "ab_ui_ver": "darwin", + "nrdapp_version": "2025.2.2.0", + } + DEFAULT_USER_AGENT = ( + "Netflix/2025.2.2.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.11.512" + DEFAULT_NETJS_VERSION = "3.0.5" + DEFAULT_PBO_COMMON = { + "sdk": "2025.2.2.0", + "platform": "2025.2.2.0", + "application": ( + "12.1.6-23045 R 2025.2 android-30-JPLAYER2 " + "ninja_6==NVIDIA/mdarcy/mdarcy:11/RQ1A.210105.003/7825230_4040.2147:user/release-keys" + ), + "uiversion": "UI-release-20260303_43809-gibbon-r100-darwinql-69067=5,78214=2,78929=8,80045=1,80048=2", + "uiPlatform": "tv_ui", + "clientVersion": "v3.11.512", + "apkVersion": "12.1.6", + } + DEFAULT_PBO_LANGUAGES = ["en-CA", "en-US", "en"] + WRAP_SALT = bytes.fromhex("027617984f6227539a630b897c017d69") + WRAP_INFO = bytes.fromhex("809f82a7addf548d3ea9dd067ff9bb91") + DH_PRIME = bytes( + [ + 0x96, + 0x94, + 0xE9, + 0xD8, + 0xD9, + 0x3A, + 0x5A, + 0xC7, + 0x4C, + 0x50, + 0x9B, + 0x4B, + 0xBC, + 0xE8, + 0x5E, + 0x92, + 0x13, + 0x2C, + 0xD1, + 0x9C, + 0xCE, + 0x47, + 0x7D, + 0x1A, + 0x7E, + 0x47, + 0xD5, + 0x27, + 0xD9, + 0xEC, + 0x29, + 0x15, + 0x15, + 0xF0, + 0xB8, + 0xB3, + 0xE1, + 0xEA, + 0xED, + 0x50, + 0x06, + 0xE1, + 0xB1, + 0xB9, + 0x1E, + 0xA2, + 0x5B, + 0x91, + 0xA0, + 0x1B, + 0x10, + 0xE2, + 0xE8, + 0x34, + 0xB8, + 0xD6, + 0x60, + 0xB2, + 0xE3, + 0x21, + 0xAD, + 0x64, + 0x4C, + 0xE1, + 0xA8, + 0x3B, + 0x32, + 0x8D, + 0x90, + 0x14, + 0xEE, + 0x7E, + 0x16, + 0xF1, + 0xE4, + 0x4F, + 0xFE, + 0x89, + 0x57, + 0x9A, + 0xC3, + 0xEE, + 0x47, + 0xD6, + 0x68, + 0xB6, + 0xB7, + 0x66, + 0x87, + 0xC2, + 0xFE, + 0x90, + 0xA3, + 0x5B, + 0x5E, + 0x60, + 0x28, + 0xFD, + 0x04, + 0xEF, + 0xEA, + 0x88, + 0x23, + 0x73, + 0xEC, + 0xF6, + 0x0B, + 0xA2, + 0xF6, + 0x37, + 0xE4, + 0xCD, + 0xAA, + 0x1B, + 0x60, + 0x89, + 0xD6, + 0xC0, + 0xB5, + 0x61, + 0xA8, + 0xE5, + 0x20, + 0xE7, + 0x96, + 0xDE, + 0x27, + 0xDF, + ] + ) + DH_P = int.from_bytes(DH_PRIME, "big") + DH_G = 5 + + def __init__( + self, + session: requests.Session, + sender: str, + keys: MSLKeys, + message_id: int, + user_auth: Optional[dict] = None, + cookies: Optional[Dict[str, str]] = None, + ) -> None: + self.session = session + self.sender = sender + self.keys = keys + self.message_id = message_id + self.user_auth = user_auth + self.cookies = cookies + + @staticmethod + def build_request_headers( + request_name: str, + user_agent: Optional[str] = None, + referer: Optional[str] = None, + viewable_id: Optional[int] = None, + profile_guid: Optional[str] = None, + esn: Optional[str] = None, + expiry_timeout: Optional[int] = 12750, + extra_headers: Optional[Dict[str, str]] = None, + ) -> Dict[str, str]: + headers: Dict[str, str] = { + "User-Agent": user_agent or MSL.DEFAULT_USER_AGENT, + "Accept": "*/*", + "Content-Type": "application/json", + "X-Netflix.Client.Request.Name": request_name, + "X-Netflix.request.attempt": "1", + "X-Netflix.Request.NonJson.Headers": "true", + "X-Netflix.Request.Client.Context": MSL.DEFAULT_REQUEST_CONTEXT, + "x-netflix.client.nrdjs.version": MSL.DEFAULT_NRDJS_VERSION, + "x-netflix.client.netjs.version": MSL.DEFAULT_NETJS_VERSION, + "x-netflix.client.last-interacted-days": "0", + } + + if expiry_timeout is not None: + headers["x-netflix.request.expiry.timeout"] = str(expiry_timeout) + + if referer: + headers["Referer"] = referer + + if viewable_id is not None: + headers["x-netflix.playback.main-content-viewable-id"] = str(viewable_id) + + if profile_guid: + headers["x-netflix.client.current-profile-guid"] = profile_guid + + if esn: + headers["x-netflix.client.ftl.esn"] = esn + + if extra_headers: + headers.update(extra_headers) + + return headers + + @staticmethod + def manifest_request_defaults() -> Tuple[str, Dict[str, str]]: + return MSL.DEFAULT_MANIFEST_ENDPOINT, dict(MSL.DEFAULT_MANIFEST_PARAMS) + + @staticmethod + def stable_json(obj: Dict[str, Any]) -> str: + return json.dumps(obj, separators=(",", ":"), ensure_ascii=False) + + @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 + + @staticmethod + def b64_encode_bytes(value: bytes) -> str: + return base64.b64encode(value).decode("ascii") + + @staticmethod + def b64_decode_strict(value: str) -> bytes: + normalized_value = value.strip().strip('"').strip("'") + return base64.b64decode(normalized_value.encode("ascii"), validate=True) + + @classmethod + def find_sidecar_file(cls, filename: str, env_name: str) -> Optional[Path]: + env_value = os.getenv(env_name) + candidates: List[Path] = [] + + if env_value: + candidates.append(Path(env_value)) + + roots = [Path.cwd(), Path(__file__).resolve().parent] + + for root in roots: + candidates.append(root / filename) + candidates.extend(root.glob(f"**/{filename}")) + + seen = set() + + for candidate in candidates: + key = str(candidate.resolve()) if candidate.exists() else str(candidate) + + if key in seen: + continue + + seen.add(key) + + if candidate.is_file(): + return candidate + + return None + + @staticmethod + def load_esnid_file(path: Path) -> str: + value = path.read_text(encoding="utf-8", errors="ignore").strip() + + if not value: + raise ValueError(f"Empty ESNID file: {path}") + + return value + + @classmethod + def load_kpe_kph_file(cls, path: Path) -> Tuple[bytes, bytes, bytes]: + raw = path.read_bytes() + + if raw.startswith(b"\xef\xbb\xbf"): + raw = raw[3:] + + text = raw.decode("utf-8", errors="strict").strip() + text = re.sub(r"\s*,\s*", ",", text) + left, right = text.split(",", 1) + enc_key = cls.b64_decode_strict(left) + hmac_key = cls.b64_decode_strict(right) + + if len(enc_key) != 16: + raise ValueError(f"Kpe must decode to 16 bytes, got {len(enc_key)}") + + if len(hmac_key) != 32: + raise ValueError(f"Kph must decode to 32 bytes, got {len(hmac_key)}") + + wrap_key = cls.derive_wrapping_key(enc_key, hmac_key) + return enc_key, hmac_key, wrap_key + + @classmethod + def load_cache_data(cls, msl_keys_path: Optional[Path] = None) -> Optional[MSLKeys]: + if not msl_keys_path or not msl_keys_path.is_file(): + return None + + loaded_keys = jsonpickle.decode(msl_keys_path.read_text(encoding="utf-8")) + + if not isinstance(loaded_keys, MSLKeys): + return None + + if loaded_keys.mastertoken: + expiry_value = json.loads( + base64.b64decode(loaded_keys.mastertoken["tokendata"]).decode("utf-8") + ).get("expiration") + + if expiry_value is not None: + expiry = datetime.fromtimestamp(int(expiry_value), tz=timezone.utc) + hours_remaining = (expiry - datetime.now(timezone.utc)).total_seconds() / 3600 + + if hours_remaining < 10: + return None + + if not hasattr(loaded_keys, "wrapdata"): + loaded_keys.wrapdata = None + + if not hasattr(loaded_keys, "derivation_key"): + loaded_keys.derivation_key = None + + return loaded_keys + + @staticmethod + def cache_keys(msl_keys: MSLKeys, msl_keys_path: Path) -> None: + msl_keys_path.parent.mkdir(parents=True, exist_ok=True) + msl_keys_path.write_text(jsonpickle.encode(msl_keys, indent=4), encoding="utf-8") + + @classmethod + def derive_wrapping_key(cls, encryption_key_16: bytes, hmac_key_32: bytes) -> bytes: + if len(encryption_key_16) != 16: + raise ValueError("encryptionKey must be 16 bytes") + + if len(hmac_key_32) != 32: + raise ValueError("hmacKey must be 32 bytes") + + inner = HMAC.new(cls.WRAP_SALT, digestmod=SHA256) + inner.update(encryption_key_16 + hmac_key_32) + + outer = HMAC.new(inner.digest(), digestmod=SHA256) + outer.update(cls.WRAP_INFO) + + return outer.digest()[:16] + + @staticmethod + def int_to_unsigned_bytes(value: int) -> bytes: + if value < 0: + raise ValueError("value must be non-negative") + + if value == 0: + return b"\x00" + + return value.to_bytes((value.bit_length() + 7) // 8, "big", signed=False) + + @staticmethod + def correct_null_bytes(value: bytes) -> bytes: + count = 0 + + for byte in value: + if byte == 0: + count += 1 + else: + break + + if count == 1: + return value + + trimmed = value[count:] + return b"\x00" + trimmed + + @classmethod + def dh_generate_keypair(cls) -> Tuple[int, bytes]: + private_key = int.from_bytes(os.urandom(len(cls.DH_PRIME)), "big") % (cls.DH_P - 3) + 2 + public_key = pow(cls.DH_G, private_key, cls.DH_P) + public_key_bytes = cls.int_to_unsigned_bytes(public_key) + + return private_key, cls.correct_null_bytes(public_key_bytes) + + @classmethod + def dh_compute_shared_secret_bytes(cls, dh_private_key: int, server_public_key_wire: bytes) -> bytes: + normalized_server_public_key = cls.correct_null_bytes(server_public_key_wire) + server_public_key_raw = ( + normalized_server_public_key[1:] + if normalized_server_public_key[:1] == b"\x00" + else normalized_server_public_key + ) + server_public_key = int.from_bytes(server_public_key_raw, "big", signed=False) + shared_secret = pow(server_public_key, dh_private_key, cls.DH_P) + + return cls.correct_null_bytes(cls.int_to_unsigned_bytes(shared_secret)) + + @classmethod + def kdf_authenticated_dh(cls, derivation_key: bytes, shared_secret_bytes: bytes) -> Tuple[bytes, bytes, bytes]: + if derivation_key is None: + raise ValueError("derivation key is required for AUTHENTICATED_DH") + + salt_key = SHA384.new(derivation_key).digest() + hmac_value = HMAC.new(salt_key, digestmod=SHA384) + hmac_value.update(shared_secret_bytes) + raw_key_material = hmac_value.digest() + + encryption_key = raw_key_material[:16] + hmac_key = raw_key_material[16:48] + wrapping_key = cls.derive_wrapping_key(encryption_key, hmac_key) + + return encryption_key, hmac_key, wrapping_key + + @staticmethod + def msl_encrypt_v1(key_id: str, encryption_key_16: bytes, plaintext_bytes: bytes) -> bytes: + iv = get_random_bytes(16) + ciphertext = AES.new(encryption_key_16, AES.MODE_CBC, iv).encrypt( + pad(plaintext_bytes, AES.block_size) + ) + envelope = { + "keyid": key_id, + "iv": base64.b64encode(iv).decode("ascii"), + "ciphertext": base64.b64encode(ciphertext).decode("ascii"), + "sha256": "AA==", + } + + return json.dumps(envelope, separators=(",", ":")).encode("utf-8") + + @staticmethod + def msl_decrypt_v1(encryption_key_16: bytes, envelope_bytes: bytes) -> bytes: + envelope = json.loads(envelope_bytes.decode("utf-8")) + iv = base64.b64decode(envelope["iv"]) + ciphertext = base64.b64decode(envelope["ciphertext"]) + padded_plaintext = AES.new(encryption_key_16, AES.MODE_CBC, iv).decrypt(ciphertext) + + return unpad(padded_plaintext, AES.block_size) + + @staticmethod + def msl_sign_b64(hmac_key_32: bytes, data_bytes: bytes) -> str: + signer = HMAC.new(hmac_key_32, digestmod=SHA256) + signer.update(data_bytes) + + return base64.b64encode(signer.digest()).decode("ascii") + + @staticmethod + def msl_verify_sig(hmac_key_32: bytes, data_bytes: bytes, signature_b64: str) -> None: + signer = HMAC.new(hmac_key_32, digestmod=SHA256) + signer.update(data_bytes) + expected_signature = signer.digest() + received_signature = base64.b64decode(signature_b64) + + if expected_signature != received_signature: + raise ValueError("Response signature verification failed: HMAC mismatch") + + @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"], + "encoderformats": ["JSON"], + }, + "timestamp": int(datetime.now(timezone.utc).timestamp()), + "sender": sender, + "nonreplayable": False, + } + + if userauthdata: + header_data["userauthdata"] = userauthdata + + if keyrequestdata: + header_data["keyrequestdata"] = [keyrequestdata] + + return json.dumps(header_data, separators=(",", ":")) + + @classmethod + def handshake( + cls, + session: requests.Session, + sender: str, + kpekph_path: Optional[Union[str, Path]] = None, + msl_keys_path: Optional[Union[str, Path]] = None, + cookies: Optional[Dict[str, str]] = None, + headers: Optional[Dict[str, str]] = None, + timeout: int = 30, + new_msl: bool = False, + ) -> "MSL": + endpoint = "https://www.netflix.com/msl/playapi/cadmium/licensedmanifest/1" + message_id = random.randint(0, pow(2, 52)) + + if cookies: + session.cookies.update(cookies) + + cache_path = Path(msl_keys_path) if msl_keys_path else None + cached_keys = None if new_msl else cls.load_cache_data(cache_path) + + if cached_keys is not None: + return cls( + session=session, + sender=sender, + keys=cached_keys, + message_id=message_id, + cookies=cookies, + ) + + if not sender: + raise RuntimeError("Missing sender or ESNID for MGK handshake") + + if kpekph_path: + resolved_kpekph_path = Path(kpekph_path) + else: + resolved_kpekph_path = cls.find_sidecar_file("KpeKph", "MSL_KPEKPH_PATH") + + if not resolved_kpekph_path: + raise RuntimeError( + "KpeKph was not found. Place KpeKph next to the client, " + "inside a child folder, in the current working directory, or set MSL_KPEKPH_PATH." + ) + + entity_encryption_key, entity_hmac_key, entity_wrapping_key = cls.load_kpe_kph_file( + resolved_kpekph_path + ) + + msl_keys = MSLKeys() + cached_wrapdata = cached_keys.wrapdata if cached_keys else None + cached_derivation_key = cached_keys.derivation_key if cached_keys else None + mechanism = "WRAP" if cached_wrapdata and cached_derivation_key else "MGK" + derivation_key = cached_derivation_key or entity_wrapping_key + + dh_private_key, dh_public_key_wire = cls.dh_generate_keypair() + key_data: Dict[str, Any] = { + "mechanism": mechanism, + "publickey": cls.b64_encode_bytes(cls.correct_null_bytes(dh_public_key_wire)), + "parametersid": "1", + } + + if mechanism == "WRAP": + key_data["wrapdata"] = cls.b64_encode_bytes(cached_wrapdata) + + key_request_data = { + "scheme": "AUTHENTICATED_DH", + "keydata": key_data, + } + entity_auth_data = EntityAuthentication.ModelGroup(sender).__dict__ + + header_plaintext = cls.generate_msg_header( + message_id=message_id, + sender=sender, + is_handshake=True, + keyrequestdata=key_request_data, + compression="GZIP", + ).encode("utf-8") + header_ciphertext = cls.msl_encrypt_v1(sender, entity_encryption_key, header_plaintext) + + payload_plaintext = cls.stable_json( + { + "messageid": message_id, + "data": "", + "sequencenumber": 1, + "endofmsg": True, + } + ).encode("utf-8") + payload_ciphertext = cls.msl_encrypt_v1(sender, entity_encryption_key, payload_plaintext) + + request_body = cls.stable_json( + { + "entityauthdata": entity_auth_data, + "headerdata": cls.b64_encode_bytes(header_ciphertext), + "signature": cls.msl_sign_b64(entity_hmac_key, header_ciphertext), + } + ) + request_body += cls.stable_json( + { + "payload": cls.b64_encode_bytes(payload_ciphertext), + "signature": cls.msl_sign_b64(entity_hmac_key, payload_ciphertext), + } + ) + + response = session.post( + url=endpoint, + data=request_body, + headers=headers or {}, + timeout=timeout, + ) + + if response.status_code != 200: + raise RuntimeError( + f"Key exchange failed: HTTP {response.status_code} {response.text[:500]}" + ) + + parsed_response = cls.parse_concatenated_json(response.text) + + if not parsed_response: + raise RuntimeError("Key exchange failed: empty MSL response") + + key_exchange = parsed_response[0] + + if "errordata" in key_exchange: + decoded_error = base64.b64decode(key_exchange["errordata"]).decode("utf-8", "ignore") + raise RuntimeError(f"Key exchange failed: {decoded_error}") + + if "headerdata" not in key_exchange: + raise RuntimeError( + f"Key exchange failed: missing headerdata in response: {str(key_exchange)[:500]}" + ) + + header_json = json.loads(base64.b64decode(key_exchange["headerdata"]).decode("utf-8")) + key_response_data = header_json["keyresponsedata"] + response_key_data = key_response_data["keydata"] + server_public_key_wire = cls.correct_null_bytes( + base64.b64decode(response_key_data["publickey"]) + ) + response_wrapdata = response_key_data.get("wrapdata") + + if response_wrapdata: + msl_keys.wrapdata = base64.b64decode(response_wrapdata) + else: + msl_keys.wrapdata = cached_wrapdata + + shared_secret = cls.dh_compute_shared_secret_bytes( + dh_private_key, + server_public_key_wire, + ) + encryption_key, sign_key, next_derivation_key = cls.kdf_authenticated_dh( + derivation_key, + shared_secret, + ) + + msl_keys.encryption = encryption_key + msl_keys.sign = sign_key + msl_keys.derivation_key = next_derivation_key + msl_keys.mastertoken = key_response_data["mastertoken"] + + if cache_path: + cls.cache_keys(msl_keys, cache_path) + + return cls( + session=session, + sender=sender, + keys=msl_keys, + message_id=message_id, + cookies=cookies, + ) + + def send_message( + self, + endpoint: Optional[str] = None, + params: Optional[Dict[str, str]] = None, + application_data: Optional[Dict[str, Any]] = None, + userauthdata: Optional[dict] = None, + headers: Optional[Dict[str, str]] = None, + timeout: int = 30, + ) -> Tuple[Dict[str, Any], Any]: + message = self.create_message(application_data or {}, userauthdata) + response = self.session.post( + url=endpoint or self.DEFAULT_MANIFEST_ENDPOINT, + data=message, + params=params or {}, + headers=headers or {}, + cookies=self.cookies, + timeout=timeout, + ) + + if response.status_code != 200: + body_preview = response.text[:500] if response.text else response.content[:200].hex() + raise RuntimeError(f"MSL request failed: HTTP {response.status_code} {body_preview}") + + header, payload_data = self.parse_message(response) + + if "errordata" in header: + decoded_error = json.loads( + base64.b64decode(header["errordata"].encode("utf-8")).decode("utf-8") + ) + raise RuntimeError(f"MSL response contains an error: {decoded_error}") + + return header, payload_data + + def create_message( + self, + application_data: Dict[str, Any], + userauthdata: Optional[dict] = None, + ) -> str: + self.message_id += 1 + + header_data = self.encrypt( + self.generate_msg_header( + message_id=self.message_id, + sender=self.sender, + is_handshake=False, + userauthdata=userauthdata, + compression="GZIP", + ) + ) + message = json.dumps( + { + "headerdata": base64.b64encode(header_data.encode("utf-8")).decode("utf-8"), + "signature": self.sign(header_data).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_chunk = self.encrypt( + json.dumps( + { + "messageid": self.message_id, + "data": compressed_application_data, + "compressionalgo": "GZIP", + "sequencenumber": 1, + "endofmsg": True, + }, + separators=(",", ":"), + ) + ) + message += json.dumps( + { + "payload": base64.b64encode(payload_chunk.encode("utf-8")).decode("utf-8"), + "signature": self.sign(payload_chunk).decode("utf-8"), + }, + separators=(",", ":"), + ) + + return message + + def parse_message(self, response: Any) -> Tuple[Dict[str, Any], Any]: + if hasattr(response, "text"): + message = response.text or "" + raw_content = response.content + status_code = getattr(response, "status_code", None) + content_type = ( + response.headers.get("Content-Type", "") + if getattr(response, "headers", None) + else "" + ) + else: + message = str(response or "") + raw_content = message.encode("utf-8", errors="ignore") + status_code = None + content_type = "" + + parsed_message = self.parse_concatenated_json(message) + + if not parsed_message: + preview = raw_content[:200] + + try: + preview_text = preview.decode("utf-8", errors="replace") + except Exception: + preview_text = repr(preview) + + raise RuntimeError( + "MSL response was empty or not concatenated JSON. " + f"status={status_code} content_type={content_type!r} body_preview={preview_text!r}" + ) + + 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 + + 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: + payload_chunk_json = json.loads( + base64.b64decode(payload_chunk["payload"]).decode("utf-8") + ) + payload_decrypted = self.aes_cbc_decrypt( + self.keys.encryption, + base64.b64decode(payload_chunk_json["iv"]), + base64.b64decode(payload_chunk_json["ciphertext"]), + ) + payload_decrypted_json = json.loads(payload_decrypted.decode("utf-8")) + payload_data = base64.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") + + data = json.loads(raw_data) + + if "error" in data: + raise RuntimeError(data["error"]) + + if "result" not in data: + return data + + return data["result"] + + @staticmethod + def gzip_compress(data: bytes) -> bytes: + output = BytesIO() + + with gzip.GzipFile(fileobj=output, mode="w") as gzip_file: + gzip_file.write(data) + + return base64.b64encode(output.getvalue()) + + 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 = os.urandom(16) + token_data = json.loads(base64.b64decode(self.keys.mastertoken["tokendata"]).decode("utf-8")) + ciphertext = self.aes_cbc_encrypt( + self.keys.encryption, + iv, + plaintext.encode("utf-8"), + ) + + return json.dumps( + { + "ciphertext": base64.b64encode(ciphertext).decode("utf-8"), + "keyid": f"{self.sender}_{token_data['sequencenumber']}", + "sha256": "AA==", + "iv": base64.b64encode(iv).decode("utf-8"), + }, + separators=(",", ":"), + ) + + def sign(self, text: str) -> bytes: + if not self.keys.sign: + raise ValueError("Sign key is not available") + + signer = HMAC.new(self.keys.sign, digestmod=SHA256) + signer.update(text.encode("utf-8")) + + return base64.b64encode(signer.digest()) + + @staticmethod + def aes_cbc_encrypt(key: bytes, iv: bytes, plaintext: bytes) -> bytes: + return AES.new(key, AES.MODE_CBC, iv).encrypt(pad(plaintext, AES.block_size)) + + @staticmethod + def aes_cbc_decrypt(key: bytes, iv: bytes, ciphertext: bytes) -> bytes: + padded_plaintext = AES.new(key, AES.MODE_CBC, iv).decrypt(ciphertext) + return unpad(padded_plaintext, AES.block_size) + + +__all__ = [ + "EntityAuthentication", + "EntityAuthenticationSchemes", + "MSL", + "MSLKeys", + "MSLObject", + "Scheme", + "UserAuthentication", + "UserAuthenticationSchemes", +]