"""MovieBox HTTP adapter. See UPSTREAM-LICENSE.txt. Signing format adapted from MovieBox-TUI crypto.rs at the revision in README. Sessions remain in memory; no media proxy or full downloads. """ import ast import base64 import hashlib import hmac import json import re import secrets import time import uuid from pathlib import Path from urllib.error import HTTPError, URLError from urllib.parse import urlsplit, parse_qsl from urllib.request import Request, urlopen class ProviderError(Exception): def __init__(self, message, status=502, retry_after=60): super().__init__(message) self.status = status self.retry_after = retry_after def fetch_json(url, headers=None, body=None): try: with urlopen(Request(url, data=body, headers=headers or {'Accept': 'application/json', 'User-Agent': 'Stremio/4.4.168 MovieBox-Addon/1.0'}), timeout=15) as response: raw = response.read(4 * 1024 * 1024 + 1) if len(raw) > 4 * 1024 * 1024: raise ProviderError('Response exceeds prototype limit') return json.loads(raw) except HTTPError as error: retry = error.headers.get('Retry-After', '60') retry = max(30, min(int(retry), 3600)) if retry.isdigit() else 60 raise ProviderError('Upstream HTTP ' + str(error.code), error.code, retry) from None except (URLError, TimeoutError, OSError, ValueError): raise ProviderError('Network, TLS, or JSON response failure') from None class MovieBox: def __init__(self): source = (Path(__file__).parent / 'reference-crypto.rs').read_text() self.key = ast.literal_eval(re.search(r'DEFAULT_SECRET_BYTES:.*?=\s*(b"[^"]+")', source).group(1)) self.cert = re.search(r'STOCK_CERT_SHA256:.*?=\s*"([a-f0-9]+)"', source).group(1) self.base = 'https://api6.aoneroom.com' self.token = None self.token_created = 0 self.ua = 'com.community.oneroom/50020121 (Linux; U; Android 13; en_US; 23078RKD5C; Build/TQ2A.230405.003; Cronet/135.0.7012.3)' self.info = dict(package_name='com.community.oneroom', version_name='4.0.01.0813.03', version_code=50020121, os='android', os_version='13', install_ch='ps', device_id=secrets.token_hex(16), install_store='ps', gaid=str(uuid.uuid4()), brand='Redmi', model='23078RKD5C', system_language='en', net='NETWORK_WIFI', region='US', timezone='America/Chicago', sp_code='40401', **{'X-Play-Mode': '2'}) def request(self, path, payload=None): url = self.base + path body = json.dumps(payload, separators=(',', ':'), sort_keys=True).encode() if payload is not None else None method = 'POST' if body is not None else 'GET' ts = str(int(time.time() * 1000)) parsed = urlsplit(url) query = '&'.join(k + '=' + v for k, v in sorted(parse_qsl(parsed.query))) canonical_url = parsed.path + ('?' + query if query else '') canonical = '\n'.join([method, 'application/json', 'application/json', str(len(body)) if body is not None else '', ts, hashlib.md5(body[:102400]).hexdigest() if body is not None else '', canonical_url]) signature = base64.b64encode(hmac.new(self.key, canonical.encode(), hashlib.md5).digest()).decode() stamp = ts + '000' + str(secrets.randbelow(1000)) build = stamp + '.' + hashlib.md5((self.cert + 'oneroom' + stamp).encode()).hexdigest() info = dict(self.info, **{'X-Client-Build': build}) headers = {'User-Agent': self.ua, 'Accept': 'application/json', 'Content-Type': 'application/json', 'x-client-token': ts + ',' + hashlib.md5(ts[::-1].encode()).hexdigest(), 'x-tr-signature': ts + '|2|' + signature, 'x-client-build': build, 'x-client-info': json.dumps(info, separators=(',', ':')), 'x-client-status': '0'} if self.token: headers['Authorization'] = 'Bearer ' + self.token response = fetch_json(url, headers, body) if response.get('code') not in (None, 0, '0', 200, '200'): raise ProviderError('Provider rejected the request') data = response.get('data', response) if not isinstance(data, dict): raise ProviderError('Unexpected API response shape') return data def initialize(self): if not self.token or time.monotonic() - self.token_created > 1800: self.token = None data = self.request('/wefeed-mobile-bff/user-api/visitor-login', {}) self.token = data.get('token') if not self.token: raise ProviderError('Guest session returned no token') self.token_created = time.monotonic() def search(self, title, kind=None, page=1): self.initialize() data = self.request('/wefeed-mobile-bff/subject-api/search/v2', dict(keyword=title, page=page, perPage=15, subjectType=0)) groups = data.get('results', []) return groups[0].get('subjects', []) if groups else data.get('list', []) def details(self, subject): self.initialize() if not str(subject).isdigit(): raise ProviderError('Invalid subject identifier') return self.request('/wefeed-mobile-bff/subject-api/get?subjectId=' + str(subject)) def seasons(self, subject): self.initialize() if not str(subject).isdigit(): raise ProviderError('Invalid subject identifier') return self.request('/wefeed-mobile-bff/subject-api/season-info?subjectId=' + str(subject)) def streams(self, subject, season=0, episode=0): self.initialize() if not str(subject).isdigit(): raise ProviderError('Invalid subject identifier') path = '/wefeed-mobile-bff/subject-api/play-info/v2?subjectId=' + str(subject) if season or episode: path += '&se=' + str(int(season)) + '&ep=' + str(int(episode)) data = self.request(path) return adapt_streams(data, self.ua) def adapt_streams(data, ua): results = [] for stream in data.get('streams', []): url = stream.get('url', '') cookie = stream.get('signCookie', '') for match in re.finditer(r'urlprefix=([^:;]+)', cookie): try: prefix = match[1].replace('-', '+').replace('_', '/') decoded = base64.b64decode(prefix + '=' * (-len(prefix) % 4)).decode().rstrip('*/') if '/dash/' in decoded and decoded.startswith(('https://', 'http://')): url = decoded + '/index.mpd' break except (ValueError, UnicodeError): continue if not url.startswith(('http://', 'https://')) or any(x in url.lower() for x in ['/notice.mp4', '1c7de0bd3393702d9191801f15f88f8d', '9a0461bc39da389663bf3dbb17091d3f', 'b164fbfb4347792950bdfbfb563d39d9']): continue if urlsplit(url).username or any(c in url for c in '\r\n') or any(c in cookie for c in '\r\n'): continue headers = {'User-Agent': ua, 'Referer': 'https://sportslive.wine'} if cookie: headers['Cookie'] = '; '.join(part.strip() for part in cookie.split(';') if part.strip()) resolutions = str(stream.get('resolutions', data.get('displayResolutions', 'Auto'))) codec = str(stream.get('codecName', stream.get('codec', stream.get('format', 'Auto')))) results.append({'name': 'MovieBox', 'title': str(data.get('title', 'MovieBox')) + ' | ' + resolutions + ' | ' + codec, 'url': url, 'behaviorHints': {'notWebReady': True, 'proxyHeaders': {'request': headers}}}) return results if __name__ == '__main__': client = MovieBox() try: client.initialize() print('Guest session: PASS (token kept in memory only)') for title in ['Night of the Living Dead', 'Sherlock']: items = client.search(title) print(json.dumps({'query': title, 'results': [{'id': str(i.get('subjectId', i.get('id', ''))), 'title': i.get('title'), 'type': i.get('subjectType'), 'year': i.get('releaseDate')} for i in items[:5]]})) except ProviderError as error: print('PROVIDER CHECK FAILED:', error) raise SystemExit(1)