Spaces:
Running
Running
| #!/usr/bin/env python3 | |
| """Channel confirmation / cookie-export regressions. No live services or real credentials.""" | |
| import json | |
| import os | |
| import pathlib | |
| import sys | |
| import tempfile | |
| import time | |
| import unittest | |
| from unittest.mock import patch | |
| ROOT = pathlib.Path(sys.argv[1]).resolve() if len(sys.argv) > 1 else pathlib.Path(__file__).parent | |
| sys.path.insert(0, str(ROOT)) | |
| os.environ.setdefault('PASSWORD', 'channel-test-password') | |
| os.environ.setdefault('DATA_DIR', tempfile.mkdtemp(prefix='channel-app-test-')) | |
| os.environ.setdefault('FFMPEG_PATH', '/bin/true') | |
| from youtube_manager import ManagedYouTube | |
| import tiktok_auth | |
| OLD = {'id': 'OLD_CHANNEL', 'title': 'Old channel'} | |
| NEW = {'id': 'NEW_CHANNEL', 'title': 'New channel'} | |
| KEY = 'alice:room1' | |
| class ChannelAPI: | |
| def __init__(self, channel=NEW): | |
| self.channel = dict(channel) | |
| self.calls = [] | |
| def configured(self): | |
| return True | |
| def call(self, resource, method='GET', params=None, body=None): | |
| self.calls.append((resource, method)) | |
| if resource != 'channels' or method != 'GET': | |
| raise AssertionError('Channel switching must not touch remote events or streams') | |
| return {'items': [{'id': self.channel['id'], 'snippet': {'title': self.channel['title']}}]} | |
| class ChannelTests(unittest.TestCase): | |
| def setUp(self): | |
| self.tmp = tempfile.TemporaryDirectory() | |
| self.addCleanup(self.tmp.cleanup) | |
| self.logs = [] | |
| self.mgr = ManagedYouTube('/bin/true', lambda u: {'status': 'offline', 'sources': []}, | |
| lambda u, m: self.logs.append((u, m)), lambda: ['alice'], | |
| pathlib.Path(self.tmp.name) / 'journal.json') | |
| self.mgr.api = ChannelAPI() | |
| self.mgr.journal.change(lambda d: d.update(channel=dict(OLD), enabled=True, | |
| streams={'alice': {'id': 'old-stream'}}, | |
| events={KEY: {'session_key': KEY, 'username': 'alice', 'room_id': 'room1', | |
| 'broadcast_id': 'old-broadcast', 'title': 'Old event'}}, | |
| history=[{'session_key': 'bob:oldroom', 'broadcast_id': 'older-broadcast'}])) | |
| self.mgr.destinations['alice'] = 'rtmps://example.invalid/old-test-destination' | |
| self.launches = [] | |
| self.mgr.launch = lambda: self.launches.append(True) | |
| def test_check_reports_new_without_adopting(self): | |
| before = self.mgr.journal.read() | |
| reported = self.mgr.check() | |
| self.assertEqual(reported, dict(NEW, previous=OLD)) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| self.assertEqual(self.mgr.api.calls, [('channels', 'GET')]) | |
| def test_unconfirmed_switch_is_still_fail_closed(self): | |
| before = self.mgr.journal.read() | |
| with self.assertRaisesRegex(RuntimeError, 'القناة المفوضة تغيرت'): | |
| self.mgr.check_channel() | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| def test_wrong_confirmation_does_not_modify_journal(self): | |
| before = self.mgr.journal.read() | |
| with self.assertRaisesRegex(ValueError, 'لا تطابق آخر فحص'): | |
| self.mgr.confirm_channel(OLD['id']) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| self.assertTrue(self.mgr.destinations) | |
| with self.assertRaises(ValueError): | |
| self.mgr.confirm_channel('') | |
| def test_confirm_stops_before_archiving_and_does_not_start(self): | |
| self.mgr.senders[KEY] = {'fake': True} | |
| stopped = [] | |
| def stop(key): | |
| stopped.append(self.mgr.journal.read()['events'][key]['username']) | |
| self.mgr.senders.pop(key) | |
| return True | |
| self.mgr.stop_sender = stop | |
| result = self.mgr.confirm_channel(NEW['id']) | |
| d = self.mgr.journal.read() | |
| self.assertEqual(result, NEW) | |
| self.assertEqual(d['channel'], NEW) | |
| self.assertFalse(d['enabled']) | |
| self.assertFalse(d['events']) | |
| self.assertFalse(d['streams']) | |
| self.assertFalse(self.mgr.destinations) | |
| self.assertEqual(stopped, ['alice']) | |
| self.assertFalse(self.launches) | |
| self.assertEqual(len(d['history']), 2) | |
| self.assertTrue(all(h['channel_id'] == OLD['id'] for h in d['history'])) | |
| archived = next(h for h in d['history'] if h['session_key'] == KEY) | |
| self.assertEqual(archived['broadcast_id'], 'old-broadcast') | |
| self.assertEqual(archived['outcome'], 'local_cancelled_channel_switch') | |
| self.assertEqual(self.mgr.api.calls, [('channels', 'GET')]) | |
| def test_failed_stop_keeps_old_channel_and_events(self): | |
| before = self.mgr.journal.read() | |
| self.mgr.senders[KEY] = {'fake': True} | |
| self.mgr.stop_sender = lambda key: False | |
| with self.assertRaisesRegex(RuntimeError, 'لم تُعتمد القناة الجديدة'): | |
| self.mgr.confirm_channel(NEW['id']) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| self.assertTrue(self.mgr.destinations) | |
| def test_wrong_first_confirmation_is_read_only(self): | |
| self.mgr.journal.change(lambda d: d.update(channel=None, enabled=False, events={}, streams={})) | |
| before = self.mgr.journal.read() | |
| with self.assertRaises(ValueError): | |
| self.mgr.confirm_channel(OLD['id']) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| def test_same_channel_confirmation_preserves_running_state(self): | |
| self.mgr.api.channel = dict(OLD) | |
| before = self.mgr.journal.read() | |
| self.assertEqual(self.mgr.confirm_channel(OLD['id']), OLD) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| self.assertTrue(self.mgr.destinations) | |
| self.assertFalse(self.launches) | |
| def test_explicit_enable_switches_then_starts_new_channel(self): | |
| self.mgr.enable(NEW['id']) | |
| d = self.mgr.journal.read() | |
| self.assertEqual(d['channel'], NEW) | |
| self.assertTrue(d['enabled']) | |
| self.assertFalse(d['events']) | |
| self.assertFalse(self.mgr.destinations) | |
| self.assertEqual(self.launches, [True]) | |
| def test_previous_channel_history_does_not_block_or_resume_room(self): | |
| self.mgr.confirm_channel(NEW['id']) | |
| old_entry = next(h for h in self.mgr.journal.read()['history'] if h['session_key'] == KEY) | |
| before_calls = list(self.mgr.api.calls) | |
| self.assertIsNone(self.mgr._history_entry(KEY)) | |
| self.assertTrue(self.mgr._republish_allowed(KEY, old_entry)) | |
| self.assertFalse(self.mgr._resume_archived_event('alice', {'room_id': 'room1'}, old_entry)) | |
| self.assertEqual(self.mgr.api.calls, before_calls) | |
| self.assertTrue(self.mgr.open_event('alice', {'room_id': 'room1', 'nickname': 'Alice'})) | |
| self.assertTrue(any(h['broadcast_id'] == 'old-broadcast' for h in self.mgr.journal.read()['history'])) | |
| def test_forget_history_only_for_current_channel(self): | |
| self.mgr.confirm_channel(NEW['id']) | |
| self.mgr.journal.change(lambda d: d['history'].append( | |
| {'session_key': KEY, 'channel_id': NEW['id'], 'broadcast_id': 'new-broadcast'})) | |
| self.assertEqual(self.mgr._history_entry(KEY)['broadcast_id'], 'new-broadcast') | |
| self.mgr._forget_history(KEY) | |
| self.assertIsNone(self.mgr._history_entry(KEY)) | |
| self.assertTrue(any(h['broadcast_id'] == 'old-broadcast' for h in self.mgr.journal.read()['history'])) | |
| def test_confirmation_http_endpoint_is_protected_and_separate_from_start(self): | |
| import importlib.util | |
| if importlib.util.find_spec('httpx') is None: | |
| self.skipTest('HTTP endpoint test needs the optional httpx test dependency') | |
| from fastapi.testclient import TestClient | |
| import app | |
| client = TestClient(app.app) | |
| self.addCleanup(client.close) | |
| before = self.mgr.journal.read() | |
| with patch.object(app, 'youtube', self.mgr): | |
| denied = client.post('/api/youtube/channel', json={'channel_id': NEW['id']}) | |
| self.assertEqual(denied.status_code, 401) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| auth = ('admin', os.environ['PASSWORD']) | |
| wrong = client.post('/api/youtube/channel', json={'channel_id': OLD['id']}, auth=auth) | |
| self.assertEqual(wrong.status_code, 400) | |
| self.assertEqual(self.mgr.journal.read(), before) | |
| accepted = client.post('/api/youtube/channel', json={'channel_id': NEW['id']}, auth=auth) | |
| self.assertEqual(accepted.status_code, 200) | |
| self.assertEqual(accepted.json()['channel'], NEW) | |
| self.assertFalse(self.mgr.journal.read()['enabled']) | |
| self.assertFalse(self.launches) | |
| class CookieExportTests(unittest.TestCase): | |
| SID = 'a' * 32 | |
| def test_json_export_filters_expired_foreign_and_unsafe_cookies(self): | |
| future = time.time() + 3600 | |
| items = [ | |
| {'name': 'sessionid', 'value': self.SID, 'domain': '.tiktok.com', 'expirationDate': future}, | |
| {'name': 'sid_guard', 'value': self.SID + '%7C1700000000%7C15552000', 'domain': '.tiktok.com'}, | |
| {'name': 'tt-target-idc', 'value': 'alisg', 'domain': '.tiktok.com'}, | |
| {'name': 'expired', 'value': 'old', 'domain': '.tiktok.com', 'expirationDate': time.time() - 1}, | |
| {'name': 'foreign', 'value': 'no', 'domain': 'tiktok.com.attacker.invalid'}, | |
| {'name': 'header_injection', 'value': 'no\r\nX-Header: x', 'domain': '.tiktok.com'}, | |
| {'name': 'wrong_path', 'value': 'no', 'domain': '.tiktok.com', 'path': '/other'}, | |
| ] | |
| sid, pairs = tiktok_auth.parse_token(json.dumps(items)) | |
| self.assertEqual(sid, self.SID) | |
| self.assertEqual(pairs['sessionid_ss'], self.SID) | |
| self.assertEqual(pairs['tt-target-idc'], 'alisg') | |
| for rejected in ('expired', 'foreign', 'header_injection', 'wrong_path'): | |
| self.assertNotIn(rejected, pairs) | |
| def test_www_cookie_wins_in_either_export_order(self): | |
| cookies = [{'name': 'sessionid', 'value': self.SID, 'domain': '.tiktok.com'}, | |
| {'name': 'msToken', 'value': 'specific==', 'domain': 'www.tiktok.com', 'hostOnly': True}, | |
| {'name': 'msToken', 'value': 'broad==', 'domain': '.tiktok.com'}] | |
| for values in (cookies, cookies[::-1]): | |
| sid, pairs = tiktok_auth.parse_token(json.dumps(values)) | |
| self.assertEqual(sid, self.SID) | |
| self.assertEqual(pairs['msToken'], 'specific==') | |
| def test_wrapped_json_and_malformed_export(self): | |
| sid, pairs = tiktok_auth.parse_token(json.dumps({'cookies': [ | |
| {'name': 'sessionid', 'value': self.SID, 'domain': '.tiktok.com'}]})) | |
| self.assertEqual(sid, self.SID) | |
| for text in ('[{broken]', '{}', '[1, null, "bad"]'): | |
| self.assertEqual(tiktok_auth.parse_token(text), ('', {})) | |
| def test_expired_session_cannot_synthesize_login(self): | |
| text = json.dumps([{'name': 'sessionid', 'value': self.SID, | |
| 'domain': '.tiktok.com', 'expirationDate': time.time() - 1}]) | |
| self.assertEqual(tiktok_auth.parse_token(text), ('', {})) | |
| def test_old_formats_still_supported(self): | |
| for token in (self.SID, 'sessionid=' + self.SID, | |
| '.tiktok.com\tTRUE\t/\tTRUE\t0\tsessionid\t' + self.SID): | |
| self.assertEqual(tiktok_auth.parse_token(token)[0], self.SID) | |
| sid, pairs = tiktok_auth.parse_token('sessionid=' + self.SID + '; msToken=extra==; tt-target-idc=alisg') | |
| self.assertEqual(sid, self.SID) | |
| self.assertEqual(pairs['msToken'], 'extra==') | |
| self.assertEqual(pairs['tt-target-idc'], 'alisg') | |
| if __name__ == '__main__': | |
| unittest.main(argv=[sys.argv[0]], verbosity=2) | |