#!/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)