Delete wpgskd/utils directory

wpgskd/utils/AtomicSQL.py

@@ -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

wpgskd/utils/BamSDK/__init__.py

@@ -1,2 +0,0 @@
-# flake8: noqa
-from wpgskd.utils.BamSDK.bamsdk import BamSdk

wpgskd/utils/BamSDK/bamsdk.py

@@ -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)

wpgskd/utils/BamSDK/services/__init__.py

@@ -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

wpgskd/utils/BamSDK/services/account.py

@@ -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()

wpgskd/utils/BamSDK/services/bamIdentity.py

@@ -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()

wpgskd/utils/BamSDK/services/content.py

@@ -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()

wpgskd/utils/BamSDK/services/device.py

@@ -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

wpgskd/utils/BamSDK/services/drm.py

@@ -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

wpgskd/utils/BamSDK/services/media.py

@@ -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()

wpgskd/utils/BamSDK/services/session.py

@@ -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()

wpgskd/utils/BamSDK/services/token.py

@@ -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()

wpgskd/utils/Logger.py

@@ -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

wpgskd/utils/MSL/MSLKeys.py

@@ -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

wpgskd/utils/MSL/MSLObject.py

@@ -1,6 +0,0 @@
-import jsonpickle
-
-
-class MSLObject:
-    def __repr__(self):
-        return "<{} {}>".format(self.__class__.__name__, jsonpickle.encode(self, unpicklable=False))

wpgskd/utils/MSL/MSL_ANDROID.py

@@ -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'(?<!^)(?=[A-Z])', '_', p).lower() for p in permissions}
-        for key in keys:
-            if key.type != "OPERATOR_SESSION":
-                continue
-            key_perms = {p.lower() for p in (getattr(key, "permissions", None) or [])}
-            if normalized_perms <= key_perms:
-                return key.key
-        return None
-
-    def send_message(
-        self,
-        endpoint: str,
-        params: Dict[str, str],
-        application_data: Dict[str, Any],
-        userauthdata: Optional[dict] = None,
-        headers: Optional[dict] = None,
-        proxy: Optional[Dict[str, str]] = None,
-    ) -> 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))

wpgskd/utils/MSL/__init__.py

@@ -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())

wpgskd/utils/MSL/schemes/EntityAuthentication.py

@@ -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
-            }
-        )

wpgskd/utils/MSL/schemes/KeyExchangeRequest.py

@@ -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}
-        )

wpgskd/utils/MSL/schemes/PlayReadyKeyExchangeScheme.py

@@ -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")

wpgskd/utils/MSL/schemes/UserAuthentication.py

@@ -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
-            }
-        )

wpgskd/utils/MSL/schemes/__init__.py

@@ -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"

wpgskd/utils/__init__.py

@@ -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))

wpgskd/utils/adobepass.py

@@ -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")),
-                }
-            )
-        )

wpgskd/utils/base64.py

@@ -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"==")

wpgskd/utils/click.py

@@ -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<season>\d+)(E(?P<episode>\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))

wpgskd/utils/collections.py

@@ -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

wpgskd/utils/drmtoday.py

@@ -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"
-}

wpgskd/utils/gen_esn.py

@@ -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

wpgskd/utils/io.py

@@ -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.")

wpgskd/utils/regex.py

@@ -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)

wpgskd/utils/sslciphers.py

@@ -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)

wpgskd/utils/subprocess.py

@@ -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"))

wpgskd/utils/xml.py

@@ -1,12 +0,0 @@
-from lxml import etree
-
-def load_xml(xml):
-    if not isinstance(xml, bytes):
-        xml = xml.encode("utf-8")
-    root = etree.fromstring(xml)
-    for elem in root.getiterator():
-        if not hasattr(elem.tag, "find"):
-            continue
-        elem.tag = etree.QName(elem).localname
-    etree.cleanup_namespaces(root)
-    return root