#!/usr/bin/python3 import sys import os import base64 import json import datetime import pytz import pkce # Add parent directory to path for o11 import sys.path.insert(0, os.path.dirname(os.path.dirname(os.path.abspath(__file__)))) import o11 # Parse command line parameters user = o11.parse_params(sys.argv, 'user') password = o11.parse_params(sys.argv, 'password') device = o11.parse_params(sys.argv, 'device') pin = o11.parse_params(sys.argv, 'pin') id = o11.parse_params(sys.argv, 'id') action = o11.parse_params(sys.argv, 'action') bind = o11.parse_params(sys.argv, 'bind') proxy = o11.parse_params(sys.argv, 'proxy') doh = o11.parse_params(sys.argv, 'doh') worker = o11.parse_params(sys.argv, 'worker') cdm = o11.parse_params(sys.argv, 'cdm') drm = o11.parse_params(sys.argv, 'drm') kid = o11.parse_params(sys.argv, 'kid') pssh = o11.parse_params(sys.argv, 'pssh') challenge = o11.parse_params(sys.argv, 'challenge') heartbeaturl = o11.parse_params(sys.argv, 'heartbeaturl') heartbeatparams = o11.parse_params(sys.argv, 'heartbeatparams') # Session setup o11Session = o11.session(bind=bind, proxy=proxy, worker=worker) req = o11Session.get_session() if doh != "": o11.dns(doh) if challenge == "cert": challenge = "CAQ=" # Configuration WVD_PATH = './WVD.wvd' authFile = '/MLB_auth.json' SCRIPT_DIR = os.path.abspath(os.path.dirname(__file__)) USER_AGENT = 'Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 (KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36' DEVICE_ID = 'dfec4af0-ef3f-4dc8-8899-2131de02d722' headers = { 'accept': 'application/json, text/plain, */*', 'authorization': '', 'content-type': 'application/json', 'origin': 'https://www.mlb.com', 'referer': 'https://www.mlb.com/', 'user-agent': USER_AGENT, 'x-client-name': 'WEB', 'x-client-version': '7.8.2', } token = '' def get_auth(): try: return json.load(open(SCRIPT_DIR + authFile)) except: return None def save_auth(auth_data): json.dump(auth_data, open(SCRIPT_DIR + authFile, 'w'), indent=2) def get_refresh_code(cookies, code_challenge): params = {'client_id': '0oap7wa857jcvPlZ5355', 'code_challenge': code_challenge, 'code_challenge_method': 'S256', 'nonce': 'nzo1v5p3D3k6Qj1XzZA7xrP8JqpEPSTainoZH7RHSlIOwezeXGhZBSpVxPLy4ObG', 'prompt': 'none', 'redirect_uri': 'https://www.mlb.com/login', 'response_mode': 'okta_post_message', 'response_type': 'code', 'state': 'Dbl98ij0D8FKvTnwBdBwPaecCwlTnTSZRnrg8QCCcqP8yJXtviS9fnlIe3pVA7uI', 'scope': 'openid email'} response = req.get('https://ids.mlb.com/oauth2/aus1m088yK07noBfh356/v1/authorize', params=params, cookies=cookies, headers={'Accept': 'text/html', 'Referer': 'https://www.mlb.com/', 'User-Agent': USER_AGENT}) return response.content.decode().split("data.code = \'")[-1].split('\'')[0].replace('\\x2D', '-') def token_refresh_request(code_verifier, code): data = {'client_id': '0oap7wa857jcvPlZ5355', 'redirect_uri': 'https://www.mlb.com/login', 'grant_type': 'authorization_code', 'code_verifier': code_verifier, 'code': code} response = req.post('https://ids.mlb.com/oauth2/aus1m088yK07noBfh356/v1/token', headers={'Accept': 'application/json', 'Content-Type': 'application/x-www-form-urlencoded', 'Origin': 'https://www.mlb.com', 'Referer': 'https://www.mlb.com/', 'User-Agent': USER_AGENT}, data=data) return response.json()['access_token'] def login(): global token print("logging in...", file=sys.stderr) auth = get_auth() if not auth: print("No auth found. Please add MLB auth cookies to auth file.", file=sys.stderr) print("Login to https://www.mlb.com/ and copy cookies from DevTools", file=sys.stderr) save_auth({}) sys.exit(1) try: code_verifier, code_challenge = pkce.generate_pkce_pair() code = get_refresh_code(auth, code_challenge) token = token_refresh_request(code_verifier, code) print("logged in successfully", file=sys.stderr) return token except Exception as e: print(f"Login failed: {e}", file=sys.stderr) sys.exit(1) def get_token(): global token if token: return token return login() def get_session_id(): session_headers = {**headers, 'authorization': 'Bearer ' + token} json_data = {'operationName': 'initSession', 'query': 'mutation initSession($device: InitSessionInput!, $clientType: ClientType!, $experience: ExperienceTypeInput) { initSession(device: $device, clientType: $clientType, experience: $experience) { deviceId sessionId entitlements { code } location { countryCode regionName zipCode latitude longitude } clientExperience features } }', 'variables': {'device': {'appVersion': '7.8.2', 'deviceFamily': 'desktop', 'knownDeviceId': DEVICE_ID, 'languagePreference': 'ENGLISH', 'manufacturer': 'Google Inc.', 'model': '', 'os': 'windows', 'osVersion': '10'}, 'clientType': 'WEB'}} response = req.post('https://media-gateway.mlb.com/graphql', headers=session_headers, json=json_data) try: return response.json()['data']['initSession']['sessionId'] except: return None def get_single(session_id, media_id): single_headers = {**headers, 'authorization': 'Bearer ' + token} json_data = {'operationName': 'initPlaybackSession', 'query': 'mutation initPlaybackSession($adCapabilities: [AdExperienceType] $mediaId: String! $deviceId: String! $sessionId: String! $quality: PlaybackQuality) { initPlaybackSession(adCapabilities: $adCapabilities mediaId: $mediaId deviceId: $deviceId sessionId: $sessionId quality: $quality) { playbackSessionId playback { url token expiration cdn } adScenarios { adParamsObj adScenarioType adExperienceType } adExperience { adExperienceTypes adEngineIdentifiers { name value } adsEnabled } heartbeatInfo { url interval } trackingObj } }', 'variables': {'adCapabilities': ['GOOGLE_STANDALONE_AD_PODS'], 'deviceId': DEVICE_ID, 'mediaId': media_id, 'quality': 'PLACEHOLDER', 'sessionId': session_id}} response = req.post('https://media-gateway.mlb.com/graphql', headers=single_headers, json=json_data) try: return response.json()['data']['initPlaybackSession']['playback']['url'] except: return None def do_action(): get_token() if action == "login": login() sys.exit() if action == "channels": output = {'Channels': []} ch_headers = {**headers, 'authorization': 'Bearer ' + token, 'accept': '*/*'} response = req.get('https://mastapi.mobile.mlbinfra.com/api/epg/v3/search', params={'exp': 'MLB'}, headers=ch_headers) try: data = response.json() for re in data.get('results', []): if re.get('gameData', {}).get('abstractGameState') == 'Live': home_name = re['gameData']['home']['teamAbbrv'] away_name = re['gameData']['away']['teamAbbrv'] for vf in re.get('videoFeeds', []): if vf.get('mediaState') == 'MEDIA_ON': title = f"{home_name} vs {away_name} [{vf.get('mediaFeedType', '')}]" channel = { 'Name': title, 'Mode': 'live', 'SessionManifest': True, 'ManifestScript': f"id={vf.get('mediaId', '')}", 'CdmType': 'none', 'UseCdm': False, 'Cdm': '', 'Video': 'best', 'OnDemand': True, 'SpeedUp': True, } output['Channels'].append(channel) print(json.dumps(output, indent=2)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "events": output = {'Events': []} print(json.dumps(output, indent=2)) elif action == "heartbeat": sys.exit() elif action == "manifest": try: media_id = id session_id = get_session_id() if not session_id: return "error" video_url = get_single(session_id, media_id) if not video_url: return "error" output = { "Cdn": [{"Name": "default", "ManifestUrl": video_url}], "ManifestUrl": video_url, "Headers": {"Manifest": {'User-Agent': USER_AGENT}, "Media": {'User-Agent': USER_AGENT}}, "Heartbeat": {"Url": '', "Params": '', "PeriodMs": 5*60*1000} } print(json.dumps(output)) except Exception as e: print(f"Error: {e}", file=sys.stderr) return "error" elif action == "cdm": print("MLB streams are typically unencrypted HLS", file=sys.stderr) else: print("invalid action: " + action, file=sys.stderr) if do_action() == "error": login() do_action()