diff --git a/wpgskd/utils/AtomicSQL.py b/wpgskd/utils/AtomicSQL.py deleted file mode 100644 index d18ec40..0000000 --- a/wpgskd/utils/AtomicSQL.py +++ /dev/null @@ -1,52 +0,0 @@ -import os -import sqlite3 -import time -from threading import Lock - -class AtomicSQL: - """Race-condition and Threading safe SQL Database Interface.""" - def __init__(self): - self.master_lock = Lock() - self.db = {} - self.cursor = {} - self.session_lock = {} - - def load(self, connection: sqlite3.Connection): - self.master_lock.acquire() - try: - session_id = None - while not session_id or session_id in self.db: - session_id = os.urandom(16) - self.db[session_id] = connection - self.cursor[session_id] = self.db[session_id].cursor() - self.session_lock[session_id] = Lock() - return session_id - finally: - self.master_lock.release() - - def safe_execute(self, session_id, action): - if session_id not in self.db: - raise ValueError(f"Session ID {session_id!r} is invalid.") - self.master_lock.acquire() - self.session_lock[session_id].acquire() - try: - failures = 0 - while True: - try: - action(db=self.db[session_id], cursor=self.cursor[session_id]) - break - except sqlite3.OperationalError: - failures += 1 - delay = 3 * failures - print(f"AtomicSQL.safe_execute failed, retrying in {delay} seconds...") - time.sleep(delay) - if failures == 10: - raise ValueError("AtomicSQL.safe_execute failed too many time's. Aborting.") - return self.cursor[session_id] - finally: - self.session_lock[session_id].release() - self.master_lock.release() - - def commit(self, session_id): - self.safe_execute(session_id, lambda db, cursor: db.commit()) - return True \ No newline at end of file diff --git a/wpgskd/utils/BamSDK/__init__.py b/wpgskd/utils/BamSDK/__init__.py deleted file mode 100644 index b706e37..0000000 --- a/wpgskd/utils/BamSDK/__init__.py +++ /dev/null @@ -1,2 +0,0 @@ -# flake8: noqa -from wpgskd.utils.BamSDK.bamsdk import BamSdk diff --git a/wpgskd/utils/BamSDK/bamsdk.py b/wpgskd/utils/BamSDK/bamsdk.py deleted file mode 100644 index d9ed289..0000000 --- a/wpgskd/utils/BamSDK/bamsdk.py +++ /dev/null @@ -1,28 +0,0 @@ -import requests - -from wpgskd.utils.BamSDK.services.account import account -from wpgskd.utils.BamSDK.services.bamIdentity import bamIdentity -from wpgskd.utils.BamSDK.services.content import content -from wpgskd.utils.BamSDK.services.device import device -from wpgskd.utils.BamSDK.services.drm import drm -from wpgskd.utils.BamSDK.services.media import media -from wpgskd.utils.BamSDK.services.session import session -from wpgskd.utils.BamSDK.services.token import token - - -class BamSdk: - def __init__(self, endpoint, session_=None): - self._session = session_ or requests.Session() - - self.config = self._session.get(endpoint).json() - self.application = self.config["application"] - self.commonHeaders = self.config["commonHeaders"] - - self.account = account(self.config["services"]["account"], self._session) - self.bamIdentity = bamIdentity(self.config["services"]["bamIdentity"], self._session) - self.content = content(self.config["services"]["content"], self._session) - self.device = device(self.config["services"]["device"], self._session) - self.drm = drm(self.config["services"]["drm"], self._session) - self.media = media(self.config["services"]["media"], self._session) - self.session = session(self.config["services"]["session"], self._session) - self.token = token(self.config["services"]["token"], self._session) diff --git a/wpgskd/utils/BamSDK/services/__init__.py b/wpgskd/utils/BamSDK/services/__init__.py deleted file mode 100644 index 0bdd32f..0000000 --- a/wpgskd/utils/BamSDK/services/__init__.py +++ /dev/null @@ -1,37 +0,0 @@ -import requests - - -class Service: - def __init__(self, cfg, session=None): - self.session = session or requests.Session() - self.client = Client(cfg.get("client") or {}) - self.disabled = cfg.get("disabled") - self.extras = cfg.get("extras") - - -class Client: - def __init__(self, data): - self.baseUrl = data.get("baseUrl") - self.endpoints = {k: Endpoint(v) for k, v in (data.get("endpoints") or {}).items()} - self.extras = data.get("extras") or {} - - -class Endpoint: - def __init__(self, data): - self.headers = data.get("headers") or {} - self.href = data["href"] - self.method = data.get("method") or "GET" - self.templated = data.get("templated") or False - self.timeout = data.get("timeout") or 15 - self.ttl = data.get("ttl") or 0 - - # noinspection PyPep8Naming - def get_headers(self, accessToken=None, apiKey=None): - token = None - if accessToken: - token = {"accessToken": accessToken} - elif apiKey: - token = {"apiKey": apiKey} - if token: - self.headers.update({"Authorization": self.headers["Authorization"].format(**token)}) - return self.headers diff --git a/wpgskd/utils/BamSDK/services/account.py b/wpgskd/utils/BamSDK/services/account.py deleted file mode 100644 index 5befe23..0000000 --- a/wpgskd/utils/BamSDK/services/account.py +++ /dev/null @@ -1,30 +0,0 @@ -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class account(Service): - def createAccountGrant(self, json: dict, access_token: str) -> dict: - endpoint = self.client.endpoints["createAccountGrant"] - return self.session.request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token), - json=json, - ).json() - - def getUserProfiles(self, access_token: str) -> dict: - endpoint = self.client.endpoints["getCurrentAccount"] - return self.session.request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token), - ).json() - - def setActiveUserProfile(self, profile_id: str, access_token: str) -> dict: - # Hardcoded since its not in v4 anymore - return self.session.put( - url=f"https://disney.api.edge.bamgrid.com/accounts/me/active-profile/{profile_id}", - headers={"Authorization": f"Bearer {access_token}"}, - ).json() diff --git a/wpgskd/utils/BamSDK/services/bamIdentity.py b/wpgskd/utils/BamSDK/services/bamIdentity.py deleted file mode 100644 index 64a236a..0000000 --- a/wpgskd/utils/BamSDK/services/bamIdentity.py +++ /dev/null @@ -1,20 +0,0 @@ -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class bamIdentity(Service): - def identityLogin(self, email, password, access_token): - endpoint = self.client.endpoints["identityLogin"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token), - json={ - "email": email, - "password": password - } - ).prepare() - res = self.session.send(req) - return res.json() diff --git a/wpgskd/utils/BamSDK/services/content.py b/wpgskd/utils/BamSDK/services/content.py deleted file mode 100644 index 2764f21..0000000 --- a/wpgskd/utils/BamSDK/services/content.py +++ /dev/null @@ -1,36 +0,0 @@ -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class content(Service): - def getDmcEpisodes(self, region, season_id, page, access_token): - endpoint = self.client.endpoints["getDmcEpisodes"] - req = Request( - method=endpoint.method, - url=f"https://disney.content.edge.bamgrid.com/svc/content/DmcEpisodes/version/5.1/region/{region}/audience/false/maturity/1850/language/en/seasonId/{season_id}/pageSize/15/page/{page}", - headers=endpoint.get_headers(accessToken=access_token) - ).prepare() - res = self.session.send(req) - return res.json() - - def getDmcSeriesBundle(self, region, media_id, access_token): - endpoint = self.client.endpoints["getDmcSeriesBundle"] - req = Request( - method=endpoint.method, - url=f"https://disney.content.edge.bamgrid.com/svc/content/DmcSeriesBundle/version/5.1/region/{region}/audience/false/maturity/1850/language/en/encodedSeriesId//{media_id}", - headers=endpoint.get_headers(accessToken=access_token) - ).prepare() - res = self.session.send(req) - return res.json() - - def getDmcVideoBundle(self, region, media_id, access_token): - endpoint = self.client.endpoints["getDmcVideoBundle"] - req = Request( - method=endpoint.method, - url=f"https://disney.content.edge.bamgrid.com/svc/content/DmcVideoBundle/version/5.1/region/{region}/audience/false/maturity/1850/language/en/encodedFamilyId/{media_id}", - headers=endpoint.get_headers(accessToken=access_token) - ).prepare() - res = self.session.send(req) - return res.json() diff --git a/wpgskd/utils/BamSDK/services/device.py b/wpgskd/utils/BamSDK/services/device.py deleted file mode 100644 index 4ce9aa0..0000000 --- a/wpgskd/utils/BamSDK/services/device.py +++ /dev/null @@ -1,24 +0,0 @@ -from json import JSONDecodeError - -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class device(Service): - - def createDeviceGrant(self, json, api_key): - endpoint = self.client.endpoints["createDeviceGrant"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(apiKey=api_key), - json=json - ).prepare() - res = self.session.send(req) - try: - data = res.json() - except JSONDecodeError: - raise Exception(f"An unexpected response occurred for bamsdk.createDeviceGrant: {res.text}") - return data diff --git a/wpgskd/utils/BamSDK/services/drm.py b/wpgskd/utils/BamSDK/services/drm.py deleted file mode 100644 index fe87b54..0000000 --- a/wpgskd/utils/BamSDK/services/drm.py +++ /dev/null @@ -1,50 +0,0 @@ -import json - -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class drm(Service): - def widevineCertificate(self): - endpoint = self.client.endpoints["widevineCertificate"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.headers - ).prepare() - res = self.session.send(req) - return res.content - - def widevineLicense(self, licence, access_token): - endpoint = self.client.endpoints["widevineLicense"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token), - data=licence - ).prepare() - res = self.session.send(req) - try: - # if it's json content, then an error occurred - res = json.loads(res.text) - raise Exception(f"Failed to obtain license: {res}") - except json.JSONDecodeError: - return res.content - - def playreadyLicense(self, licence, access_token): - endpoint = self.client.endpoints["playReadyLicense"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token), - data=licence - ).prepare() - res = self.session.send(req) - try: - # if it's json content, then an error occurred - res = json.loads(res.text) - raise Exception(f"Failed to obtain license: {res}") - except json.JSONDecodeError: - return res.content diff --git a/wpgskd/utils/BamSDK/services/media.py b/wpgskd/utils/BamSDK/services/media.py deleted file mode 100644 index 797367e..0000000 --- a/wpgskd/utils/BamSDK/services/media.py +++ /dev/null @@ -1,24 +0,0 @@ -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class media(Service): - def __init__(self, cfg, session=None): - super().__init__(cfg, session) - self.uhd_allowed = self.extras["isUhdAllowed"] - self.default_scenario = self.extras["playbackScenarioDefault"] - self.scenarios = self.extras["playbackScenarios"] - self.restricted_scenario = self.extras["restrictedPlaybackScenario"] - self.security_requirements = self.extras["securityCheckRequirements"] - - def mediaPayload(self, media_id, scenario, access_token): - endpoint = self.client.endpoints["mediaPayload"] - req = Request( - method=endpoint.method, - url=f"{self.client.baseUrl}/media/{media_id}/scenarios/{scenario}", - headers=endpoint.get_headers(accessToken=access_token) - ).prepare() - res = self.session.send(req) - return res.json() diff --git a/wpgskd/utils/BamSDK/services/session.py b/wpgskd/utils/BamSDK/services/session.py deleted file mode 100644 index 7b58953..0000000 --- a/wpgskd/utils/BamSDK/services/session.py +++ /dev/null @@ -1,26 +0,0 @@ -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class session(Service): - def getInfo(self, access_token): - endpoint = self.client.endpoints["getInfo"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token) - ).prepare() - res = self.session.send(req) - return res.json() - - def getLocation(self, access_token): - endpoint = self.client.endpoints["getLocation"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(accessToken=access_token) - ).prepare() - res = self.session.send(req) - return res.json() diff --git a/wpgskd/utils/BamSDK/services/token.py b/wpgskd/utils/BamSDK/services/token.py deleted file mode 100644 index 3eaaa37..0000000 --- a/wpgskd/utils/BamSDK/services/token.py +++ /dev/null @@ -1,21 +0,0 @@ -from requests import Request - -from wpgskd.utils.BamSDK.services import Service - - -# noinspection PyPep8Naming -class token(Service): - def __init__(self, cfg, session=None): - super().__init__(cfg, session) - self.subject_tokens = self.extras["subjectTokenTypes"] - - def exchange(self, data, api_key): - endpoint = self.client.endpoints["exchange"] - req = Request( - method=endpoint.method, - url=endpoint.href, - headers=endpoint.get_headers(apiKey=api_key), - data=data - ).prepare() - res = self.session.send(req) - return res.json() diff --git a/wpgskd/utils/Logger.py b/wpgskd/utils/Logger.py deleted file mode 100644 index c74366c..0000000 --- a/wpgskd/utils/Logger.py +++ /dev/null @@ -1,67 +0,0 @@ -from __future__ import annotations - -import logging -import sys -from pathlib import Path -from typing import Any, IO, NoReturn, Optional, Union - -import coloredlogs - -LOG_FORMAT = "{asctime} [{levelname[0]}] {name} : {message}" -LOG_DATE_FORMAT = "%Y-%m-%d %H:%M:%S" -LOG_STYLE = "{" -LOG_FORMATTER = logging.Formatter(LOG_FORMAT, LOG_DATE_FORMAT, LOG_STYLE) - - -class Logger(logging.Logger): - def __init__(self, name: str = "root", level: int = logging.NOTSET, color: bool = True): - """Initialize the logger with a name and an optional level.""" - super().__init__(name, level) - if self.name == "root": - self.add_stream_handler() - if color: - self.install_color() - - def exit(self, msg: str, *args: Any, code: int = 1, **kwargs: Any) -> NoReturn: - """ - Log 'msg % args' with severity 'CRITICAL' and terminate the program - with a default exit code of 1. - - To pass exception information, use the keyword argument exc_info with - a true value, e.g. - - logger.exit("Houston, we have a %s", "major disaster", exc_info=1) - """ - self.critical(msg, *args, **kwargs) - sys.exit(code) - - def add_stream_handler(self, stream: Optional[IO[str]] = None) -> None: - """Add a stream handler to log. Stream defaults to stdout.""" - sh = logging.StreamHandler(stream) - sh.setFormatter(LOG_FORMATTER) - self.addHandler(sh) - - def add_file_handler(self, fp: Union[IO, Path, str]) -> None: - """Convenience alias func for add_stream_handler, deals with type of fp object input.""" - if not isinstance(fp, IO): - fp = Path(fp) - fp = fp.open("w", encoding="utf8") - self.add_stream_handler(fp) - - def install_color(self) -> None: - """Use coloredlogs to set up colors on the log output.""" - if self.level == logging.DEBUG: - coloredlogs.install(level=self.level, fmt=LOG_FORMAT, datefmt=LOG_DATE_FORMAT, style=LOG_STYLE) - coloredlogs.install(level=self.level, logger=self, fmt=LOG_FORMAT, datefmt=LOG_DATE_FORMAT, style=LOG_STYLE) - - -# Cache already used loggers to make sure their level is preserved -_loggers: dict[str, Logger] = {} - - -# noinspection PyPep8Naming -def getLogger(name: Optional[str] = None, level: int = logging.NOTSET) -> Logger: - name = name or "root" - _log = _loggers.get(name, Logger(name)) - _log.setLevel(level) - return _log diff --git a/wpgskd/utils/MSL/MSLKeys.py b/wpgskd/utils/MSL/MSLKeys.py deleted file mode 100644 index bba1b43..0000000 --- a/wpgskd/utils/MSL/MSLKeys.py +++ /dev/null @@ -1,10 +0,0 @@ -from wpgskd.utils.MSL.MSLObject import MSLObject - - -class MSLKeys(MSLObject): - def __init__(self, encryption=None, sign=None, rsa=None, mastertoken=None, cdm_session=None): - self.encryption = encryption - self.sign = sign - self.rsa = rsa - self.mastertoken = mastertoken - self.cdm_session = cdm_session diff --git a/wpgskd/utils/MSL/MSLObject.py b/wpgskd/utils/MSL/MSLObject.py deleted file mode 100644 index 47264bb..0000000 --- a/wpgskd/utils/MSL/MSLObject.py +++ /dev/null @@ -1,6 +0,0 @@ -import jsonpickle - - -class MSLObject: - def __repr__(self): - return "<{} {}>".format(self.__class__.__name__, jsonpickle.encode(self, unpicklable=False)) diff --git a/wpgskd/utils/MSL/MSL_ANDROID.py b/wpgskd/utils/MSL/MSL_ANDROID.py deleted file mode 100644 index 63737e3..0000000 --- a/wpgskd/utils/MSL/MSL_ANDROID.py +++ /dev/null @@ -1,602 +0,0 @@ -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 -from Cryptodome.Hash import HMAC, SHA256 -from Cryptodome.PublicKey.RSA import RsaKey -from Cryptodome.Random import get_random_bytes -from Cryptodome.Util import Padding -from scripts.pywidevine import Cdm as WidevineCdm, Device as WidevineDevice, PSSH - -class MSLAndroidObject: - def __repr__(self) -> str: - return f"<{self.__class__.__name__} {jsonpickle.encode(self, unpicklable=False)}>" - - -class MSLAndroidKeys(MSLAndroidObject): - def __init__( - self, - encryption: Optional[bytes] = None, - sign: Optional[bytes] = None, - rsa: Optional[RsaKey] = None, - mastertoken: Optional[dict] = None, - cdm_session: Any = None, - ): - self.encryption = encryption - self.sign = sign - self.rsa = rsa - self.mastertoken = mastertoken - self.cdm_session = cdm_session - -class MSLAndroid: - 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 = { - "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 = { - "sdk": "18.26.0", - "platform": "18.26.0", - "application": "Netflix Android 18.26.0", - "uiversion": "18.26.0", - "uiPlatform": "android", - "clientVersion": "18.26.0", - "apkVersion": "18.26.0", - } - DEFAULT_PBO_LANGUAGES = ["en-US", "en"] - DEFAULT_DEVICE_MODEL = "SM-F711N" - - def __init__( - self, - session: requests.Session, - keys: MSLAndroidKeys, - message_id: int, - sender: str, - user_auth: Optional[dict] = None, - drm: str = "widevine", - ): - self.session = session - self.keys = keys - self.sender = sender - self.user_auth = user_auth - self.message_id = message_id - self.drm = drm - - @classmethod - def handshake( - cls, - msl_keys_path: str, - session: requests.Session, - sender: str, - cdm: Any, - cdm_device: Any, - new_msl: bool, - cookies: Optional[Dict[str, str]], - drm: str, - endpoint: Optional[str] = None, - headers: Optional[Dict[str, str]] = None, - ) -> MSLAndroidKeys: - if cookies: - session.cookies.update(cookies) - - cache_path = Path(msl_keys_path) - msl_keys = MSLAndroid.load_cache_data(cache_path) - if msl_keys is not None and not new_msl: - return msl_keys - - if drm != "widevine": - raise ValueError(f"Unsupported DRM mode: {drm}") - - if not cdm: - raise ValueError("Widevine CDM is required for this iOS MSL flow") - - message_id = random.randint(0, pow(2, 52)) - msl_keys = MSLAndroidKeys() - - 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), - ) - wv_request = base64.b64encode(challenge).decode("utf-8") - keyrequestdata = { - "scheme": "WIDEVINE", - "keydata": { - "keyrequest": wv_request, - }, - } - - data = jsonpickle.encode( - { - "entityauthdata": { - "scheme": "NONE", - "authdata": { - "identity": sender, - }, - }, - "headerdata": base64.standard_b64encode( - MSLAndroid.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"Key exchange failed: HTTP {res.status_code} {res.text[:500]}") - - parsed = cls.parse_concatenated_json(res.text) - if not parsed: - raise RuntimeError("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") - 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]}") - - 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"] - - cdm.parse_license(msl_keys.cdm_session, key_data["cdmkeyresponse"]) - keys = cdm.get_keys(msl_keys.cdm_session) - msl_keys.encryption = MSLAndroid.get_widevine_key( - kid=base64.standard_b64decode(key_data["encryptionkeyid"]), - keys=keys, - permissions=["AllowEncrypt", "AllowDecrypt"], - ) - msl_keys.sign = MSLAndroid.get_widevine_key( - kid=base64.standard_b64decode(key_data["hmackeyid"]), - keys=keys, - permissions=["AllowSign", "AllowSignatureVerify"], - ) - - msl_keys.mastertoken = key_response_data["mastertoken"] - MSLAndroid.cache_keys(msl_keys, cache_path) - return msl_keys - - @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, - host: Optional[str] = "android15.prod.cloud.netflix.com", - language: Optional[str] = "en-US", - device_model: Optional[str] = None, - ) -> Dict[str, str]: - headers: Dict[str, str] = { - "Host": host or "android15.prod.cloud.netflix.com", - "Accept": "*/*", - "User-Agent": user_agent or MSLAndroid.DEFAULT_USER_AGENT, - "Accept-Language": language or "en-US", - "Accept-Encoding": "gzip, deflate, br", - "Content-Type": "application/json", - "Connection": "keep-alive", - "Content-Encoding": "msl_v1", - "X-DeviceModel": device_model or MSLAndroid.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.client.context": MSLAndroid.DEFAULT_REQUEST_CONTEXT, - "x-netflix.request.client.languages": "en-US", - "x-netflix.request.client.timezoneid": "America/New_York", - "x-netflix.clienttype": "samurai", - "x-netflix.deviceformfactor": "PHONE", - "x-netflix.devicememorylevel": "HIGH", - "x-netflix.androidapi": "35", - "x-netflix.context.os-version": "35", - "x-netflix.context.form-factor": "phone", - "x-netflix.context.ui-flavor": "android", - "x-netflix.appver": "9.60.0", - "x-netflix.context.app-version": "9.60.0", - "x-netflix.esnprefix": "NFANDROID1-PRV-P-", - "x-netflix.zuul.brotli.allowed": "true", - "x-netflix.request.client.supportskidstop10": "true", - "x-netflix.request.client.supportsgames": "true", - "x-netflix.request.routing": '{"path":"\\/nq\\/android\\/playback\\/~1.0.0\\/router"}', - "x-netflix.context.locales": "en-US", - "x-netflix.context.android.installer-source": "com.android.vending", - } - 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 - headers["x-netflix.esn"] = esn - if expiry_timeout is not None: - headers["x-netflix.request.expiry.timeout"] = str(expiry_timeout) - if extra_headers: - headers.update(extra_headers) - return headers - - @staticmethod - def manifest_request_defaults() -> Tuple[str, Dict[str, str]]: - return MSLAndroid.DEFAULT_MANIFEST_ENDPOINT, dict(MSLAndroid.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'(? 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 - ) - - @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[MSLAndroidKeys]: - 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: MSLAndroidKeys, 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)) diff --git a/wpgskd/utils/MSL/__init__.py b/wpgskd/utils/MSL/__init__.py deleted file mode 100644 index 231ecfd..0000000 --- a/wpgskd/utils/MSL/__init__.py +++ /dev/null @@ -1,401 +0,0 @@ -import base64 -import gzip -import json -import logging -import os -import random -import re -import sys -import time -import zlib -from datetime import datetime -from io import BytesIO - -import jsonpickle -import requests -from Cryptodome.Cipher import AES, PKCS1_OAEP -from Cryptodome.Hash import HMAC, SHA256 -from Cryptodome.PublicKey import RSA -from Cryptodome.Random import get_random_bytes -from Cryptodome.Util import Padding - -from wpgskd.utils.MSL.MSLKeys import MSLKeys -from wpgskd.utils.MSL.schemes import EntityAuthenticationSchemes # noqa: F401 -from wpgskd.utils.MSL.schemes import KeyExchangeSchemes -from wpgskd.utils.MSL.schemes.EntityAuthentication import EntityAuthentication -from wpgskd.utils.MSL.schemes.KeyExchangeRequest import KeyExchangeRequest - -class MSL: - log = logging.getLogger("MSL") - - def __init__(self, session, endpoint, sender, keys, message_id, user_auth=None): - self.session = session - self.endpoint = endpoint - self.sender = sender - self.keys = keys - self.user_auth = user_auth - self.message_id = message_id - - @classmethod - def handshake(cls, scheme, session, endpoint, sender, cdm=None, msl_keys_path=None): - message_id = random.randint(0, pow(2, 52)) - msl_keys = MSL.load_cache_data(msl_keys_path) - if msl_keys is not None: - cls.log.info("Using cached MSL data") - else: - msl_keys = MSLKeys() - if scheme != KeyExchangeSchemes.Widevine: - msl_keys.rsa = RSA.generate(2048) - - if not cdm: - raise cls.log.exit("- No cached data and no CDM specified") - - if not msl_keys_path: - raise cls.log.exit("- No cached data and no MSL key path specified") - - if scheme == KeyExchangeSchemes.Widevine: - msl_keys.cdm_session = cdm.open( - pssh=b"\x0A\x7A\x00\x6C\x38\x2B", - raw=True, - offline=True - ) - keyrequestdata = KeyExchangeRequest.Widevine( - keyrequest=cdm.get_license_challenge(msl_keys.cdm_session) - ) - else: - keyrequestdata = KeyExchangeRequest.AsymmetricWrapped( - keypairid="superKeyPair", - mechanism="JWK_RSA", - publickey=msl_keys.rsa.publickey().exportKey(format="DER") - ) - - data = jsonpickle.encode({ - "entityauthdata": EntityAuthentication.Unauthenticated(sender), - "headerdata": base64.b64encode(MSL.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.b64encode(json.dumps({ - "messageid": message_id, - "data": "", - "sequencenumber": 1, - "endofmsg": True - }).encode("utf-8")).decode("utf-8"), - "signature": "" - }) - - try: - r = session.post( - url=endpoint, - data=data - ) - except requests.HTTPError as e: - raise cls.log.exit(f"- Key exchange failed, response data is unexpected: {e.response.text}") - - key_exchange = r.json() - if "errordata" in key_exchange: - raise cls.log.exit("- Key exchange failed: " + json.loads(base64.b64decode( - key_exchange["errordata"] - ).decode())["errormsg"]) - - # parse the crypto keys - key_response_data = json.JSONDecoder().decode(base64.b64decode( - key_exchange["headerdata"] - ).decode("utf-8"))["keyresponsedata"] - - if key_response_data["scheme"] != str(scheme): - raise cls.log.exit("- Key exchange scheme mismatch occurred") - - key_data = key_response_data["keydata"] - if scheme == KeyExchangeSchemes.Widevine: - if hasattr(cdm, 'device') and isinstance(cdm.device, RemoteDevice): - msl_keys.encryption, msl_keys.sign = cdm.device.exchange( - cdm.sessions[msl_keys.cdm_session], - license_res=key_data["cdmkeyresponse"], - enc_key_id=base64.b64decode(key_data["encryptionkeyid"]), - hmac_key_id=base64.b64decode(key_data["hmackeyid"]) - ) - cdm.parse_license(msl_keys.cdm_session, key_data["cdmkeyresponse"]) - else: - cdm.parse_license(msl_keys.cdm_session, key_data["cdmkeyresponse"]) - keys = cdm.get_keys(msl_keys.cdm_session) - msl_keys.encryption = MSL.get_widevine_key( - kid=base64.b64decode(key_data["encryptionkeyid"]), - keys=keys, - permissions=["AllowEncrypt", "AllowDecrypt"] - ) - msl_keys.sign = MSL.get_widevine_key( - kid=base64.b64decode(key_data["hmackeyid"]), - keys=keys, - permissions=["AllowSign", "AllowSignatureVerify"] - ) - else: - cipher_rsa = PKCS1_OAEP.new(msl_keys.rsa) - msl_keys.encryption = MSL.base64key_decode( - json.JSONDecoder().decode(cipher_rsa.decrypt( - base64.b64decode(key_data["encryptionkey"]) - ).decode("utf-8"))["k"] - ) - msl_keys.sign = MSL.base64key_decode( - json.JSONDecoder().decode(cipher_rsa.decrypt( - base64.b64decode(key_data["hmackey"]) - ).decode("utf-8"))["k"] - ) - msl_keys.mastertoken = key_response_data["mastertoken"] - - MSL.cache_keys(msl_keys, msl_keys_path) - cls.log.info("MSL handshake successful") - return cls( - session=session, - endpoint=endpoint, - sender=sender, - keys=msl_keys, - message_id=message_id - ) - - @staticmethod - def load_cache_data(msl_keys_path=None): - if not msl_keys_path or not os.path.exists(msl_keys_path): - return None - with open(msl_keys_path, encoding="utf-8") as fd: - msl_keys = jsonpickle.decode(fd.read()) - if msl_keys.rsa: - # noinspection PyTypeChecker - # expects RsaKey, but is a string, this is because jsonpickle can't pickle RsaKey object - # so as a workaround it exports to PEM, and then when reading, it imports that PEM back - # to an RsaKey :) - msl_keys.rsa = RSA.importKey(msl_keys.rsa) - # If it's expired or close to, return None as it's unusable - if msl_keys.mastertoken and ((datetime.utcfromtimestamp(int(json.JSONDecoder().decode( - base64.b64decode(msl_keys.mastertoken["tokendata"]).decode("utf-8") - )["expiration"])) - datetime.now()).total_seconds() / 60 / 60) < 10: - return None - return msl_keys - - @staticmethod - def cache_keys(msl_keys, msl_keys_path): - os.makedirs(os.path.dirname(msl_keys_path), exist_ok=True) - if msl_keys.rsa: - # jsonpickle can't pickle RsaKey objects :( - msl_keys.rsa = msl_keys.rsa.export_key() - with open(msl_keys_path, "w", encoding="utf-8") as fd: - fd.write(jsonpickle.encode(msl_keys)) - if msl_keys.rsa: - # re-import now - msl_keys.rsa = RSA.importKey(msl_keys.rsa) - - @staticmethod - def generate_msg_header(message_id, sender, is_handshake, userauthdata=None, keyrequestdata=None, - compression="GZIP"): - """ - The MSL header carries all MSL data used for entity and user authentication, message encryption - and verification, and service tokens. Portions of the MSL header are encrypted. - https://github.com/Netflix/msl/wiki/Messages#header-data - - :param message_id: number against which payload chunks are bound to protect against replay. - :param sender: ESN - :param is_handshake: This flag is set true if the message is a handshake message and will not include any - payload chunks. It will include keyrequestdata. - :param userauthdata: UserAuthData - :param keyrequestdata: KeyRequestData - :param compression: Supported compression algorithms. - - :return: The base64 encoded JSON String of the header - """ - header_data = { - "messageid": message_id, - "renewable": True, # MUST be True if is_handshake - "handshake": is_handshake, - "capabilities": { - "compressionalgos": [compression] if compression else [], - "languages": ["en-US"], # bcp-47 - "encoderformats": ["JSON"] - }, - "timestamp": int(time.time()), - # undocumented or unused: - "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) - - @classmethod - def get_widevine_key(cls, kid, keys, permissions): - for key in keys: - if key.kid != kid: - continue - if key.type != "OPERATOR_SESSION": - cls.log.warning(f"Widevine Key Exchange: Wrong key type (not operator session) key {key}") - continue - if not set(permissions) <= set(key.permissions): - cls.log.warning(f"Widevine Key Exchange: Incorrect permissions, key {key}, needed perms {permissions}") - continue - return key.key - return None - - def send_message(self, endpoint, params, application_data, userauthdata=None): - message = self.create_message(application_data, userauthdata) - res = self.session.post(url=endpoint, data=message, params=params) - header, payload_data = self.parse_message(res.text) - if "errordata" in header: - raise self.log.exit( - "- MSL response message contains an error: {}".format( - json.loads(base64.b64decode(header["errordata"].encode("utf-8")).decode("utf-8")) - ) - ) - return header, payload_data - - def create_message(self, application_data, userauthdata=None): - self.message_id += 1 # new message must ue a new message id - - headerdata = self.encrypt(self.generate_msg_header( - message_id=self.message_id, - sender=self.sender, - is_handshake=False, - userauthdata=userauthdata - )) - - header = json.dumps({ - "headerdata": base64.b64encode(headerdata.encode("utf-8")).decode("utf-8"), - "signature": self.sign(headerdata).decode("utf-8"), - "mastertoken": self.keys.mastertoken - }) - - payload_chunks = [self.encrypt(json.dumps({ - "messageid": self.message_id, - "data": self.gzip_compress(json.dumps(application_data).encode("utf-8")).decode("utf-8"), - "compressionalgo": "GZIP", - "sequencenumber": 1, # todo ; use sequence_number from master token instead? - "endofmsg": True - }))] - - message = header - for payload_chunk in payload_chunks: - message += json.dumps({ - "payload": base64.b64encode(payload_chunk.encode("utf-8")).decode("utf-8"), - "signature": self.sign(payload_chunk).decode("utf-8") - }) - - return message - - def decrypt_payload_chunks(self, payload_chunks): - """ - Decrypt and extract data from payload chunks - - :param payload_chunks: List of payload chunks - :return: json object - """ - raw_data = "" - - for payload_chunk in payload_chunks: - # todo ; verify signature of payload_chunk["signature"] against payload_chunk["payload"] - # expecting base64-encoded json string - payload_chunk = json.loads(base64.b64decode(payload_chunk["payload"]).decode("utf-8")) - # decrypt the payload - payload_decrypted = AES.new( - key=self.keys.encryption, - mode=AES.MODE_CBC, - iv=base64.b64decode(payload_chunk["iv"]) - ).decrypt(base64.b64decode(payload_chunk["ciphertext"])) - payload_decrypted = Padding.unpad(payload_decrypted, 16) - payload_decrypted = json.loads(payload_decrypted.decode("utf-8")) - # decode and uncompress data if compressed - payload_data = base64.b64decode(payload_decrypted["data"]) - if payload_decrypted.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: - error = data["error"] - error_display = error.get("display") - error_detail = re.sub(r" \(E3-[^)]+\)", "", error.get("detail", "")) - - if error_display: - self.log.critical(f"- {error_display}") - if error_detail: - self.log.critical(f"- {error_detail}") - - if not (error_display or error_detail): - self.log.critical(f"- {error}") - - sys.exit(1) - - return data["result"] - - def parse_message(self, message): - """ - Parse an MSL message into a header and list of payload chunks - - :param message: MSL message - :returns: a 2-item tuple containing message and list of payload chunks if available - """ - parsed_message = json.loads("[{}]".format(message.replace("}{", "},{"))) - - header = parsed_message[0] - encrypted_payload_chunks = parsed_message[1:] if len(parsed_message) > 1 else [] - if encrypted_payload_chunks: - payload_chunks = self.decrypt_payload_chunks(encrypted_payload_chunks) - else: - payload_chunks = {} - - return header, payload_chunks - - @staticmethod - def gzip_compress(data): - out = BytesIO() - with gzip.GzipFile(fileobj=out, mode="w") as fd: - fd.write(data) - return base64.b64encode(out.getvalue()) - - @staticmethod - def base64key_decode(payload): - 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): - """ - Encrypt the given Plaintext with the encryption key - :param plaintext: - :return: Serialized JSON String of the encryption Envelope - """ - iv = get_random_bytes(16) - return json.dumps({ - "ciphertext": base64.b64encode( - AES.new( - self.keys.encryption, - AES.MODE_CBC, - iv - ).encrypt( - Padding.pad(plaintext.encode("utf-8"), 16) - ) - ).decode("utf-8"), - "keyid": "{}_{}".format(self.sender, json.loads( - base64.b64decode(self.keys.mastertoken["tokendata"]).decode("utf-8") - )["sequencenumber"]), - "sha256": "AA==", - "iv": base64.b64encode(iv).decode("utf-8") - }) - - def sign(self, text): - """ - Calculates the HMAC signature for the given text with the current sign key and SHA256 - :param text: - :return: Base64 encoded signature - """ - return base64.b64encode(HMAC.new(self.keys.sign, text.encode("utf-8"), SHA256).digest()) diff --git a/wpgskd/utils/MSL/schemes/EntityAuthentication.py b/wpgskd/utils/MSL/schemes/EntityAuthentication.py deleted file mode 100644 index b46dd94..0000000 --- a/wpgskd/utils/MSL/schemes/EntityAuthentication.py +++ /dev/null @@ -1,59 +0,0 @@ -from wpgskd.utils.MSL import EntityAuthenticationSchemes -from wpgskd.utils.MSL.MSLObject import MSLObject - - -# noinspection PyPep8Naming -class EntityAuthentication(MSLObject): - def __init__(self, scheme, authdata): - """ - Data used to identify and authenticate the entity associated with a message. - https://github.com/Netflix/msl/wiki/Entity-Authentication-%28Configuration%29 - - :param scheme: Entity Authentication Scheme identifier - :param authdata: Entity Authentication data - """ - self.scheme = str(scheme) - self.authdata = authdata - - @classmethod - def Unauthenticated(cls, identity): - """ - The unauthenticated entity authentication scheme does not provide encryption or authentication and only - identifies the entity. Therefore entity identities can be harvested and spoofed. The benefit of this - authentication scheme is that the entity has control over its identity. This may be useful if the identity is - derived from or related to other data, or if retaining the identity is desired across state resets or in the - event of MSL errors requiring entity re-authentication. - """ - return cls( - scheme=EntityAuthenticationSchemes.Unauthenticated, - authdata={"identity": identity} - ) - - @classmethod - def Widevine(cls, devtype, keyrequest): - """ - The Widevine entity authentication scheme is used by devices with the Widevine CDM. It does not provide - encryption or authentication and only identifies the entity. Therefore entity identities can be harvested - and spoofed. The entity identity is composed from the provided device type and Widevine key request data. The - Widevine CDM properties can be extracted from the key request data. - - When coupled with the Widevine key exchange scheme, the entity identity can be cryptographically validated by - comparing the entity authentication key request data against the key exchange key request data. - - Note that the local entity will not know its entity identity when using this scheme. - - > Devtype - - An arbitrary value identifying the device type the local entity wishes to assume. The data inside the Widevine - key request may be optionally used to validate the claimed device type. - - :param devtype: Local entity device type - :param keyrequest: Widevine key request - """ - return cls( - scheme=EntityAuthenticationSchemes.Widevine, - authdata={ - "devtype": devtype, - "keyrequest": keyrequest - } - ) diff --git a/wpgskd/utils/MSL/schemes/KeyExchangeRequest.py b/wpgskd/utils/MSL/schemes/KeyExchangeRequest.py deleted file mode 100644 index 31eff61..0000000 --- a/wpgskd/utils/MSL/schemes/KeyExchangeRequest.py +++ /dev/null @@ -1,80 +0,0 @@ -import base64 - -from wpgskd.utils.MSL import KeyExchangeSchemes -from wpgskd.utils.MSL.MSLObject import MSLObject - - -# noinspection PyPep8Naming -class KeyExchangeRequest(MSLObject): - def __init__(self, scheme, keydata): - """ - Session key exchange data from a requesting entity. - https://github.com/Netflix/msl/wiki/Key-Exchange-%28Configuration%29 - - :param scheme: Key Exchange Scheme identifier - :param keydata: Key Request data - """ - self.scheme = str(scheme) - self.keydata = keydata - - @classmethod - def AsymmetricWrapped(cls, keypairid, mechanism, publickey): - """ - Asymmetric wrapped key exchange uses a generated ephemeral asymmetric key pair for key exchange. It will - typically be used when there is no other data or keys from which to base secure key exchange. - - This mechanism provides perfect forward secrecy but does not guarantee that session keys will only be available - to the requesting entity if the requesting MSL stack has been modified to perform the operation on behalf of a - third party. - - > Key Pair ID - - The key pair ID is included as a sanity check. - - > Mechanism & Public Key - - The following mechanisms are associated public key formats are currently supported. - - Field Public Key Format Description - RSA SPKI RSA-OAEP encrypt/decrypt - ECC SPKI ECIES encrypt/decrypt - JWEJS_RSA SPKI RSA-OAEP JSON Web Encryption JSON Serialization - JWE_RSA SPKI RSA-OAEP JSON Web Encryption Compact Serialization - JWK_RSA SPKI RSA-OAEP JSON Web Key - JWK_RSAES SPKI RSA PKCS#1 JSON Web Key - - :param keypairid: key pair ID - :param mechanism: asymmetric key type - :param publickey: public key - """ - return cls( - scheme=KeyExchangeSchemes.AsymmetricWrapped, - keydata={ - "keypairid": keypairid, - "mechanism": mechanism, - "publickey": base64.b64encode(publickey).decode("utf-8") - } - ) - - @classmethod - def Widevine(cls, keyrequest): - """ - Google Widevine provides a secure key exchange mechanism. When requested the Widevine component will issue a - one-time use key request. The Widevine server library can be used to authenticate the request and return - randomly generated symmetric keys in a protected key response bound to the request and Widevine client library. - The key response also specifies the key identities, types and their permitted usage. - - The Widevine key request also contains a model identifier and a unique device identifier with an expectation of - long-term persistence. These values are available from the Widevine client library and can be retrieved from - the key request by the Widevine server library. - - The Widevine client library will protect the returned keys from inspection or misuse. - - :param keyrequest: Base64-encoded Widevine CDM license challenge (PSSH: b'\x0A\x7A\x00\x6C\x38\x2B') - """ - if not isinstance(keyrequest, str): - keyrequest = base64.b64encode(keyrequest).decode() - return cls( - scheme=KeyExchangeSchemes.Widevine, - keydata={"keyrequest": keyrequest} - ) diff --git a/wpgskd/utils/MSL/schemes/PlayReadyKeyExchangeScheme.py b/wpgskd/utils/MSL/schemes/PlayReadyKeyExchangeScheme.py deleted file mode 100644 index 843a8e9..0000000 --- a/wpgskd/utils/MSL/schemes/PlayReadyKeyExchangeScheme.py +++ /dev/null @@ -1,111 +0,0 @@ -import base64 -import json -import os -from Cryptodome.Cipher import AES, PKCS1_OAEP -from Cryptodome.Hash import HMAC, SHA256 -from Cryptodome.Random import get_random_bytes -from Cryptodome.Util import Padding - -from wpgskd.utils.MSL.schemes.KeyExchangeRequest import KeyExchangeRequest - -class PlayReady(KeyExchangeRequest): - """ - Implementation of the PlayReady Key Exchange Scheme for MSL. - """ - - def __init__(self): - self.encryption_key = None - self.sign_key = None - self.sender = None - - def perform_key_exchange(self, session, endpoint, sender, cdm): - """ - Performs a key exchange using PlayReady. - - Parameters: - session: HTTP session with necessary cookies - endpoint: Endpoint for key exchange - sender: ESN of the device - cdm: CDM instance - """ - self.sender = sender - - # Generate random keys for encryption and signing - self.encryption_key = get_random_bytes(16) # AES-128 - self.sign_key = get_random_bytes(32) # HMAC-SHA256 - - # Return keys in the format required by MSL - return { - "encryptionkey": base64.b64encode(self.encryption_key).decode("utf-8"), - "hmackey": base64.b64encode(self.sign_key).decode("utf-8") - } - - def encrypt(self, data, encryption_envelope=None): - """ - Encrypts data using the encryption key. - - Parameters: - data: Data to encrypt - encryption_envelope: Not used in PlayReady - """ - if not self.encryption_key: - raise ValueError("No encryption key available") - - # Generate a random IV - iv = get_random_bytes(16) - - # Encrypt the data - cipher = AES.new(self.encryption_key, AES.MODE_CBC, iv) - ciphertext = cipher.encrypt(Padding.pad(data.encode("utf-8"), 16)) - - # Return encrypted data in the format required by MSL - return { - "keyid": self.sender, - "iv": base64.b64encode(iv).decode("utf-8"), - "ciphertext": base64.b64encode(ciphertext).decode("utf-8") - } - - def decrypt(self, data): - """ - Decrypts data using the encryption key. - - Parameters: - data: Encrypted data (with iv and ciphertext) - """ - if not self.encryption_key: - raise ValueError("No encryption key available") - - # Decode IV and ciphertext - iv = base64.b64decode(data["iv"]) - ciphertext = base64.b64decode(data["ciphertext"]) - - # Decrypt the data - cipher = AES.new(self.encryption_key, AES.MODE_CBC, iv) - plaintext = Padding.unpad(cipher.decrypt(ciphertext), 16) - - return plaintext - - def sign(self, data): - """ - Signs data using the sign key. - - Parameters: - data: Data to sign - """ - if not self.sign_key: - raise ValueError("No sign key available") - - # Sign the data with HMAC-SHA256 - signature = HMAC.new(self.sign_key, data.encode("utf-8"), SHA256).digest() - return base64.b64encode(signature) - - def verify(self, data, signature): - """ - Verifies a signature. - - Parameters: - data: Data that was signed - signature: Signature to verify - """ - expected_signature = self.sign(data) - return signature == expected_signature.decode("utf-8") \ No newline at end of file diff --git a/wpgskd/utils/MSL/schemes/UserAuthentication.py b/wpgskd/utils/MSL/schemes/UserAuthentication.py deleted file mode 100644 index 7385964..0000000 --- a/wpgskd/utils/MSL/schemes/UserAuthentication.py +++ /dev/null @@ -1,59 +0,0 @@ -from wpgskd.utils.MSL.MSLObject import MSLObject -from wpgskd.utils.MSL.schemes import UserAuthenticationSchemes - - -# noinspection PyPep8Naming -class UserAuthentication(MSLObject): - def __init__(self, scheme, authdata): - """ - Data used to identify and authenticate the user associated with a message. - https://github.com/Netflix/msl/wiki/User-Authentication-%28Configuration%29 - - :param scheme: User Authentication Scheme identifier - :param authdata: User Authentication data - """ - self.scheme = str(scheme) - self.authdata = authdata - - @classmethod - def EmailPassword(cls, email, password): - """ - Email and password is a standard user authentication scheme in wide use. - - :param email: user email address - :param password: user password - """ - return cls( - scheme=UserAuthenticationSchemes.EmailPassword, - authdata={ - "email": email, - "password": password - } - ) - - @classmethod - def NetflixIDCookies(cls, netflixid, securenetflixid): - """ - Netflix ID HTTP cookies are used when the user has previously logged in to a web site. Possession of the - cookies serves as proof of user identity, in the same manner as they do when communicating with the web site. - - The Netflix ID cookie and Secure Netflix ID cookie are HTTP cookies issued by the Netflix web site after - subscriber login. The Netflix ID cookie is encrypted and identifies the subscriber and analogous to a - subscriber’s username. The Secure Netflix ID cookie is tied to a Netflix ID cookie and only sent over HTTPS - and analogous to a subscriber’s password. - - In some cases the Netflix ID and Secure Netflix ID cookies will be unavailable to the MSL stack or application. - If either or both of the Netflix ID or Secure Netflix ID cookies are absent in the above data structure the - HTTP cookie headers will be queried for it; this is only acceptable when HTTPS is used as the underlying - transport protocol. - - :param netflixid: Netflix ID cookie - :param securenetflixid: Secure Netflix ID cookie - """ - return cls( - scheme=UserAuthenticationSchemes.NetflixIDCookies, - authdata={ - "netflixid": netflixid, - "securenetflixid": securenetflixid - } - ) diff --git a/wpgskd/utils/MSL/schemes/__init__.py b/wpgskd/utils/MSL/schemes/__init__.py deleted file mode 100644 index a1f61d7..0000000 --- a/wpgskd/utils/MSL/schemes/__init__.py +++ /dev/null @@ -1,24 +0,0 @@ -from enum import Enum - - -class Scheme(Enum): - def __str__(self): - return str(self.value) - - -class EntityAuthenticationSchemes(Scheme): - """https://github.com/Netflix/msl/wiki/Entity-Authentication-%28Configuration%29""" - Unauthenticated = "NONE" - Widevine = "WIDEVINE" - - -class UserAuthenticationSchemes(Scheme): - """https://github.com/Netflix/msl/wiki/User-Authentication-%28Configuration%29""" - EmailPassword = "EMAIL_PASSWORD" - NetflixIDCookies = "NETFLIXID" - - -class KeyExchangeSchemes(Scheme): - """https://github.com/Netflix/msl/wiki/Key-Exchange-%28Configuration%29""" - AsymmetricWrapped = "ASYMMETRIC_WRAPPED" - Widevine = "WIDEVINE" diff --git a/wpgskd/utils/__init__.py b/wpgskd/utils/__init__.py deleted file mode 100644 index 47fd6c2..0000000 --- a/wpgskd/utils/__init__.py +++ /dev/null @@ -1,71 +0,0 @@ -from langcodes import Language, closest_match -from typing import Any, Optional, Union -import base64 -from hashlib import md5 - -from wpgskd.core.constants import LANGUAGE_MAX_DISTANCE -from wpgskd.vendor.pymp4.parser import Box - -from pywidevine.cdm import Cdm -from pywidevine.license_protocol_pb2 import WidevinePsshData - -def get_boxes(data: bytes, box_type: bytes, as_bytes: bool = False): - """Scan a byte array for a wanted box, then parse and yield each find.""" - if not isinstance(data, (bytes, bytearray)): - raise ValueError("data must be bytes") - while True: - try: - index = data.index(box_type) - except ValueError: - break - if index < 0: - break - if index > 4: - index -= 4 # size is before box type and is 4 bytes long - data = data[index:] - try: - box = Box.parse(data) - except IOError: - break - if as_bytes: - box = Box.build(box) - yield box - -def is_close_match(language: Any, languages: Any) -> bool: - if not (language and languages and all(languages)): - return False - languages = list(map(str, [x for x in languages if x])) - return closest_match(language, languages)[1] <= LANGUAGE_MAX_DISTANCE - -def get_closest_match(language: Any, languages: Any) -> Optional[Language]: - match, distance = closest_match(language, list(map(str, languages))) - if distance > LANGUAGE_MAX_DISTANCE: - return None - return Language.get(match) - -def numeric_quality(quality: Any) -> int: - if not quality: - return 0 - if quality == "SD": - return 576 - return int(quality) - -def try_get(obj: Any, func: callable) -> Any: - try: - return func(obj) - except (AttributeError, IndexError, KeyError, TypeError): - return None - -def short_hash(input: Any) -> str: - return base_encode(int(md5(str(input).encode()).hexdigest(), 16)) - -def base_encode(num: int) -> str: - alphabet = "0123456789ABCDEFGHIJKLMNOPQRSTUVWXYZabcdefghijklmnopqrstuvwxyz" - if num == 0: - return alphabet[0] - arr = [] - base = len(alphabet) - while num: - num, rem = divmod(num, base) - arr.append(alphabet[rem]) - return "".join(reversed(arr)) \ No newline at end of file diff --git a/wpgskd/utils/adobepass.py b/wpgskd/utils/adobepass.py deleted file mode 100644 index 7a11e7d..0000000 --- a/wpgskd/utils/adobepass.py +++ /dev/null @@ -1,19 +0,0 @@ -import os -from abc import ABC - -from yt_dlp import YoutubeDL -from yt_dlp.extractor.adobepass import AdobePassIE - - -class AdobePassVT(AdobePassIE, ABC): - def __init__(self, credential, get_cache): - super().__init__( - YoutubeDL( - { - "ap_mso": credential.extra, # See yt_dlp.extractor.adobepass for supported MSO providers - "ap_username": credential.username, - "ap_password": credential.password, - "cachedir": os.path.realpath(get_cache("adobepass")), - } - ) - ) diff --git a/wpgskd/utils/base64.py b/wpgskd/utils/base64.py deleted file mode 100644 index cad6567..0000000 --- a/wpgskd/utils/base64.py +++ /dev/null @@ -1,25 +0,0 @@ -import base64 - - -def encode(s): - if isinstance(s, str): - s = s.encode() - return base64.b64encode(s).decode() - - -def decode(s): - if isinstance(s, str): - s = s.encode() - return base64.b64decode(s + b"==") - - -def urlsafe_encode(s): - if isinstance(s, str): - s = s.encode() - return base64.urlsafe_b64encode(s).decode().rstrip("=") - - -def urlsafe_decode(s): - if isinstance(s, str): - s = s.encode() - return base64.urlsafe_b64decode(s + b"==") diff --git a/wpgskd/utils/click.py b/wpgskd/utils/click.py deleted file mode 100644 index 471b9b2..0000000 --- a/wpgskd/utils/click.py +++ /dev/null @@ -1,142 +0,0 @@ -import re -import logging -import click -from wpgskd import servicookies as services -from wpgskd.utils.collections import as_list - -log = logging.getLogger("click") - -class ContextData: - def __init__(self, config, vaults, cdm, profile=None, cookies=None, credentials=None): - self.config = config - self.vaults = vaults - self.cdm = cdm - self.profile = profile - self.cookies = cookies - self.credentials = credentials - -class AliasedGroup(click.Group): - def get_command(self, ctx, cmd_name): - rv = click.Group.get_command(self, ctx, cmd_name) - if rv is not None: - return rv - - for key, aliases in services.SERVICE_MAP.items(): - if cmd_name.lower() in map(str.lower, aliases): - if hasattr(services, key): - service_cls = getattr(services, key) - if hasattr(service_cls, 'cli'): - return service_cls.cli - return None - - service_key = services.get_service_key(cmd_name) - if not service_key: - for name in services.SERVICE_MAP: - if hasattr(services, name): - service_cls = getattr(services, name) - title_re = as_list(getattr(service_cls, "TITLE_RE", [])) - for regex in title_re: - m = re.search(regex, cmd_name) - if m and m.group().startswith(("http://", "https://", "urn:")): - service_key = name - ctx.params["service_name"] = name - if "id" in m.groupdict(): - ctx.params["title"] = m.group("id") - else: - ctx.params["title"] = cmd_name - break - if service_key: - break - - if service_key and hasattr(services, service_key): - service_cls = getattr(services, service_key) - if hasattr(service_cls, 'cli'): - return service_cls.cli - - def list_commands(self, ctx): - return sorted(self.commands, key=str.casefold) - -def _choice(ctx, param, value, value_map): - if value is None: return None - if value.lower() in value_map: - return value_map[value.lower()] - else: - valid_values = {x: None for x in value_map.values()} - valid_values = ", ".join(repr(x) for x in valid_values) - ctx.fail(f"Invalid value for {param.name!r}: {value!r} is not one of {valid_values}.") - -def acodec_param(ctx, param, value): - return _choice(ctx, param, value, { - "aac": "AAC", "ac3": "AC3", "ac-3": "AC3", "dd": "AC3", - "ec3": "EC3", "ec-3": "EC3", "eac3": "EC3", "e-ac3": "EC3", "e-ac-3": "EC3", "dd+": "EC3", "ddp": "EC3", - "vorb": "VORB", "vorbis": "VORB", "opus": "OPUS", "flac": "FLAC", - }) - -def channels_param(ctx, param, value): - return _choice(ctx, param, value, { - "2": "2.0", "2.0": "2.0", "5.1": "5.1", "6": "5.1", - "7.1": "7.1", "atmos": "16/JOC", "16/joc": "16/JOC", - }) - -def language_param(ctx, param, value): - if isinstance(value, list): return value - if not value: return [] - return re.split(r"\s*[,;]\s*", value) - -def quality_param(ctx, param, value): - if not value: return None - if value.lower() == "sd": return "SD" - if value.lower() == "hd720": return "HD720" - if value.lower() == "4k": return 2160 - try: - return int(value.lower().rstrip("p")) - except TypeError: - ctx.fail(f"expected string for int() conversion, got {value!r}", param, ctx) - except ValueError: - ctx.fail(f"{value!r} is not a valid integer", param, ctx) - -def range_param(ctx, param, value): - return _choice(ctx, param, value, { - "sdr": "SDR", "hdr": "HDR10", "hdr10": "HDR10", - "hdr10+": "HDR10+", "hdr10plus": "HDR10+", - "hlg": "HLG", "dv": "DV", "dovi": "DV", - "dv+hdr": "DV+HDR", "dvhdr": "DV+HDR" - }) - -def vcodec_param(ctx, param, value): - return _choice(ctx, param, value, { - "h264": "H264", "avc": "H264", "h265": "H265", "hevc": "H265", - "vp9": "VP9", "av1": "AV1", - }) - -def wanted_param(ctx, param, value): - MIN_EPISODE = 0 - MAX_EPISODE = 9999 - - def parse_tokens(*tokens): - if len(tokens) == 0: return [] - computed, exclusions = [], [] - for token in tokens: - exclude = token.startswith("-") - if exclude: token = token[1:] - parsed = [re.match(r"^S(?P\d+)(E(?P\d+))?$", x, re.IGNORECASE) for x in re.split(r"[:-]", token)] - if len(parsed) > 2: ctx.fail(f"Invalid token: {token}") - if len(parsed) == 1: parsed.append(parsed[0]) - if any(x is None for x in parsed): ctx.fail(f"Invalid token syntax: {token}") - - from_season, from_episode = [int(v) if v is not None else MIN_EPISODE for k, v in parsed[0].groupdict().items() if parsed[0]] - to_season, to_episode = [int(v) if v is not None else MAX_EPISODE for k, v in parsed[1].groupdict().items() if parsed[1]] - - if from_season > to_season: ctx.fail(f"Invalid range: {token}") - if from_season == to_season and from_episode > to_episode: ctx.fail(f"Invalid range: {token}") - - for s in range(from_season, to_season + 1): - for e in range(from_episode if s == from_season else 0, (MAX_EPISODE if s < to_season else to_episode) + 1): - (computed if not exclude else exclusions).append(f"{s}x{e}") - - for exclusion in exclusions: - if exclusion in computed: computed.remove(exclusion) - return list(set(computed)) - - if value: - return parse_tokens(*re.split(r"\s*[,;]\s*", value)) \ No newline at end of file diff --git a/wpgskd/utils/collections.py b/wpgskd/utils/collections.py deleted file mode 100644 index e763c60..0000000 --- a/wpgskd/utils/collections.py +++ /dev/null @@ -1,54 +0,0 @@ -import itertools -from typing import Iterable, Sequence - -class CaseInsensitiveDict(dict): - """A dictionary with case-insensitive keys.""" - def __init__(self, *args, **kwargs): - super().__init__(*args, **kwargs) - for k in self.keys(): - if not isinstance(k, (str, bytes)): - raise ValueError(f"dictionary keys must be str or bytes, not {type(k)}") - - def _resolve_key(self, key): - if not isinstance(key, (str, bytes)): - raise ValueError(f"dictionary keys must be str or bytes, not {type(key)}") - return next((x for x in self.keys() if x.casefold() == key.casefold()), key) - - def __contains__(self, key): return super().__contains__(self._resolve_key(key)) - def __getitem__(self, key): return super().__getitem__(self._resolve_key(key)) - def __setitem__(self, key, value): return super().__setitem__(self._resolve_key(key), value) - def get(self, key, default=None): return super().get(self._resolve_key(key), default) - def pop(self, key): return super().pop(self._resolve_key(key)) - def setdefault(self, key, value=None): return super().setdefault(self._resolve_key(key), value) - -def as_lists(*args): - for item in args: - yield item if isinstance(item, list) else [item] - -def as_list(*args): - if args == (None,): return [] - return list(itertools.chain.from_iterable(as_lists(*args))) - -def first(iterable): return next(iter(iterable)) -def first_or_else(iterable, default): - item = next(iter(iterable or []), None) - return default if item is None else item -def first_or_none(iterable): return first_or_else(iterable, None) - -def flatten(items, ignore_types=str): - if isinstance(items, (Iterable, Sequence)) and not isinstance(items, ignore_types): - for i in items: - yield from flatten(i, ignore_types) - else: - yield items - -def merge_dict(*dicts): - """Recursively merge dicts into dest in-place.""" - dest = dicts[0] - for d in dicts[1:]: - for key, value in d.items(): - if isinstance(value, dict): - node = dest.setdefault(key, {}) - merge_dict(node, value) - else: - dest[key] = value \ No newline at end of file diff --git a/wpgskd/utils/drmtoday.py b/wpgskd/utils/drmtoday.py deleted file mode 100644 index 59983b9..0000000 --- a/wpgskd/utils/drmtoday.py +++ /dev/null @@ -1,25 +0,0 @@ -# Most of these were taken from the old player JS. -# 40002 was inferred based on tests with actual keys. -DRMTODAY_RESPONSE_CODES = { - "00000": "Success", - "01000": "General Internal Error", - "02000": "General Request Error", - "03000": "General Request Authentication Error", - "30000": "General DRM Error", - "40000": "General Widevine Modular Error", - "40001": "Widevine Device Certificate Revocation", - "40002": "Widevine Device Certificate Serial Number Revocation", - "41000": "General Widevine Classic Error", - "42000": "General PlayReady Error", - "43000": "General FairPlay Error", - "44000": "General OMA Error", - "44001": "OMA Device Registration Failed", - "45000": "General CDRM Error", - "45001": "CDRM Device Registration Failed", - "70000": "General Output Protection Error", - "70001": "All keys filtered by EOP settings", - "80000": "General CSL Error", - "80001": "Too many concurrent streams", - "90000": "General GBL Error", - "90001": "License delivery prohibited in your region" -} diff --git a/wpgskd/utils/gen_esn.py b/wpgskd/utils/gen_esn.py deleted file mode 100644 index 40b32a3..0000000 --- a/wpgskd/utils/gen_esn.py +++ /dev/null @@ -1,31 +0,0 @@ -from datetime import datetime, timedelta -import os -import logging -import random - - -log = logging.getLogger("NF-ESN") - -def chrome_esn_generator(): - - ESN_GEN = "".join(random.choice("0123456789ABCDEF") for _ in range(30)) - esn_file = '.esn' - - def gen_file(): - with open(esn_file, 'w') as file: - file.write(f'NFCDIE-03-{ESN_GEN}') - - if not os.path.isfile(esn_file): - log.warning("Generating a new Chrome ESN") - gen_file() - - file_datetime = datetime.fromtimestamp(os.path.getmtime(esn_file)) - time_diff = datetime.now() - file_datetime - if time_diff > timedelta(hours=6): - log.warning("Old ESN detected, Generating a new Chrome ESN") - gen_file() - - with open(esn_file, 'r') as f: - esn = f.read() - - return esn diff --git a/wpgskd/utils/io.py b/wpgskd/utils/io.py deleted file mode 100644 index 6476136..0000000 --- a/wpgskd/utils/io.py +++ /dev/null @@ -1,208 +0,0 @@ -import asyncio -import contextlib -import logging -import os -import re -import shutil -import subprocess -import sys -from pathlib import Path - -import httpx -import pproxy -import requests -import yaml -import tqdm - -log = logging.getLogger("io") - -def load_yaml(path: str) -> dict: - if not os.path.isfile(path): - return {} - with open(path) as fd: - return yaml.safe_load(fd) - -_ip_info = None - -def get_ip_info(session=None, fresh=False) -> dict: - """Use multiple services to get IP location information.""" - global _ip_info - if fresh or not _ip_info: - session = session or httpx - try: - resp = session.get("https://ipwho.is/").json() - if resp.get("success") is not False: - _ip_info = resp - return _ip_info - except Exception: - pass - try: - resp = session.get("http://ip-api.com/json/").json() - if "countryCode" in resp: - _ip_info = {"country_code": resp["countryCode"]} - return _ip_info - except Exception: - pass - logging.getLogger("io").warning("Failed to get IP info. Assuming US.") - _ip_info = {"country_code": "US"} - return _ip_info - -@contextlib.asynccontextmanager -async def start_pproxy(host, port, username, password): - rerouted_proxy = "http://localhost:8081" - server = pproxy.Server(rerouted_proxy) - remote = pproxy.Connection(f"http+ssl://{host}:{port}#{username}:{password}") - handler = await server.start_server(dict(rserver=[remote])) - try: - yield rerouted_proxy - finally: - handler.close() - await handler.wait_closed() - -async def aria2c(uri, out, headers=None, proxy=None): - """Downloads file(s) using Aria2(c).""" - executable = shutil.which("aria2c") or shutil.which("aria2") - if not executable: - raise EnvironmentError("Aria2c executable not found...") - - arguments = [ - executable, "-c", "--remote-time", - "-o", os.path.basename(out), - "-x", "16", "-j", "16", "-s", "16", - "--allow-overwrite=true", "--auto-file-renaming=false", - "--retry-wait", "5", "--max-tries", "15", - "--max-file-not-found", "15", "--summary-interval", "0", - "--file-allocation", "none" if sys.platform == "win32" else "falloc", - "--console-log-level", "warn", "--download-result", "hide", - "--user-agent", "Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36" - ] - - for header, value in (headers or {}).items(): - if header.lower() == "accept-encoding": continue - arguments.extend(["--header", f"{header}: {value}"]) - - segmented = isinstance(uri, list) - segments_dir = f"{out}_segments" - - if segmented: - uri = "\n".join([ - f"{url}\n\tdir={segments_dir}\n\tout={i:08}.mp4" - for i, url in enumerate(uri) - ]) - - if proxy: - arguments.append("--all-proxy") - if proxy.lower().startswith("https://"): - auth, hostname = proxy[8:].split("@") - async with start_pproxy(*hostname.split(":"), *auth.split(":")) as pproxy_: - arguments.extend([pproxy_, "-d"]) - if segmented: - arguments.extend([segments_dir, "-i-"]) - proc = await asyncio.create_subprocess_exec(*arguments, stdin=subprocess.PIPE) - await proc.communicate(uri.encode("utf-8")) - else: - arguments.extend([os.path.dirname(out), uri]) - proc = await asyncio.create_subprocess_exec(*arguments) - await proc.communicate() - else: - arguments.append(proxy) - - try: - if segmented: - subprocess.run(arguments + ["-d", segments_dir, "-i-"], input=uri, encoding="utf-8", check=True) - else: - subprocess.run(arguments + ["-d", os.path.dirname(out), uri], check=True) - except subprocess.CalledProcessError: - raise ValueError("Aria2c failed too many times, aborting") - - if segmented: - with open(out, "wb") as ofd: - for file in sorted(os.listdir(segments_dir)): - file_path = os.path.join(segments_dir, file) - with open(file_path, "rb") as ifd: - data = ifd.read() - # Apple TV+ audio decryption fix - data = re.sub(b"(tfhd\x00\x02\x00\x1a\x00\x00\x00\x01\x00\x00\x00)\x02", b"\\g<1>\x01", data) - ofd.write(data) - os.unlink(file) - os.rmdir(segments_dir) - -async def m3u8re(uri, out, headers=None, proxy=None, key=None): - out = Path(out) - if headers: - headers.update({k: v for k, v in headers.items() if k.lower() != "accept-encoding"}) - - executable = shutil.which("m3u8re") or shutil.which("N_m3u8DL-RE") - if not executable: - raise EnvironmentError("N_m3u8DL-RE executable not found...") - - if isinstance(uri, list): - uri = uri[0] - - arguments = [ - executable, uri, - "--tmp-dir", str(out.parent), - "--save-dir", str(out.parent), - "--save-name", out.name.replace('.mp4','').replace('.vtt','').replace('.m4a',''), - "--auto-subtitle-fix", "False", - "--thread-count", "32", - "--log-level", "INFO" - ] - - if key: - arguments.extend(["--key", key]) - - if headers: - arguments.extend(["--header", "\r\n".join([f"{k}: {v}" for k, v in headers.items()])]) - - if proxy: - arguments.extend(["--custom-proxy", proxy]) - - try: - subprocess.run(arguments, check=True) - except subprocess.CalledProcessError: - raise ValueError("N_m3u8DL-RE failed too many times, aborting") - -async def n_m3u8dl_re_dash(manifest_url, save_path, headers=None, proxy=None, track_id=None, kid_key_pairs=None): - out = Path(save_path) - executable = shutil.which("N_m3u8DL-RE") or shutil.which("m3u8re") - if not executable: - raise EnvironmentError("N_m3u8DL-RE executable was not found.") - - arguments = [ - executable, manifest_url, - "--tmp-dir", str(out.parent), "--save-dir", str(out.parent), - "--save-name", out.stem, "--log-level", "INFO", - "--mux-after-done", "format=mkv", - ] - - if headers: - arguments.extend(["--header", "\r\n".join([f"{k}: {v}" for k, v in headers.items()])]) - - if proxy: - arguments.extend(["--custom-proxy", proxy]) - else: - arguments.append("--no-proxy") - - if track_id: - if 'video' in out.name.lower(): - arguments.extend(["-sv", f"id={track_id}", "-sa", "best"]) - elif 'audio' in out.name.lower(): - arguments.extend(["-sa", f"id={track_id}"]) - - if kid_key_pairs: - for kid, key in kid_key_pairs: - arguments.extend(["--key", f"{kid}:{key}"]) - - try: - subprocess.run(arguments, check=True, capture_output=True, text=True, encoding='utf-8', errors='ignore') - except subprocess.CalledProcessError as e: - raise ValueError("N_m3u8DL-RE (fallback) failed, aborting") - - expected_mkv_path = out.with_suffix('.mkv') - if expected_mkv_path.exists(): - if out.exists() and out.resolve() != expected_mkv_path.resolve(): - out.unlink() - expected_mkv_path.rename(out) - elif not out.exists(): - raise FileNotFoundError(f"N_m3u8DL-RE finished, but expected output file '{out}' was not found.") \ No newline at end of file diff --git a/wpgskd/utils/regex.py b/wpgskd/utils/regex.py deleted file mode 100644 index fcdb44a..0000000 --- a/wpgskd/utils/regex.py +++ /dev/null @@ -1,10 +0,0 @@ -import re - - -def find(pattern, string, group=None): - if group: - m = re.search(pattern, string) - if m: - return m.group(group) - else: - return next(iter(re.findall(pattern, string)), None) diff --git a/wpgskd/utils/sslciphers.py b/wpgskd/utils/sslciphers.py deleted file mode 100644 index e3eb210..0000000 --- a/wpgskd/utils/sslciphers.py +++ /dev/null @@ -1,77 +0,0 @@ -import ssl -from typing import Optional - -from requests.adapters import HTTPAdapter - - -class SSLCiphers(HTTPAdapter): - """ - Custom HTTP Adapter to change the TLS Cipher set and security requirements. - - Security Level may optionally be provided. A level above 0 must be used at all times. - A list of Security Levels and their security is listed below. Usually 2 is used by default. - Do not set the Security level via @SECLEVEL in the cipher list. - - Level 0: - Everything is permitted. This retains compatibility with previous versions of OpenSSL. - - Level 1: - The security level corresponds to a minimum of 80 bits of security. Any parameters - offering below 80 bits of security are excluded. As a result RSA, DSA and DH keys - shorter than 1024 bits and ECC keys shorter than 160 bits are prohibited. All export - cipher suites are prohibited since they all offer less than 80 bits of security. SSL - version 2 is prohibited. Any cipher suite using MD5 for the MAC is also prohibited. - - Level 2: - Security level set to 112 bits of security. As a result RSA, DSA and DH keys shorter - than 2048 bits and ECC keys shorter than 224 bits are prohibited. In addition to the - level 1 exclusions any cipher suite using RC4 is also prohibited. SSL version 3 is - also not allowed. Compression is disabled. - - Level 3: - Security level set to 128 bits of security. As a result RSA, DSA and DH keys shorter - than 3072 bits and ECC keys shorter than 256 bits are prohibited. In addition to the - level 2 exclusions cipher suites not offering forward secrecy are prohibited. TLS - versions below 1.1 are not permitted. Session tickets are disabled. - - Level 4: - Security level set to 192 bits of security. As a result RSA, DSA and DH keys shorter - than 7680 bits and ECC keys shorter than 384 bits are prohibited. Cipher suites using - SHA1 for the MAC are prohibited. TLS versions below 1.2 are not permitted. - - Level 5: - Security level set to 256 bits of security. As a result RSA, DSA and DH keys shorter - than 15360 bits and ECC keys shorter than 512 bits are prohibited. - """ - - def __init__(self, cipher_list: Optional[str] = None, security_level: int = 0, *args, **kwargs): - if cipher_list: - if not isinstance(cipher_list, str): - raise TypeError(f"Expected cipher_list to be a str, not {cipher_list!r}") - if "@SECLEVEL" in cipher_list: - raise ValueError("You must not specify the Security Level manually in the cipher list.") - if not isinstance(security_level, int): - raise TypeError(f"Expected security_level to be an int, not {security_level!r}") - if security_level not in range(6): - raise ValueError(f"The security_level must be a value between 0 and 5, not {security_level}") - - if not cipher_list: - # cpython's default cipher list differs to Python-requests cipher list - cipher_list = "DEFAULT" - - cipher_list += f":@SECLEVEL={security_level}" - - ctx = ssl.create_default_context() - ctx.check_hostname = False # For some reason this is needed to avoid a verification error - ctx.set_ciphers(cipher_list) - - self._ssl_context = ctx - super().__init__(*args, **kwargs) - - def init_poolmanager(self, *args, **kwargs): - kwargs["ssl_context"] = self._ssl_context - return super().init_poolmanager(*args, **kwargs) - - def proxy_manager_for(self, *args, **kwargs): - kwargs["ssl_context"] = self._ssl_context - return super().proxy_manager_for(*args, **kwargs) diff --git a/wpgskd/utils/subprocess.py b/wpgskd/utils/subprocess.py deleted file mode 100644 index 152e9f9..0000000 --- a/wpgskd/utils/subprocess.py +++ /dev/null @@ -1,29 +0,0 @@ -import json -import subprocess - - -def ffprobe(uri): - """Use ffprobe on the provided data to get stream information.""" - args = [ - "ffprobe", - "-v", "quiet", - "-of", "json", - "-show_streams" - ] - if isinstance(uri, str): - args.extend([ - "-f", "lavfi", - "-i", "movie={}[out+subcc]".format(uri.replace("\\", '/').replace(":", "\\\\:")) - ]) - elif isinstance(uri, bytes): - args.append("pipe:") - try: - ff = subprocess.run( - args, - input=uri if isinstance(uri, bytes) else None, - check=True, - capture_output=True - ) - except subprocess.CalledProcessError: - return {} - return json.loads(ff.stdout.decode("utf-8")) diff --git a/wpgskd/utils/ttml2ssa.py b/wpgskd/utils/ttml2ssa.py deleted file mode 100644 index 32290f2..0000000 --- a/wpgskd/utils/ttml2ssa.py +++ /dev/null @@ -1,1177 +0,0 @@ -# encoding: utf-8 -# -# -------------------------------------------- -# based on https://github.com/yuppity/ttml2srt -# -------------------------------------------- -# SPDX-License-Identifier: LGPL-2.1-or-later - -from __future__ import unicode_literals, absolute_import, division - -import re -import io -import os.path -import json -from collections import OrderedDict -from copy import deepcopy -import tempfile - -try: - from defusedxml import minidom # type: ignore -except: - from xml.dom import minidom - - -class TimestampConverter(object): - - def __init__(self, frame_rate=23.976, tick_rate=1): - self.tick_rate = tick_rate - self.frame_rate = frame_rate - - def timeexpr_to_ms(self, *args): - return self._timeexpr_to_ms(*args) - - def _timeexpr_to_ms(self, time_expr): - """Use the given time expression to get a matching conversion method - to overwrite self.timeexpr_to_ms() with. - """ - - self.timeexpr_to_ms = self.determine_ms_convfn(time_expr) - return self.timeexpr_to_ms(time_expr) - - def _hhmmss_to_ms(self, hh, mm, ss): - return hh * 3600 * 1000 + mm * 60 * 1000 + ss * 1000 - - def subrip_to_ms(self, timestamp): - """Desconstruct SubRip timecode down to milliseconds - """ - - hh, mm, ss, ms = re.split(r'[:,]', timestamp) - return int(int(hh) * 3.6e6 + int(mm) * 60000 + int(ss) * 1000 + int(ms)) - - def _metric_to_ms(self, metric_multiplier, metric_value): - return int(metric_multiplier * metric_value) - - def _ms_to_hhmmssms(self, ms): - hh = int(ms / 3.6e6) - mm = int((ms % 3.6e6) / 60000) - ss = int((ms % 60000) / 1000) - ms = int(ms % 1000) - - return hh, mm, ss, ms - - def ms_to_subrip(self, ms): - """Build SubRip timecode from milliseconds - """ - - hh, mm, ss, ms = self._ms_to_hhmmssms(ms) - return '{:02d}:{:02d}:{:02d},{:03d}'.format(hh, mm, ss, ms) - - def ms_to_ssa(self, ms): - """Build SSA/ASS timecode from milliseconds - """ - - hh, mm, ss, ms = self._ms_to_hhmmssms(ms) - return '{:01d}:{:02d}:{:02d}.{:02d}'.format(hh, mm, ss, int(ms / 10)) - - def frames_to_ms(self, frames): - """Convert frame count to ms - """ - - return int(int(frames) * (1000 / self.frame_rate)) - - def offset_frames_to_ms(self, time): - """Convert offset-time expression with f metric to milliseconds. - """ - - frames = float(time[:-1]) - return int(int(frames) * (1000 / self.frame_rate)) - - def offset_ticks_to_ms(self, time): - """Convert offset-time expression with t metric to milliseconds. - """ - - ticks = int(time[:-1]) - seconds = 1.0 / self.tick_rate - return (seconds * ticks) * 1000 - - def offset_hours_to_ms(self, time): - """Convert offset-time expression with h metric to milliseconds. - """ - - hours = float(time[:-1]) - return self._metric_to_ms(3.6e6, hours) - - def offset_minutes_to_ms(self, time): - """Convert offset-time expression with m metric to milliseconds. - """ - - return self._metric_to_ms(60 * 1000, float(time[:-1])) - - def offset_seconds_to_ms(self, time): - """Convert offset-time expression with s metric to milliseconds. - """ - - seconds = float(time[:-1]) - return self._metric_to_ms(1000, seconds) - - def offset_ms_to_ms(self, time): - """Convert offset-time expression with ms metric to milliseconds. - """ - - ms = int(time[:-2]) - return ms - - def fraction_timestamp_to_ms(self, timestamp): - """Convert hh:mm:ss.fraction to milliseconds - """ - - hh, mm, ss, fraction = re.split(r'[:.]', timestamp) - hh, mm, ss = [int(i) for i in (hh, mm, ss)] - # Resolution beyond ms is useless for our purposes - ms = int(fraction[:3]) - - return self._hhmmss_to_ms(hh, mm, ss) + ms - - def frame_timestamp_to_ms(self, timestamp): - """Convert hh:mm:ss:frames to milliseconds - - Will handle hh:mm:ss:frames.sub-frames by discarding the sub-frame part - """ - - hh, mm, ss, frames = [int(i) for i in timestamp.split('.')[0].split(':')] - hhmmss_ms = self._hhmmss_to_ms(hh, mm, ss) - ms = self.frames_to_ms(frames) - return hhmmss_ms + ms - - def determine_ms_convfn(self, time_expr): - """Determine approriate ms conversion fn to pass the time expression to. - - Args: - time_exrp (str): TTML time expression - - Return: - Conversion method (callable) - - Strips the time expression of digits and uses the resulting string as - a key to a dict of conversion methods. - """ - - # Map time expression delimiters to conversion methods. Saves - # us from having to exec multibranch code on each line but assumes all - # time expressions to be of the same form. - time_expr_fns = { - - # clock-time, no frames or fraction - # Example(s): "00:02:23" - '::': self.frame_timestamp_to_ms, - - # clock-time, frames - # Example(s): "00:02:23:12", "00:02:23:12.222" - ':::': self.frame_timestamp_to_ms, - ':::.': self.frame_timestamp_to_ms, - - # clock-time, fraction - # Example(s): "00:02:23.283" - '::.': self.fraction_timestamp_to_ms, - - # offset-time, hour metric - # Example(s): "1h", "1.232837372637h" - 'h': self.offset_hours_to_ms, - '.h': self.offset_hours_to_ms, - - # offset-time, minute metric - # Example(s): "1m", "13.72986323m" - 'm': self.offset_minutes_to_ms, - '.m': self.offset_minutes_to_ms, - - # offset-time, second metric - # Example(s): "1s", "113.2312312s" - 's': self.offset_seconds_to_ms, - '.s': self.offset_seconds_to_ms, - - # offset-time, millisecond metric - # Example(s): "1ms", "1000.1231231231223ms" - 'ms': self.offset_ms_to_ms, - '.ms': self.offset_ms_to_ms, - - # offset-time, frame metric - # Example(s): "100f" - 'f': self.offset_frames_to_ms, - '.f': self.offset_frames_to_ms, - - # offset-time, tick metric - # Example(s): "19298323t" - 't': self.offset_ticks_to_ms, - '.t': self.offset_ticks_to_ms, - - } - - try: - delims = ''.join([i for i in time_expr if not i.isdigit()]) - return time_expr_fns[delims] - except KeyError: - raise NotImplementedError( - 'Unknown timestamp format ("{}")'.format(time_expr)) - - -class Ttml2Ssa(object): - - VERSION = '0.3.8' - - TIME_BASES = [ - 'media', - 'smpte', - ] - - SCALE = { - 'NTSC2PAL' : 23.976/25, - 'PAL2NTSC' : 25/23.976, - 'NTSC2FILM' : 23.976/24, - 'PAL2FILM' : 25/24, - 'FILM2NTSC' : 24/23.976, - 'FILM2PAL' : 24/25 - } - - TOP_MARKER = '{\\an8}' - - def __init__(self, shift=0, source_fps=23.976, scale_factor=1, subtitle_language=None): - self.shift = shift - self.source_fps = source_fps - self.subtitle_language = subtitle_language - self.scale_factor = scale_factor - self.ssa_timestamp_min_sep = 200 - self.use_cosmetic_filter = True - self.use_language_filter = True - self.fix_amazon_errors = True - - self.allow_italics = True - self.allow_top_pos = True - - self.allow_timestamp_manipulation = True - self.fix_timestamp_collisions = True - self.fix_duplicated_entries = True - - try: - self.cache_directory = tempfile.gettempdir() # Fails on Android - self.cache_downloaded_subtitles = True - except: - self.cache_directory = None - self.cache_downloaded_subtitles = False - - self._styles = {} - self._italic_style_ids = [] - self._top_regions_ids = [] - - self._allowed_style_attrs = ( - 'color', - 'fontStyle', - 'fontWeight', - ) - - ## This variable stores the language ID from the xml file. - # But it may not exist or it may be wrong. - self.lang = None - - self.ssa_style = OrderedDict([ - ('Fontname', 'Arial'), - ('Fontsize', 50), - ('PrimaryColour', '&H00EEEEEE'), - ('SecondaryColour', '&H000000FF'), - ('BackColour', '&H40000000'), - ('OutlineColour', '&H00000000'), - ('Bold', 0), - ('Italic', 0), - ('Underline', 0), - ('Alignment', 2), - ('BorderStyle', 1), - ('Outline', 2), - ('Shadow', 3), - ('MarginL', 0), - ('MarginR', 0), - ('MarginV', 40), - ('StrikeOut', 0), - ('ScaleX', 100), - ('ScaleY', 100), - ('Spacing', 0), - ('Angle', 0), - ('Encoding', 1) - ]) - self.ssa_playresx = 1280 - self.ssa_playresy = 720 - - self.entries = [] - - def set_video_aspect_ratio(self, ratio): - """ Adjust the SSA options PlaResX and PlayRexY according to the aspect ratio of the video """ - self.ssa_playresy = int(self.ssa_playresx / ratio) - - def parse_subtitle_file(self, filename, file_encoding=None): - """Read and parse a subtitle file. - If the file has the vtt or srt extension it will be parsed as a vtt. Otherwise it will be parsed as ttml. - The result is stored in the `entries` list, as begin (ms), end (ms), text, position. - """ - - extension = os.path.splitext(filename)[1].lower() - if extension == ".srt" or extension == ".vtt": - self.parse_vtt_file(filename, file_encoding) - else: - self.parse_ttml_file(filename, file_encoding) - - def parse_ttml_file(self, filename, file_encoding=None): - """Read and parse a ttml/xml/dfxp file. - The result is stored in the `entries` list, as begin (ms), end (ms), text, position. - """ - - doc = self._read_file(filename, file_encoding) - self.parse_ttml_from_string(doc.encode('utf-8')) - - def parse_ttml_from_string(self, doc): - """Read and parse a ttml/xml/dfxp subtitle from a string. - The result is stored in the `entries` list, as begin (ms), end (ms), text, position. - """ - - def extract_rate(s): - try: - m = s.split(' ') - return int(m[0]) / int(m[1]) - except: - return 1 - - del self.entries [:] - self._tc = TimestampConverter() - - ttml_dom = minidom.parseString(doc) - self._encoding = ttml_dom.encoding - - if self._encoding and self._encoding.lower() not in ['utf8', 'utf-8']: - # Don't bother with subtitles that aren't utf-8 encoded - # but assume utf-8 when the encoding attr is missing - raise NotImplementedError('Source is not utf-8 encoded') - - # Get the root tt element (assume the file contains - # a single subtitle document) - tt_element = ttml_dom.getElementsByTagNameNS('*', 'tt')[0] - - # Extract doc language - # https://tools.ietf.org/html/rfc4646#section-2.1 - language_tag = tt_element.getAttribute('xml:lang') or '' - self.lang = re.split(r'\s+', language_tag.strip())[0].split('-')[0] - - # Store TT parameters as instance vars (in camel case) - opttime = {} - for ttp_name, defval, convfn in ( - # (tt param, default val, fn to process the str) - ('frameRate', 0, lambda x: float(x)), - ('tickRate', 0, lambda x: int(x)), - ('timeBase', 'media', lambda x: x), - ('clockMode', '', lambda x: x), - #('frameRateMultiplier', 1, lambda x: int(x)), - ('frameRateMultiplier', 1, lambda x: extract_rate(x)), - ('subFrameRate', 1, lambda x: int(x)), - ('markerMode', '', lambda x: x), - ('dropMode', '', lambda x: x), - ): - ttp_val = getattr( - tt_element.attributes.get('ttp:' + ttp_name), 'value', defval) - opttime[Ttml2Ssa._snake_to_camel(ttp_name)] = convfn(ttp_val) - - if opttime['time_base'] not in Ttml2Ssa.TIME_BASES: - raise NotImplementedError('No support for "{}" time base'.format( - opttime['time_base'])) - - # Set effective tick rate as per - # https://www.w3.org/TR/ttml1/#parameter-attribute-tickRate - # This will obviously only be made use of if we encounter offset-time - # expressions that have the tick metric. - self._tc.tick_rate = opttime['tick_rate'] - if not opttime['tick_rate'] and opttime['frame_rate']: - self._tc.tick_rate = int(opttime['frame_rate'] * opttime['sub_frame_rate']) - elif not opttime['tick_rate']: - self._tc.tick_rate = 1 - - # Set FPS to source_fps if no TT param - self._tc.frame_rate = opttime['frame_rate'] or self.source_fps - - # Grab