# Copyright (C) 2024–present Loren Eteval & contributors # # This file is part of Furious. # # This program is free software: you can redistribute it and/or modify # it under the terms of the GNU General Public License as published by # the Free Software Foundation, either version 3 of the License, or # (at your option) any later version. # # This program is distributed in the hope that it will be useful, # but WITHOUT ANY WARRANTY; without even the implied warranty of # MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the # GNU General Public License for more details. # # You should have received a copy of the GNU General Public License # along with this program. If not, see . """Exercise event-driven connection startup with real Qt event delivery.""" from __future__ import annotations from Furious.Controllers.SettingsController import SettingsController from Furious.Interface import CoreRuntime, RuntimeState from Furious.Plugins import PreparedRuntime, CoreRuntimeStartup from Furious.Qt.Signals import singleShotWeakly from Furious.Service.ConnectionManager import ( ConnectionManager, ConnectionStartStage, _ConditionProbe, ConnectionStartOperation, ) from Furious.Service.DnsResolver import DnsResolutionOperation, DnsResolver from Furious.Frozenlib import AppSettings from Furious.Core.Tun2socks import Tun2socks as Tun2socksRuntime from Furious.Widget.ConnectionProgressWidget import ConnectionProgressBar from PySide6 import QtCore, QtNetwork from shiboken6 import isValid, delete as deleteQObject from tests.support import application, processQtEvents, waitFor from unittest import TestCase, mock from types import SimpleNamespace from pathlib import Path import importlib import io import shlex import unittest import threading class _Configuration: """Provide the endpoint values used by manager-only tests.""" def httpProxy(self): """Return a deterministic local proxy endpoint.""" return '127.0.0.1:18080' def socksProxy(self): """Return a deterministic local SOCKS endpoint.""" return '127.0.0.1:18081' def remoteAddress(self): """Avoid DNS during mocked TUN tests.""" return '192.0.2.1' class _Runtime(CoreRuntime): """Provide an observable in-process stand-in for a core runtime.""" def __init__(self, exitCallback=None): """Initialize an idle runtime and lifecycle counters.""" super().__init__(exitCallback) self.alive = False self.startOptions = [] self.stopCount = 0 self.disposeCount = 0 @staticmethod def name(): """Return the fixture runtime name.""" return 'Async Fixture' @staticmethod def version(): """Return a fixture version.""" return '1.0' def start(self, _configuration=None, *_args, **kwargs): """Become live and retain launch options.""" self.startOptions.append(dict(kwargs)) self.alive = True self.setState(RuntimeState.Alive) return True def stop(self): """Stop this exact runtime.""" self.stopCount += 1 self.alive = False self.setState(RuntimeState.Exited) def dispose(self): """Record final resource disposal.""" self.disposeCount += 1 self._exitCallback = None def isRunning(self): """Return the controlled process-survival state.""" return self.alive def fail(self, exitcode=61, *, report=True): """Simulate an early child exit through the runtime event authority.""" self.alive = False self.setState(RuntimeState.Failed) if report: self.publishExit(self.interpretExit(exitcode)) class _Registry: """Return prepared launches in deterministic order.""" def __init__(self, launches): """Retain a launch queue.""" self.launches = list(launches) def createCoreRuntime(self, *_args, **_kwargs): """Return the next prepared runtime launch.""" launch = self.launches.pop(0) launch.runtime.bindExitCallback(_kwargs.get('exitCallback')) return launch class _ResolverFixture(QtCore.QObject): """Complete one recursive-resolution state through a real Qt timer.""" def __init__(self): """Initialize without a pending result.""" super().__init__() self.resultMap = None @staticmethod def _newResultMap(domain): """Return the minimum state required by the observer.""" return { 'domain': domain, 'depth': 0, 'error': False, 'reference': [], 'result': {}, } def _beginResolve(self, resultMap): """Schedule asynchronous completion.""" self.resultMap = resultMap resultMap['depth'] = 1 singleShotWeakly(0, self, '_complete') def _complete(self): """Publish one deterministic address through shared state.""" self.resultMap['result']['192.0.2.10'] = True self.resultMap['depth'] = 0 class ConnectionStartupAsyncTest(TestCase): """Verify readiness, cancellation, rollback, and compatibility.""" def testConditionCallbackCanDestroyItsProbe(self): """In-flight predicate delivery cannot use a deleted child timer.""" for _ in range(30): probe = _ConditionProbe(lambda: True, 'fixture') probe._predicate = lambda: (deleteQObject(probe), True)[1] probe.start() self.assertFalse(isValid(probe)) self.assertFalse(isValid(probe._timer)) def testConditionDeadlineCanBeDisabledWithoutChangingOrdinaryTimeouts(self): """Only explicitly unbounded observers survive a long pending condition.""" for timeout in (None, 10000): with self.subTest(timeout=timeout): owner = QtCore.QObject() ready, results = [False], [] probe = _ConditionProbe( lambda: ready[0], 'fixture', timeout=timeout, parent=owner ) probe.finished.connect(results.append) try: probe.start() probe._elapsed = mock.Mock() probe._elapsed.isValid.return_value = True probe._elapsed.elapsed.return_value = 86400000 probe._poll() if timeout is None: self.assertEqual(results, []) self.assertTrue(probe._timer.isActive()) ready[0] = True probe._poll() self.assertEqual(results, [True]) else: self.assertEqual(results, [False]) self.assertFalse(probe._timer.isActive()) finally: probe.cancel() deleteQObject(owner) self.assertFalse(isValid(probe)) self.assertFalse(isValid(probe._timer)) def testNativeDestructionDuringStagesReleasesUncommittedRuntime(self): """Qt destruction ends acquisition and rolls back before child deletion.""" for stage in (ConnectionStartStage.Preparing, ConnectionStartStage.Committing): for _ in range(20): manager = self._manager() runtime = _Runtime() registry = _Registry([PreparedRuntime(runtime)]) with mock.patch( 'Furious.Service.ConnectionManager.getPluginRegistry', return_value=registry, ), mock.patch('sys.excepthook') as qtErrors: operation = manager.startAsync(_Configuration(), '', deepcopy=False) operation.stageChanged.connect( lambda current: ( deleteQObject(operation) if current is stage else None ) ) operation.start() processQtEvents() qtErrors.assert_not_called() self.assertFalse(isValid(operation)) self.assertIsNone(manager._activeStartOperation) self.assertFalse(operation.attempt.leases) self.assertFalse(manager.runtimes) if stage is ConnectionStartStage.Committing: self.assertFalse(runtime.alive) self.assertEqual(runtime.disposeCount, 1) else: self.assertFalse(runtime.startOptions) def testTerminalObserversCanDestroyStartupOperation(self): """All terminal paths retire manager ownership after native deletion.""" for method, signalName in ( ('cancel', 'cancelled'), ('_fail', 'failed'), ('_commit', 'succeeded'), ): for _ in range(20): manager = self._manager() operation = manager.startAsync(_Configuration(), '', deepcopy=False) getattr(operation, signalName).connect( lambda *_args: deleteQObject(operation) ) with mock.patch('sys.excepthook') as qtErrors: getattr(operation, method)() processQtEvents() qtErrors.assert_not_called() self.assertFalse(isValid(operation)) self.assertIsNone(manager._activeStartOperation) def testNativeOperationDestructionRetainsARefusedRuntimeRelease(self): """Incomplete cleanup transfers to the surviving manager for retry.""" manager = self._manager() runtime = _Runtime() operation = manager.startAsync(_Configuration(), '', deepcopy=False) lease = operation.attempt.ownRuntime(runtime) runtime.start() stop = runtime.stop runtime.stop = mock.Mock(side_effect=RuntimeError('fixture refuses stop')) with mock.patch('sys.excepthook') as qtErrors: deleteQObject(operation) processQtEvents() qtErrors.assert_not_called() self.assertFalse(isValid(operation)) self.assertIsNone(manager._activeStartOperation) self.assertFalse(operation.attempt.leases) self.assertEqual(manager._pendingReleases, [lease]) self.assertTrue(runtime.alive) self.assertTrue(isValid(lease.router)) runtime.stop = stop manager.stopAll() processQtEvents() self.assertFalse(runtime.alive) self.assertFalse(manager._pendingReleases) self.assertFalse(isValid(lease.router)) def testStageListenerCancellationStopsNextAcquisitionOrCommit(self): """Cancel through real Qt stage signals before the next ownership boundary.""" for stage in ( ConnectionStartStage.Preparing, ConnectionStartStage.StartingPrimary, ConnectionStartStage.Committing, ): with self.subTest(stage=stage): manager = ConnectionManager() self.managers.append(manager) runtime = _Runtime() registry = _Registry([PreparedRuntime(runtime)]) with mock.patch.object( manager, '_prepareTUNPolicy', return_value=(False, False) ) as prepare, mock.patch( 'Furious.Service.ConnectionManager.getPluginRegistry', return_value=registry, ), mock.patch( 'sys.excepthook' ) as qtErrors: operation = manager.startAsync(_Configuration(), '', deepcopy=False) operation.stageChanged.connect( lambda current: operation.cancel() if current is stage else None ) operation.start() self.assertEqual(manager.runtimes, []) self.assertIsNone(manager._activeStartOperation) if stage is ConnectionStartStage.Preparing: prepare.assert_not_called() if stage is not ConnectionStartStage.Committing: self.assertEqual(runtime.startOptions, []) else: self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) processQtEvents() qtErrors.assert_not_called() def setUp(self): """Ensure a Qt application exists for real timer/socket delivery.""" self.app = application() # SettingsController owns TUN preference registration, including focused runs. self._previousTUNBackend = AppSettings.get('ApplicationTUNBackend') AppSettings.set('ApplicationTUNBackend', 'tun2socks') self.servers = [] self.managers = [] def tearDown(self): """Release every exact manager and local listener.""" for manager in self.managers: manager.cleanup() for server in self.servers: server.close() server.deleteLater() processQtEvents() AppSettings.set('ApplicationTUNBackend', self._previousTUNBackend) def _manager(self): """Return a manager with host TUN disabled at the test boundary.""" manager = ConnectionManager() manager._prepareTUNPolicy = mock.Mock(return_value=(False, False)) self.managers.append(manager) return manager def _server(self): """Listen on one real loopback endpoint.""" server = QtNetwork.QTcpServer(self.app) self.assertTrue( server.listen( QtNetwork.QHostAddress.SpecialAddress.LocalHost, 0, ) ) self.servers.append(server) return server def _operation(self, manager, launch): """Start one operation through an injected registry.""" registry = _Registry([launch]) patcher = mock.patch( 'Furious.Service.ConnectionManager.getPluginRegistry', return_value=registry, ) patcher.start() self.addCleanup(patcher.stop) return manager.startAsync( _Configuration(), 'Global', deepcopy=False, ) def testDefaultReadinessAllowsSlowStartupAndStillEnforcesDeadline(self): """A listener can become ready after 2.5 seconds, before the 10-second limit.""" for ready in (True, False): with self.subTest(ready=ready): server = self._server() port = server.serverPort() server.close() runtime = _Runtime() manager = self._manager() operation = self._operation( manager, PreparedRuntime( runtime, readiness=CoreRuntimeStartup(endpoint=f'127.0.0.1:{port}'), ), ) succeeded, failures = [], [] operation.succeeded.connect(succeeded.append) operation.failed.connect(lambda *args: failures.append(args)) self.assertTrue(waitFor(lambda: operation._readinessProbe is not None)) probe = operation._readinessProbe probe._elapsed = mock.Mock() probe._elapsed.isValid.return_value = True for elapsed in (3000, 9999): probe._elapsed.elapsed.return_value = elapsed probe._poll() self.assertFalse(operation._terminal) self.assertEqual(failures, []) if ready: self.assertTrue( server.listen( QtNetwork.QHostAddress.SpecialAddress.LocalHost, port ) ) self.assertTrue(waitFor(lambda: bool(succeeded))) self.assertEqual(failures, []) self.assertEqual(manager.runtimes, [runtime]) else: probe._elapsed.elapsed.return_value = 10000 probe._poll() self.assertEqual(len(failures), 1) self.assertEqual(failures[0][2], 'core readiness check timed out') self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) def testLocalEndpointReadinessCompletesEarlyAndKeepsQtResponsive(self): """Commit as soon as a real listener accepts while timers still run.""" server = self._server() endpoint = f'127.0.0.1:{server.serverPort()}' runtime = _Runtime() launch = PreparedRuntime( runtime, readiness=CoreRuntimeStartup(endpoint=endpoint, timeout=1500), ) manager = self._manager() operation = self._operation(manager, launch) succeeded = [] timerEvents = [] operation.succeeded.connect(succeeded.append) QtCore.QTimer.singleShot(0, lambda: timerEvents.append(True)) self.assertEqual(operation.stage, ConnectionStartStage.Pending) self.assertTrue(waitFor(lambda: bool(succeeded), timeout=0.5)) self.assertEqual(timerEvents, [True]) self.assertEqual(manager.runtimes, [runtime]) self.assertEqual(runtime.startOptions[0], {}) def testReadinessTimeoutRollsBackTheExactRuntime(self): """Fail a live process whose promised local endpoint never appears.""" server = self._server() port = server.serverPort() server.close() runtime = _Runtime() manager = self._manager() operation = self._operation( manager, PreparedRuntime( runtime, readiness=CoreRuntimeStartup( endpoint=f'127.0.0.1:{port}', timeout=60, retryInterval=5, ), ), ) failures = [] operation.failed.connect(lambda *_args: failures.append(_args)) self.assertTrue(waitFor(lambda: bool(failures))) self.assertEqual(failures[0][1], 'Failed to start core') self.assertEqual(failures[0][2], 'core readiness check timed out') self.assertEqual(manager.lastStartError, 'Failed to start core') self.assertEqual(manager.runtimes, []) self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) def testObservedConfigurationExitPreservesSemanticFailure(self): """Let the runtime's typed exit remain the sole terminal authority.""" runtime = _Runtime() manager = self._manager() operation = self._operation( manager, PreparedRuntime( runtime, readiness=CoreRuntimeStartup( endpoint='127.0.0.1:1', timeout=500, retryInterval=5, ), ), ) failures = [] operation.failed.connect(lambda *_args: failures.append(_args)) processQtEvents() runtime.fail(CoreRuntime.ExitCode.ConfigurationError.value) self.assertTrue(waitFor(lambda: bool(failures))) processQtEvents(5) self.assertEqual(len(failures), 1) self.assertIs(failures[0][0], operation) self.assertEqual(failures[0][1], 'Invalid server configuration') self.assertEqual( failures[0][2], 'Async Fixture exited during waiting-primary with code 23; ' 'reason=invalid-configuration', ) self.assertEqual(manager.lastStartError, 'Invalid server configuration') self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) def testQueuedConfigurationExitOutranksReadinessTimeout(self): """Preserve code 23 when timeout delivery races a queued runtime event.""" runtime = _Runtime() manager = self._manager() operation = self._operation( manager, PreparedRuntime( runtime, readiness=CoreRuntimeStartup( endpoint='127.0.0.1:1', timeout=500, retryInterval=5, ), ), ) failures = [] operation.failed.connect(lambda *_args: failures.append(_args)) processQtEvents() runtime.fail(CoreRuntime.ExitCode.ConfigurationError.value) operation._readinessProbe._finishFailed('core readiness check timed out') self.assertTrue(waitFor(lambda: bool(failures))) processQtEvents(5) self.assertEqual(len(failures), 1) self.assertEqual(failures[0][1], 'Invalid server configuration') self.assertEqual( failures[0][2], 'Async Fixture exited during waiting-primary with code 23; ' 'reason=invalid-configuration', ) self.assertEqual(manager.lastStartError, 'Invalid server configuration') self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) def testEarlyRuntimeExitFailsOnceAndIgnoresLateProbeEvents(self): """Let process termination own the terminal result before timeout.""" runtime = _Runtime() manager = self._manager() operation = self._operation( manager, PreparedRuntime( runtime, readiness=CoreRuntimeStartup( endpoint='127.0.0.1:1', timeout=500, ), ), ) failures = [] operation.failed.connect(lambda *_args: failures.append(_args)) processQtEvents() singleShotWeakly(0, runtime, 'fail') self.assertTrue(waitFor(lambda: bool(failures))) processQtEvents(5) self.assertEqual(len(failures), 1) self.assertEqual(failures[0][1], 'Core terminated unexpectedly') self.assertEqual( failures[0][2], 'Async Fixture exited during waiting-primary with code 61; ' 'reason=unexpected', ) self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) def testCancellationAndReplacementCannotCommitAStaleGeneration(self): """Cancel the first generation before a second ready launch commits.""" server = self._server() first = _Runtime() second = _Runtime() manager = self._manager() registry = _Registry( [ PreparedRuntime( first, readiness=CoreRuntimeStartup( endpoint='127.0.0.1:1', timeout=5000, ), ), PreparedRuntime( second, readiness=CoreRuntimeStartup( endpoint=f'127.0.0.1:{server.serverPort()}', timeout=500, ), ), ] ) with mock.patch( 'Furious.Service.ConnectionManager.getPluginRegistry', return_value=registry, ): firstOperation = manager.startAsync( _Configuration(), 'Global', deepcopy=False, ) cancelled = [] firstOperation.cancelled.connect(cancelled.append) processQtEvents() secondOperation = manager.startAsync( _Configuration(), 'Global', deepcopy=False, ) succeeded = [] secondOperation.succeeded.connect(succeeded.append) self.assertTrue(waitFor(lambda: bool(succeeded))) self.assertEqual(cancelled, [firstOperation]) self.assertEqual(first.stopCount, 1) self.assertEqual(first.disposeCount, 1) self.assertEqual(manager.runtimes, [second]) def testPreparedLaunchWithoutReadinessStartsAndCommitsDirectly(self): """Start a runtime without inventing an endpoint-readiness policy.""" runtime = _Runtime() manager = self._manager() operation = self._operation( manager, PreparedRuntime(runtime), ) succeeded = [] operation.succeeded.connect(succeeded.append) self.assertTrue(waitFor(lambda: bool(succeeded))) self.assertEqual(runtime.startOptions, [{}]) self.assertEqual(manager.runtimes, [runtime]) def testDnsStartRespectsCancellationAndDestructionDuringRequestCreation(self): """Request hooks cannot restart a cancelled or destroyed observer's timer.""" for action in ('cancel', 'destroy'): with self.subTest(action=action): for _ in range(20): resolver = mock.Mock() resolver._newResultMap = DnsResolver._newResultMap operation = DnsResolutionOperation(resolver, 'example.test') results = [] destroyed = [] operation.finished.connect(lambda *_args: results.append(True)) operation.destroyed.connect(lambda *_args: destroyed.append(True)) def beginResolve(resultMap): resultMap['depth'] = 1 if action == 'cancel': operation.cancel() else: deleteQObject(operation) resolver._beginResolve.side_effect = beginResolve try: with mock.patch('sys.excepthook') as qtErrors: operation.start() processQtEvents() qtErrors.assert_not_called() self.assertFalse(results) if action == 'cancel': self.assertTrue(isValid(operation)) self.assertFalse(operation._timer.isActive()) else: self.assertFalse(isValid(operation)) self.assertFalse(isValid(operation._timer)) self.assertEqual(destroyed, [True]) finally: if isValid(operation): deleteQObject(operation) def testDnsTimeoutAbortCanDestroyItsObserver(self): """Abort listeners may destroy the observer before terminal publication.""" class PendingReply(QtNetwork.QNetworkReply): def abort(self): self.setFinished(True) self.finished.emit() def readData(self, _maximumLength): return bytes() for _ in range(20): resolver = mock.Mock() resolver._newResultMap = DnsResolver._newResultMap operation = DnsResolutionOperation(resolver, 'example.test', timeout=1) reply = PendingReply() results = [] destroyed = [] operation.finished.connect(lambda *_args: results.append(True)) operation.destroyed.connect(lambda *_args: destroyed.append(True)) operation._resultMap['depth'] = 1 operation._resultMap['reference'].append(reply) operation._elapsed = mock.Mock() operation._elapsed.elapsed.return_value = 1 reply.finished.connect(lambda: deleteQObject(operation)) try: with ( self.assertLogs('Furious.Service.DnsResolver', level='ERROR'), mock.patch('sys.excepthook') as qtErrors, ): operation._poll() processQtEvents() qtErrors.assert_not_called() self.assertTrue(reply.isFinished()) self.assertFalse(results) self.assertEqual(destroyed, [True]) self.assertFalse(isValid(operation)) self.assertFalse(isValid(operation._timer)) finally: if isValid(operation): deleteQObject(operation) deleteQObject(reply) def testDnsCancellationAndTimeoutIgnoreAlreadyDestroyedReplies(self): """Earlier recursive replies may be deleted before later requests stop.""" class PendingReply(QtNetwork.QNetworkReply): def __init__(self): super().__init__() self.abortCount = 0 def abort(self): self.abortCount += 1 self.setFinished(True) for synchronous in (False, True): with self.subTest(synchronous=synchronous): resolver = mock.Mock() resolver._newResultMap = DnsResolver._newResultMap operation = DnsResolutionOperation(resolver, 'example.test') completed = PendingReply() pending = PendingReply() operation._resultMap['reference'] = [completed, pending] operation._resultMap['depth'] = 1 completed.deleteLater() processQtEvents() try: if synchronous: with self.assertLogs( 'Furious.Service.DnsResolver', level='ERROR' ): DnsResolver.wait(operation._resultMap, timeout=0) else: operation.cancel() operation.cancel() self.assertEqual(pending.abortCount, 1) finally: pending.deleteLater() operation.deleteLater() processQtEvents() def testDnsResolutionOperationCompletesAndCancelsWithoutNestedWait(self): """Observe recursive DNS state through timers and suppress stale cancel.""" resolver = _ResolverFixture() operation = DnsResolutionOperation( resolver, 'example.test', timeout=200, ) results = [] operation.finished.connect( lambda error, addresses: results.append((error, addresses)) ) operation.start() self.assertTrue(waitFor(lambda: bool(results))) self.assertEqual(results, [(False, ['192.0.2.10'])]) cancelled = DnsResolutionOperation( resolver, 'cancelled.test', timeout=200, ) staleResults = [] cancelled.finished.connect( lambda error, addresses: staleResults.append((error, addresses)) ) cancelled.start() cancelled.cancel() processQtEvents(5) self.assertEqual(staleResults, []) operation.deleteLater() cancelled.deleteLater() resolver.deleteLater() def testLinuxTunPreservesDeviceBeforeRuntimeOrderingWithoutNestedWait(self): """Create and observe the Linux device before launching tun2socks.""" module = importlib.import_module('Furious.Service.ConnectionManager') server = self._server() primary = _Runtime() tun = _Runtime() tun.cleanup = None events = [] deviceChecks = 0 def tunFactory(*_args, **kwargs): tun.bindExitCallback(kwargs.get('exitCallback')) return tun originalTunStart = tun.start def startTun(*args, **kwargs): events.append('tun-start') return originalTunStart(*args, **kwargs) tun.start = startTun def findDevice(_name): nonlocal deviceChecks deviceChecks += 1 events.append('find-device') return deviceChecks >= 2 def executeScript(*_args, **_kwargs): events.append('script') return True manager = ConnectionManager() manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) self.managers.append(manager) registry = _Registry( [ PreparedRuntime( primary, readiness=CoreRuntimeStartup( endpoint=f'127.0.0.1:{server.serverPort()}', timeout=500, ), ) ] ) shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5) with ( mock.patch.object(module, 'PLATFORM', 'Linux'), mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), mock.patch.object( module, 'CoreRuntimeStartup', return_value=shortSurvival, ), mock.patch.object( module, 'getPluginRegistry', return_value=registry, ), mock.patch.object( module.SystemRoutingTable, 'managedRoutes', [], ), mock.patch.object( module.SystemRoutingTable, 'getDefaultGateway', return_value=[('192.0.2.254', 'eth0')], ), mock.patch.object( module.SystemRoutingTable, 'LinuxFindTUNDevice', side_effect=findDevice, ), mock.patch.object( module.SystemRoutingTable, 'LinuxGetIpRoute', return_value='', ), mock.patch.object( module.SystemRoutingTable, 'LinuxExecutePrivilegedScript', side_effect=executeScript, ), mock.patch.object( module.SystemRoutingTable, 'LinuxDeleteTUNDevice', ), mock.patch.object( module.SystemRoutingTable, 'deleteRelations', ), mock.patch.object( module.SystemRuntime, 'flatpakID', return_value='', ), mock.patch.object( module.PySide6Legacy, 'eventLoopWait', side_effect=AssertionError('nested event loop used'), ), ): operation = manager.startAsync( _Configuration(), 'Global', deepcopy=False, ) succeeded = [] operation.succeeded.connect(succeeded.append) self.assertTrue(waitFor(lambda: bool(succeeded))) self.assertEqual( events[:4], ['find-device', 'script', 'find-device', 'tun-start'], ) self.assertEqual(manager.runtimes, [primary, tun]) self.assertTrue(manager.usesApplicationTUN()) self.assertEqual(tun.startOptions[0], {}) def testWindowsTunStartsRuntimeBeforeObservingAndMutatingDevice(self): """Keep the Windows launch, device, then host-mutation sequence.""" module = importlib.import_module('Furious.Service.ConnectionManager') server = self._server() primary = _Runtime() tun = _Runtime() tun.cleanup = None events = [] def tunFactory(*_args, **kwargs): tun.bindExitCallback(kwargs.get('exitCallback')) return tun originalTunStart = tun.start def startTun(*args, **kwargs): events.append('tun-start') return originalTunStart(*args, **kwargs) tun.start = startTun def findDevice(_name): events.append('find-device') return True def addRelations(): events.append('add-relations') manager = ConnectionManager() manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) self.managers.append(manager) registry = _Registry( [ PreparedRuntime( primary, readiness=CoreRuntimeStartup( endpoint=f'127.0.0.1:{server.serverPort()}', timeout=500, ), ) ] ) with ( mock.patch.object(module, 'PLATFORM', 'Windows'), mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), mock.patch.object( module, 'getPluginRegistry', return_value=registry, ), mock.patch.object( module.SystemRoutingTable, 'managedRoutes', [], ), mock.patch.object( module.SystemRoutingTable, 'delete', ), mock.patch.object( module.SystemRoutingTable, 'getDefaultGateway', return_value=[('192.0.2.254', '192.0.2.10')], ), mock.patch.object( module.SystemRoutingTable, 'WIN32IpconfigFindContent', side_effect=findDevice, ), mock.patch.object( module.SystemRoutingTable, 'WIN32GetInterfaceAliasByIP', return_value='Ethernet', ), mock.patch.object( module.SystemRoutingTable, 'WIN32SetInterfaceDNS', ), mock.patch.object( module.SystemRoutingTable, 'WIN32FlushDNSCache', ), mock.patch.object( module.SystemRoutingTable, 'setDeviceGateway', ), mock.patch.object( module.SystemRoutingTable, 'addRelations', side_effect=addRelations, ), mock.patch.object( module.SystemRoutingTable, 'deleteRelations', ), mock.patch.object( module.PySide6Legacy, 'eventLoopWait', side_effect=AssertionError('nested event loop used'), ), ): operation = manager.startAsync( _Configuration(), 'Global', deepcopy=False, ) succeeded = [] operation.succeeded.connect(succeeded.append) self.assertTrue(waitFor(lambda: bool(succeeded))) self.assertEqual( events[:3], ['tun-start', 'find-device', 'add-relations'], ) self.assertEqual(manager.runtimes, [primary, tun]) self.assertTrue(manager.usesApplicationTUN()) def testDarwinTunSurvivalPrecedesDnsAndRouteMutation(self): """Observe tun2socks survival before applying macOS host networking.""" module = importlib.import_module('Furious.Service.ConnectionManager') server = self._server() primary = _Runtime() tun = _Runtime() tun.cleanup = None events = [] def tunFactory(*_args, **kwargs): tun.bindExitCallback(kwargs.get('exitCallback')) return tun originalTunStart = tun.start def startTun(*args, **kwargs): events.append('tun-start') return originalTunStart(*args, **kwargs) tun.start = startTun def dnsServers(): events.append('read-dns') return [('Wi-Fi', ['192.0.2.53'])] manager = ConnectionManager() manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) self.managers.append(manager) registry = _Registry( [ PreparedRuntime( primary, readiness=CoreRuntimeStartup( endpoint=f'127.0.0.1:{server.serverPort()}', timeout=500, ), ) ] ) shortSurvival = CoreRuntimeStartup(timeout=20, retryInterval=5) with ( mock.patch.object(module, 'PLATFORM', 'Darwin'), mock.patch.object(module, 'Tun2socks', side_effect=tunFactory), mock.patch.object( module, 'CoreRuntimeStartup', return_value=shortSurvival, ), mock.patch.object( module, 'getPluginRegistry', return_value=registry, ), mock.patch.object( module.SystemRoutingTable, 'managedRoutes', [], ), mock.patch.object( module.SystemRoutingTable, 'getDefaultGateway', return_value=['192.0.2.254'], ), mock.patch.object( module.SystemRoutingTable, 'DarwinGetDNSServers', side_effect=dnsServers, ), mock.patch.object( module.SystemRoutingTable, 'DarwinSetDNSServers', ), mock.patch.object( module.SystemRoutingTable, 'setDeviceGateway', ), mock.patch.object( module.SystemRoutingTable, 'addRelations', ), mock.patch.object( module.SystemRoutingTable, 'deleteRelations', ), mock.patch.object( module.PySide6Legacy, 'eventLoopWait', side_effect=AssertionError('nested event loop used'), ), ): operation = manager.startAsync( _Configuration(), 'Global', deepcopy=False, ) succeeded = [] operation.succeeded.connect(succeeded.append) self.assertTrue(waitFor(lambda: bool(succeeded))) self.assertEqual(events, ['tun-start', 'read-dns']) self.assertEqual(manager.runtimes, [primary, tun]) self.assertTrue(manager.usesApplicationTUN()) def testRepeatedCancellationReleasesOperationTimersAndRuntimes(self): """Keep repeated startup cancellation bounded and independently owned.""" manager = self._manager() for _index in range(12): runtime = _Runtime() operation = self._operation( manager, PreparedRuntime( runtime, readiness=CoreRuntimeStartup( endpoint='127.0.0.1:1', timeout=5000, ), ), ) processQtEvents() self.assertTrue(manager.cancelStart(operation)) processQtEvents() self.assertEqual(runtime.stopCount, 1) self.assertEqual(runtime.disposeCount, 1) self.assertIsNone(manager._activeStartOperation) class LinuxTun2socksScriptTest(TestCase): def testProgressContinuesDuringAuthorizationAndCancellationKeepsWorkerOwned(self): """A real progress timer advances while the helper waits off the GUI thread.""" application() module = importlib.import_module('Furious.Service.ConnectionManager') runtimeModule = importlib.import_module('Furious.Core.Tun2socks') class ProcessFreeTun(Tun2socksRuntime): def start(self): raise AssertionError('a cancelled startup must not start forwarding') entered, released = threading.Event(), threading.Event() guiThread = threading.get_ident() scriptPath = [] primary = _Runtime() registry = _Registry([PreparedRuntime(primary)]) manager = ConnectionManager() manager._prepareTUNPolicy = mock.Mock(return_value=(False, True)) bar = ConnectionProgressBar() bar.setValue(0) def execute(path, **kwargs): self.assertNotEqual(threading.get_ident(), guiThread) scriptPath.append(Path(path)) entered.set() if not released.wait(5): raise RuntimeError('fixture authorization was not released') self.assertTrue(Path(path + '.cancelled').exists()) return False with ( mock.patch.object(module, 'PLATFORM', 'Linux'), mock.patch.object(runtimeModule, 'PLATFORM', 'Linux'), mock.patch.object(module, 'Tun2socks', ProcessFreeTun), mock.patch.object(module, 'getPluginRegistry', return_value=registry), mock.patch.object(module.SystemRoutingTable, 'managedRoutes', []), mock.patch.object( module.SystemRoutingTable, 'getDefaultGateway', return_value=[('192.0.2.254', 'eth0')], ), mock.patch.object( module.SystemRoutingTable, 'LinuxFindTUNDevice', return_value=False ), mock.patch.object( module.SystemRoutingTable, 'LinuxGetIpRoute', return_value='' ), mock.patch.object( module.SystemRoutingTable, 'LinuxExecutePrivilegedScript', side_effect=execute, ), mock.patch.object(module.SystemRoutingTable, 'LinuxDeleteTUNDevice'), mock.patch.object(module.SystemRoutingTable, 'deleteRelations'), mock.patch.object(module.SystemRuntime, 'flatpakID', return_value=''), mock.patch('sys.excepthook') as qtErrors, ): try: operation = manager.startAsync( _Configuration(), 'Global', deepcopy=False ) bar.start(1) self.assertTrue(waitFor(entered.is_set)) self.assertTrue(waitFor(lambda: bar.value() > 0)) self.assertFalse(operation._terminal) self.assertTrue(scriptPath[0].exists()) self.assertIsNone(operation._conditionProbe._timeout) tun = operation._tun operation.cancel() cleanup = manager.stopAsync() cleanup.start() self.assertFalse(cleanup._terminal) self.assertTrue(tun.cleanupPending) self.assertTrue(scriptPath[0].exists()) released.set() self.assertTrue(waitFor(lambda: cleanup._terminal)) self.assertFalse(tun.cleanupPending) self.assertFalse(scriptPath[0].exists()) self.assertFalse(Path(str(scriptPath[0]) + '.cancelled').exists()) self.assertEqual(manager._pendingReleases, []) finally: released.set() cleanup = manager.stopAsync() cleanup.start() self.assertTrue(waitFor(lambda: cleanup._terminal)) manager.cleanup() bar.stop() bar.deleteLater() processQtEvents() qtErrors.assert_not_called() def testBothStartupPathsQuoteArgumentsWithoutChangingRouteDetection(self): module = importlib.import_module('Furious.Service.ConnectionManager') interface = "eth'$x" device = "tun'$x" source, gateway = '192.0.2.1', '192.0.2.254' for asynchronous in (True, False): for existing in (True, False): with self.subTest(asynchronous=asynchronous, existing=existing): class ScriptFile(io.StringIO): def __exit__(self, *args): return False file = ScriptFile() file.name = 'tmpfixture.sh' scripts = [] tun = _Runtime() rawRoute = f'{source} via {gateway} dev {interface}' def executeScript(_path, **kwargs): scripts.append(file.getvalue()) self.assertEqual(kwargs['shell'], 'bash') return True with ( mock.patch.object(module, 'PLATFORM', 'Linux'), mock.patch.object( module, 'APPLICATION_TUN2SOCKS_DEVICE_NAME', device ), mock.patch.object(module, 'Tun2socks', return_value=tun), mock.patch.object( module.tempfile, 'NamedTemporaryFile', return_value=file ), mock.patch.object( module.SystemRuntime, 'flatpakID', return_value='' ), mock.patch.object( module.SystemRoutingTable, 'getDefaultGateway', return_value=[(gateway, interface)], ), mock.patch.object( module.SystemRoutingTable, 'managedRoutes', [] if not asynchronous else [[source, gateway]], ), mock.patch.object( module.SystemRoutingTable, 'LinuxFindTUNDevice', return_value=False, ), mock.patch.object( module.SystemRoutingTable, 'LinuxGetIpRoute', return_value=rawRoute if existing else '', ), mock.patch.object( module.SystemRoutingTable, 'LinuxExecutePrivilegedScript', side_effect=executeScript, ), ): if asynchronous: operation = SimpleNamespace( _isCurrent=lambda: True, _setStage=mock.Mock(), _interface=interface, _tun=tun, _waitForTunDevice=mock.Mock(), _observeCondition=mock.Mock(), _fail=mock.Mock(), ) ConnectionStartOperation._prepareLinuxTun(operation) self.assertTrue(waitFor(tun.linuxPreparation.workCompleted)) operation._fail.assert_not_called() else: attempt = SimpleNamespace( runtimeConfiguration=_Configuration(), tun2socksSettingsCallers=module._Tun2socksSettingsCallers( {} ), ownRuntime=mock.Mock(), rollback=mock.Mock(), ) manager = SimpleNamespace( waitForTUNDeviceBroughtUp=mock.Mock(return_value=True) ) self.assertTrue( ConnectionManager._startApplicationTun2socks( manager, attempt, None, None ) ) tun.stop() attempt.rollback.assert_not_called() self.assertEqual(len(scripts), 1) self.assertIn(shlex.quote(device), scripts[0]) self.assertEqual('via ' in scripts[0], not existing) if not existing: self.assertIn(shlex.quote(interface), scripts[0]) bypass = next( line for line in scripts[0].splitlines() if 'via ' in line ) self.assertEqual( shlex.split(bypass), [ 'ip', 'route', 'add', source, 'via', gateway, 'dev', interface, ], ) if __name__ == '__main__': application() unittest.main()