Max/LorenEteval_Furious
1
0
Fork 0
mirror of https://github.com/LorenEteval/Furious.git synced 2026-10-13 01:38:33 +03:00
Code Issues Packages Projects Releases Wiki Activity

Quote Linux tun2socks script arguments

Browse Source 
Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-10-09 20:07:51 +08:00
parent c4e18124e6
commit c46088f922
3 changed files with 142 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Furious/Service/ConnectionManager.py
+23 -13
View File
@@ -69,6 +69,7 @@ import copy
import time
import ipaddress
import weakref
import shlex
__all__ = ['ConnectionManager', 'ConnectionStartOperation', 'ConnectionStartStage']
@@ -1193,27 +1194,31 @@ class ConnectionStartOperation(QtCore.QObject):
self._tun.cleanup = _linuxCleanup
deviceName, interface = (
shlex.quote(APPLICATION_TUN2SOCKS_DEVICE_NAME),
shlex.quote(str(self._interface)),
)
commandBringUpTUN = ''
if not SystemRoutingTable.LinuxFindTUNDevice(APPLICATION_TUN2SOCKS_DEVICE_NAME):
commandBringUpTUN = (
f'ip tuntap add mode tun dev '
f'{APPLICATION_TUN2SOCKS_DEVICE_NAME}\n'
f'{deviceName}\n'
f'ip addr add 10.10.10.10/24 dev '
f'{APPLICATION_TUN2SOCKS_DEVICE_NAME}\n'
f'ip link set dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} up'
f'{deviceName}\n'
f'ip link set dev {deviceName} up'
)
commandAddDefaultRoute = (
f'ip route add default dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} ' 'metric 5'
)
commandAddDefaultRoute = f'ip route add default dev {deviceName} ' 'metric 5'
def route(source, destination):
return f'{source} via {destination} dev {self._interface}'
iproute = SystemRoutingTable.LinuxGetIpRoute()
commandBypass = '\n'.join(
f'ip route add {route(sourceIP, destinationIP)}'
f'ip route add {shlex.quote(str(sourceIP))} '
f'via {shlex.quote(str(destinationIP))} dev {interface}'
for sourceIP, destinationIP in SystemRoutingTable.managedRoutes
if iproute.find(route(sourceIP, destinationIP)) == -1
)
@@ -2048,6 +2053,11 @@ class ConnectionManager(Mixins.CleanupOnExit):
tun.cleanup = functools.partial(_linuxCleanup)
deviceName, quotedInterface = (
shlex.quote(APPLICATION_TUN2SOCKS_DEVICE_NAME),
shlex.quote(str(interface)),
)
if SystemRoutingTable.LinuxFindTUNDevice(APPLICATION_TUN2SOCKS_DEVICE_NAME):
logger.info(
f'find TUN device {APPLICATION_TUN2SOCKS_DEVICE_NAME} success. '
@@ -2062,14 +2072,13 @@ class ConnectionManager(Mixins.CleanupOnExit):
)
commandBringUpTUN = (
f'ip tuntap add mode tun dev {APPLICATION_TUN2SOCKS_DEVICE_NAME}\n'
f'ip addr add 10.10.10.10/24 dev {APPLICATION_TUN2SOCKS_DEVICE_NAME}\n'
f'ip link set dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} up'
f'ip tuntap add mode tun dev {deviceName}\n'
f'ip addr add 10.10.10.10/24 dev {deviceName}\n'
f'ip link set dev {deviceName} up'
)
commandAddDefaultRoute = (
f'ip route add default dev {APPLICATION_TUN2SOCKS_DEVICE_NAME} '
'metric 5'
f'ip route add default dev {deviceName} ' 'metric 5'
)
def route(source, destination) -> str:
@@ -2080,7 +2089,8 @@ class ConnectionManager(Mixins.CleanupOnExit):
commandBypass = '\n'.join(
list(
f'ip route add {route(sourceIP, destinationIP)}'
f'ip route add {shlex.quote(str(sourceIP))} '
f'via {shlex.quote(str(destinationIP))} dev {quotedInterface}'
for sourceIP, destinationIP in SystemRoutingTable.managedRoutes
if iproute.find(route(sourceIP, destinationIP)) == -1
)
tests/README.md
+1 -1
View File
@@ -102,7 +102,7 @@ worker. Choose tests by the changed contract rather than by filename alone.
| --- | --- |
| [test_application_process.py](test_application_process.py) | Exact application-child ownership, spawned factory/runner failure classification and crash reporting, captured and asserted expected child diagnostics, partial Qt startup, crash-write failure, intentional exits, exception/signal handling and command-line dispatch. |
| [test_architecture_refactors.py](test_architecture_refactors.py) | Startup acquisition/rollback including partial controller construction and cleanup retry, singleton election and real isolated IPC race, tray/exit policy, host integration ownership, bounded core-log transport, connection transactions, stylesheet composition. |
| [test_connection_startup_async.py](test_connection_startup_async.py) | Real local-listener readiness, timeout/cancel/replacement, semantic exits, DNS reply lifetime, mocked platform-specific TUN sequencing. |
| [test_connection_startup_async.py](test_connection_startup_async.py) | Real local-listener readiness, timeout/cancel/replacement, semantic exits, DNS reply lifetime, mocked platform-specific TUN sequencing, quoted Linux tun2socks scripts and unchanged route-existence detection in both startup paths. |
| [test_controllers.py](test_controllers.py) | Connection state/error/reconnect transitions, reentrant cancellation/replacement before launch and during completion, startup restoration, shared settings, routing fallback persistence and tray/selector agreement after custom-routing disable/re-enable. |
| [test_runtime_lifecycle.py](test_runtime_lifecycle.py) | Qt-thread exit dispatch, commit/exit races, duplicate and late exits, idempotent release and retained failure/retry ownership, failed reap/handle close, spawn failure, queue/timer disposal on preparation failure. |
| [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, failed reap/thread-join retry, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. |
tests/test_connection_startup_async.py
+118
View File
@@ -27,6 +27,7 @@ from Furious.Service.ConnectionManager import (
ConnectionManager,
ConnectionStartStage,
_ConditionProbe,
ConnectionStartOperation,
)
from Furious.Service.DnsResolver import DnsResolutionOperation, DnsResolver
from Furious.Frozenlib import AppSettings
@@ -38,8 +39,11 @@ from shiboken6 import isValid, delete as deleteQObject
from tests.support import application, processQtEvents, waitFor
from unittest import TestCase, mock
from types import SimpleNamespace
import importlib
import io
import shlex
import unittest
@@ -1185,6 +1189,120 @@ class ConnectionStartupAsyncTest(TestCase):
self.assertIsNone(manager._activeStartOperation)
class LinuxTun2socksScriptTest(TestCase):
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):
file = io.StringIO()
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(),
_fail=mock.Mock(),
)
ConnectionStartOperation._prepareLinuxTun(operation)
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()