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

Fix native TUN ownership semantics

Browse Source 
Signed-off-by: Loren Eteval <loren.eteval@proton.me>
This commit is contained in:
Loren Eteval
2026-08-17 18:19:55 +08:00
parent 9bb1ffec11
commit 8510a0cb9b
9 changed files with 478 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Furious/Backends/Hysteria2/Plugin.py
+13 -20
View File
@@ -161,47 +161,40 @@ class Hysteria2KernelFactory(KernelFactory):
kernelTypes = (Hysteria2,)
def prepareTUN(self, config) -> bool:
"""Add Hysteria 2 native TUN mode when enabled and safe to route."""
"""Preserve user TUN or replace it with Furious-managed native TUN."""
if not isHysteria2TUNEnabled():
# Work on ConnectionManager's copy so a stored native TUN block
# cannot run alongside the external tun2socks implementation.
config.pop('tun', None)
return False
# Presence is authoritative even when the block is malformed: the
# core must report that error instead of Furious silently changing
# the connection to application tun2socks.
return hasHysteria2TUNConfig(config)
if PLATFORM == 'Linux' and not SystemRuntime.isAdmin():
# Native Hysteria 2 creates the interface and routing table in its
# own process, so it cannot use ConnectionManager's privileged helper.
# Keep the existing external tun2socks path available instead.
config.pop('tun', None)
logger.warning(
'Hysteria 2 native TUN requires superuser privileges on '
'Linux; falling back to external tun2socks'
raise TUNPreparationError(
'Hysteria 2 native TUN requires Linux superuser privileges'
)
return False
settings = getHysteria2TUNSettings()
serverAddresses = resolveHysteria2ServerAddresses(config)
route = settings.get('route', {})
hasManualExclusions = bool(route.get('ipv4Exclude') or route.get('ipv6Exclude'))
if not serverAddresses and not hasManualExclusions:
config.pop('tun', None)
logger.error(
'Hysteria 2 native TUN disabled for this connection because '
raise TUNPreparationError(
'Hysteria 2 native TUN cannot start because '
'the server address could not be resolved and no manual route '
'exclusion is configured'
)
return False
config['tun'] = buildHysteria2TUNConfig(settings, serverAddresses)
return True
def usesApplicationTun2socks(self, config) -> bool:
"""Use host tun2socks only when the runtime has no native TUN block."""
return not hasHysteria2TUNConfig(config)
def create(self, request: KernelRequest):
"""Create a prepared Hysteria 2 kernel launch."""
if request.log:
Furious/Backends/Hysteria2/TUN.py
+12
View File
@@ -31,6 +31,7 @@ __all__ = [
'DEFAULT_HYSTERIA2_TUN_SETTINGS',
'buildHysteria2TUNConfig',
'getHysteria2TUNSettings',
'hasHysteria2TUNConfig',
'isHysteria2TUNEnabled',
'resolveHysteria2ServerAddresses',
'saveHysteria2TUNSettings',
@@ -72,6 +73,17 @@ def setHysteria2TUNEnabled(enabled: bool):
)
def hasHysteria2TUNConfig(config) -> bool:
"""Return whether a Hysteria 2 document explicitly defines native TUN.
Presence, rather than shape, is intentional. A malformed user-provided
block remains authoritative for a normal connection so Hysteria 2 can
report its validation error instead of Furious silently selecting
application tun2socks.
"""
return 'tun' in config
def _normalizedStringList(value) -> list[str]:
"""Return stripped, non-empty strings from a list-like value."""
if not isinstance(value, (list, tuple)):
Furious/Backends/Xray/Plugin.py
+9 -2
View File
@@ -170,9 +170,12 @@ class XrayKernelFactory(KernelFactory):
return None
def prepareTUN(self, config) -> bool:
"""Add the configured Xray native TUN inbound when enabled."""
"""Preserve user TUN or replace it with Furious-managed native TUN."""
if not isXrayTUNEnabled():
return False
# An explicit user inbound is authoritative when Furious-managed
# native TUN is disabled. Reporting it as handled prevents the
# normal connection path from also starting application tun2socks.
return hasXrayTUNInbound(config)
inbounds = config.get('inbounds')
@@ -191,6 +194,10 @@ class XrayKernelFactory(KernelFactory):
return True
def usesApplicationTun2socks(self, config) -> bool:
"""Use host tun2socks only when the runtime has no native TUN inbound."""
return not hasXrayTUNInbound(config)
def routingOptions(self, config=None):
"""Return built-in and named routing modes supported by Xray."""
options = [
Furious/Plugins/API.py
+13 -1
View File
@@ -51,11 +51,16 @@ __all__ = [
'TrafficCounters',
'TrafficStatsMonitor',
'TrafficStatsProvider',
'TUNPreparationError',
]
PLUGIN_API_VERSION = 3
class TUNPreparationError(RuntimeError):
"""Report that requested native TUN cannot be prepared safely."""
class CapabilityKind(str, Enum):
"""Identify independently discoverable plugin extension points."""
@@ -429,7 +434,14 @@ class KernelFactory(PluginCapability):
return None
def prepareTUN(self, config) -> bool:
"""Prepare native TUN and return whether the backend handles it."""
"""Prepare normal-connection TUN and report native TUN ownership.
Implementations must preserve an explicit user native-TUN definition
when host-managed native TUN is disabled. Proxy-only operations strip
native TUN explicitly in their own preparation method instead. Raise
``TUNPreparationError`` when requested managed TUN cannot be prepared
safely; the host must not silently choose another TUN implementation.
"""
return False
def usesApplicationTun2socks(self, config) -> bool:
Furious/Plugins/Registry.py
+4
View File
@@ -904,6 +904,10 @@ class PluginRegistry:
raise TypeError('kernel TUN preparation result must be a boolean')
return handled
except TUNPreparationError:
# A managed native-TUN request must fail the connection rather than
# silently changing the user's selected networking implementation.
raise
except Exception as ex:
# Any non-exit exceptions
Furious/Plugins/__init__.py
+2
View File
@@ -47,6 +47,7 @@ from .API import (
TrafficCounters,
TrafficStatsMonitor,
TrafficStatsProvider,
TUNPreparationError,
)
from .Profile import (
blankConfiguration,
@@ -95,6 +96,7 @@ __all__ = [
'TrafficCounters',
'TrafficStatsMonitor',
'TrafficStatsProvider',
'TUNPreparationError',
'blankConfiguration',
'blankProfile',
'configurationFromAny',
Furious/Service/ConnectionManager.py
+9 -1
View File
@@ -157,7 +157,15 @@ class ConnectionManager(Mixins.CleanupOnExit):
if tunModeRequested:
registry = getPluginRegistry()
pluginTUN = registry.prepareTUN(configcopy)
try:
pluginTUN = registry.prepareTUN(configcopy)
except TUNPreparationError as ex:
self._lastStartError = str(ex)
return abortStart(
f'native TUN preparation failed: {self._lastStartError}'
)
if not pluginTUN:
applicationTun2socks = registry.usesApplicationTun2socks(configcopy)
tests/README.md
+2 -1
View File
@@ -16,6 +16,7 @@ system proxy, TUN, routing, update network clients, or real proxy cores.
| SOCKS/SIP002 Shadowsocks codecs and generated round trips | `test_socks_uri.py`, `test_shadowsocks_uri.py` |
| Subscription-group reconciliation and ownership isolation | `test_subscription_sync.py` |
| External process launch, output, shutdown, threads, TUN metadata | `test_external_core.py` |
| Xray/Hysteria2 native-TUN ownership and proxy-only stripping | `test_native_tun_semantics.py` |
| Rolling metrics, stable buckets, lazy rendering, and hover | `test_metrics_behavior.py` |
| Settings sandbox and navigation overlay behavior | `test_isolation_and_navigation.py` |
| Editor mappings, lazy log rendering, routing/message-box/connection UI | `test_ui_behavior.py` |
@@ -61,7 +62,7 @@ environment, so activate the project's virtual environment first when needed.
python -m unittest discover -s tests -v
# Fast logic, persistence, plugin, controller, codec, and UI behavior
python -m unittest tests.test_models_and_services tests.test_plugin_architecture tests.test_controllers tests.test_subscription_sync tests.test_socks_uri tests.test_shadowsocks_uri tests.test_metrics_behavior tests.test_isolation_and_navigation tests.test_ui_behavior -v
python -m unittest tests.test_models_and_services tests.test_plugin_architecture tests.test_controllers tests.test_subscription_sync tests.test_socks_uri tests.test_shadowsocks_uri tests.test_native_tun_semantics tests.test_metrics_behavior tests.test_isolation_and_navigation tests.test_ui_behavior -v
# Direct Qt/process integration and destruction/lifetime checks
python -m unittest tests.test_external_core tests.test_qt_lifetime -v
tests/test_native_tun_semantics.py
+414
View File
@@ -0,0 +1,414 @@
# Copyright (C) 2024–present Loren Eteval & contributors <loren.eteval@proton.me>
#
# 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 <https://www.gnu.org/licenses/>.
"""Protect native-TUN ownership for normal and proxy-only operations."""
from __future__ import annotations
from unittest import mock
import copy
import unittest
from Furious.Backends.Configuration import ConfigHysteria2, ConfigXray
from Furious.Backends.Hysteria2 import Plugin as Hysteria2PluginModule
from Furious.Backends.Hysteria2.Plugin import Hysteria2KernelFactory
from Furious.Backends.Xray import Plugin as XrayPluginModule
from Furious.Backends.Xray.Plugin import XrayKernelFactory
from Furious.Plugins.API import TUNPreparationError
from Furious.Plugins.Registry import PluginRegistry
from Furious.Service.ConnectionManager import ConnectionManager
class _RecordingConnectionManager(ConnectionManager):
"""Capture the runtime document without starting a core process."""
def _startKernel(self, config, *_args, **_kwargs):
"""Record the prepared runtime configuration as a successful launch."""
self.runtimeConfiguration = config
return None, True
class XrayNativeTUNTest(unittest.TestCase):
"""Verify Xray runtime copies preserve or replace native TUN intentionally."""
customTUN = {
'tag': 'user-tun',
'protocol': 'tun',
'settings': {'gateway': ['192.0.2.1/24']},
}
generatedTUN = {
'tag': 'tun',
'protocol': 'tun',
'settings': {'gateway': ['10.0.0.1/16']},
}
httpInbound = {'tag': 'http', 'protocol': 'http'}
def setUp(self):
"""Create one stateless Xray kernel capability."""
self.factory = XrayKernelFactory()
def configuration(self, *, nativeTUN: bool) -> ConfigXray:
"""Return one minimal Xray document with optional custom native TUN."""
inbounds = [copy.deepcopy(self.httpInbound)]
if nativeTUN:
inbounds.append(copy.deepcopy(self.customTUN))
return ConfigXray({'inbounds': inbounds, 'outbounds': []})
def prepare(self, original, enabled):
"""Prepare an independent runtime copy under one toggle state."""
runtime = original.deepcopy()
with (
mock.patch.object(
XrayPluginModule,
'isXrayTUNEnabled',
return_value=enabled,
),
mock.patch.object(
XrayPluginModule,
'buildXrayTUNInbound',
return_value=copy.deepcopy(self.generatedTUN),
),
):
handled = self.factory.prepareTUN(runtime)
return runtime, handled
def testExistingTUNIsReplacedWhenManagedTUNIsEnabled(self):
"""Replace every runtime TUN inbound without changing persisted JSON."""
original = self.configuration(nativeTUN=True)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, True)
self.assertTrue(handled)
self.assertEqual(original, snapshot)
self.assertEqual(runtime['inbounds'], [self.httpInbound, self.generatedTUN])
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testExistingTUNIsPreservedWhenManagedTUNIsDisabled(self):
"""Treat a user TUN inbound as authoritative for normal connection."""
original = self.configuration(nativeTUN=True)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, False)
self.assertTrue(handled)
self.assertEqual(runtime, original)
self.assertEqual(original, snapshot)
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testGeneratedTUNIsInjectedWhenEnabledAndMissing(self):
"""Inject Furious-managed Xray TUN into only the runtime copy."""
original = self.configuration(nativeTUN=False)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, True)
self.assertTrue(handled)
self.assertEqual(original, snapshot)
self.assertEqual(runtime['inbounds'], [self.httpInbound, self.generatedTUN])
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testMissingTUNLeavesApplicationFallbackAvailableWhenDisabled(self):
"""Report no native owner so global TUN may use application tun2socks."""
original = self.configuration(nativeTUN=False)
runtime, handled = self.prepare(original, False)
self.assertFalse(handled)
self.assertEqual(runtime, original)
self.assertTrue(self.factory.usesApplicationTun2socks(runtime))
def testDownloadTestExplicitlyStripsNativeTUN(self):
"""Keep normal custom TUN while deriving a TUN-free speed-test copy."""
original = self.configuration(nativeTUN=True)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, False)
speedTest = self.factory.prepareDownloadTest(original, 18080)
self.assertTrue(handled)
self.assertEqual(runtime, original)
self.assertEqual(original, snapshot)
self.assertFalse(
any(inbound.get('protocol') == 'tun' for inbound in speedTest['inbounds'])
)
def testConnectionManagerDoesNotPairCustomTUNWithTun2socks(self):
"""Stop host tun2socks selection once Xray reports custom native TUN."""
original = self.configuration(nativeTUN=True)
registry = mock.Mock()
registry.prepareTUN.side_effect = self.factory.prepareTUN
registry.usesApplicationTun2socks.side_effect = (
self.factory.usesApplicationTun2socks
)
manager = _RecordingConnectionManager()
with (
mock.patch.object(
XrayPluginModule,
'isXrayTUNEnabled',
return_value=False,
),
mock.patch(
'Furious.Service.ConnectionManager.SystemRuntime.isTUNMode',
return_value=True,
),
mock.patch(
'Furious.Service.ConnectionManager.getPluginRegistry',
return_value=registry,
),
mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks,
):
self.assertTrue(manager.start(original, 'Global'))
registry.usesApplicationTun2socks.assert_not_called()
tun2socks.assert_not_called()
self.assertEqual(manager.runtimeConfiguration, original)
manager.cleanup()
def testManagedTUNPreparationFailureDoesNotFallBackToTun2socks(self):
"""Fail an unavailable managed TUN instead of changing networking mode."""
original = self.configuration(nativeTUN=True)
registry = mock.Mock()
registry.prepareTUN.side_effect = TUNPreparationError('managed TUN failed')
manager = _RecordingConnectionManager()
with (
mock.patch(
'Furious.Service.ConnectionManager.SystemRuntime.isTUNMode',
return_value=True,
),
mock.patch(
'Furious.Service.ConnectionManager.getPluginRegistry',
return_value=registry,
),
mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks,
):
self.assertFalse(manager.start(original, 'Global'))
self.assertEqual(manager.lastStartError, 'managed TUN failed')
registry.usesApplicationTun2socks.assert_not_called()
tun2socks.assert_not_called()
self.assertFalse(hasattr(manager, 'runtimeConfiguration'))
def testRegistryPropagatesIntentionalTUNPreparationFailure(self):
"""Keep the explicit failure distinct from an unsupported capability."""
registry = PluginRegistry()
factory = mock.Mock(factoryId='managed-tun-factory')
factory.prepareTUN.side_effect = TUNPreparationError('managed TUN failed')
registry.factoryForConfig = mock.Mock(return_value=factory)
with self.assertRaisesRegex(TUNPreparationError, 'managed TUN failed'):
registry.prepareTUN(self.configuration(nativeTUN=False))
class Hysteria2NativeTUNTest(unittest.TestCase):
"""Verify Hysteria 2 runtime copies preserve or replace native TUN."""
customTUN = {
'name': 'user-tun',
'address': {'ipv4': '192.0.2.1/30'},
}
generatedTUN = {
'name': 'hytun',
'address': {'ipv4': '100.100.100.101/30'},
}
def setUp(self):
"""Create one stateless Hysteria 2 kernel capability."""
self.factory = Hysteria2KernelFactory()
def configuration(self, *, nativeTUN: bool) -> ConfigHysteria2:
"""Return one minimal Hysteria 2 document with optional custom TUN."""
config = ConfigHysteria2(
{
'server': '203.0.113.10:443',
'http': {'listen': '127.0.0.1:10809'},
}
)
if nativeTUN:
config['tun'] = copy.deepcopy(self.customTUN)
return config
def prepare(self, original, enabled):
"""Prepare an independent runtime copy under one toggle state."""
runtime = original.deepcopy()
with (
mock.patch.object(
Hysteria2PluginModule,
'isHysteria2TUNEnabled',
return_value=enabled,
),
mock.patch.object(Hysteria2PluginModule, 'PLATFORM', 'Windows'),
mock.patch.object(
Hysteria2PluginModule,
'getHysteria2TUNSettings',
return_value={},
),
mock.patch.object(
Hysteria2PluginModule,
'resolveHysteria2ServerAddresses',
return_value=['203.0.113.10'],
),
mock.patch.object(
Hysteria2PluginModule,
'buildHysteria2TUNConfig',
return_value=copy.deepcopy(self.generatedTUN),
),
):
handled = self.factory.prepareTUN(runtime)
return runtime, handled
def testExistingTUNIsReplacedWhenManagedTUNIsEnabled(self):
"""Replace the runtime block without changing the persisted document."""
original = self.configuration(nativeTUN=True)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, True)
self.assertTrue(handled)
self.assertEqual(runtime['tun'], self.generatedTUN)
self.assertEqual(original, snapshot)
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testExistingTUNIsPreservedWhenManagedTUNIsDisabled(self):
"""Treat a user TUN block as authoritative for normal connection."""
original = self.configuration(nativeTUN=True)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, False)
self.assertTrue(handled)
self.assertEqual(runtime, original)
self.assertEqual(original, snapshot)
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testMalformedExplicitTUNStillPreventsSilentFallback(self):
"""Leave validation of an explicit malformed block to Hysteria 2."""
original = self.configuration(nativeTUN=False)
original['tun'] = 'malformed-user-value'
runtime, handled = self.prepare(original, False)
self.assertTrue(handled)
self.assertEqual(runtime['tun'], 'malformed-user-value')
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testGeneratedTUNIsInjectedWhenEnabledAndMissing(self):
"""Inject Furious-managed Hysteria 2 TUN into only the runtime copy."""
original = self.configuration(nativeTUN=False)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, True)
self.assertTrue(handled)
self.assertEqual(runtime['tun'], self.generatedTUN)
self.assertEqual(original, snapshot)
self.assertFalse(self.factory.usesApplicationTun2socks(runtime))
def testMissingTUNLeavesApplicationFallbackAvailableWhenDisabled(self):
"""Report no native owner so global TUN may use application tun2socks."""
original = self.configuration(nativeTUN=False)
runtime, handled = self.prepare(original, False)
self.assertFalse(handled)
self.assertEqual(runtime, original)
self.assertTrue(self.factory.usesApplicationTun2socks(runtime))
def testUnavailableManagedTUNFailsWithoutRemovingUserTUN(self):
"""Do not replace a requested native TUN with application tun2socks."""
original = self.configuration(nativeTUN=True)
runtime = original.deepcopy()
with (
mock.patch.object(
Hysteria2PluginModule,
'isHysteria2TUNEnabled',
return_value=True,
),
mock.patch.object(Hysteria2PluginModule, 'PLATFORM', 'Linux'),
mock.patch.object(
Hysteria2PluginModule.SystemRuntime,
'isAdmin',
return_value=False,
),
):
with self.assertRaisesRegex(TUNPreparationError, 'superuser privileges'):
self.factory.prepareTUN(runtime)
self.assertEqual(runtime, original)
def testDownloadTestExplicitlyStripsNativeTUN(self):
"""Keep normal custom TUN while deriving a TUN-free speed-test copy."""
original = self.configuration(nativeTUN=True)
snapshot = copy.deepcopy(original)
runtime, handled = self.prepare(original, False)
speedTest = self.factory.prepareDownloadTest(original, 18080)
self.assertTrue(handled)
self.assertEqual(runtime, original)
self.assertEqual(original, snapshot)
self.assertNotIn('tun', speedTest)
def testConnectionManagerDoesNotPairCustomTUNWithTun2socks(self):
"""Stop host tun2socks selection once Hysteria 2 owns native TUN."""
original = self.configuration(nativeTUN=True)
registry = mock.Mock()
registry.prepareTUN.side_effect = self.factory.prepareTUN
registry.usesApplicationTun2socks.side_effect = (
self.factory.usesApplicationTun2socks
)
manager = _RecordingConnectionManager()
with (
mock.patch.object(
Hysteria2PluginModule,
'isHysteria2TUNEnabled',
return_value=False,
),
mock.patch(
'Furious.Service.ConnectionManager.SystemRuntime.isTUNMode',
return_value=True,
),
mock.patch(
'Furious.Service.ConnectionManager.getPluginRegistry',
return_value=registry,
),
mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks,
):
self.assertTrue(manager.start(original, 'Global'))
registry.usesApplicationTun2socks.assert_not_called()
tun2socks.assert_not_called()
self.assertEqual(manager.runtimeConfiguration, original)
manager.cleanup()
if __name__ == '__main__':
unittest.main()