Integrate sing-tun application TUN backend

Signed-off-by: Loren Eteval <loren.eteval@proton.me>
2026-10-12 17:28:35 +03:00 · 2026-10-06 00:37:17 +08:00
parent 714fad2d4b
commit 0dc11b6755
26 changed files with 2342 additions and 101 deletions
@@ -442,6 +442,7 @@ class DesktopApplication(ApplicationRunner, SingletonApplication):
        self._userServers = None
        self._userSubs = None
        self._userTUNSettings = None
+        self._userSingTUNSettings = None

        # ThreadPool
        self.threadPool = QtCore.QThreadPool(self)
@@ -624,6 +625,7 @@ class DesktopApplication(ApplicationRunner, SingletonApplication):
        self._userServers = Storage.UserServers()
        self._userSubs = Storage.UserSubs()
        self._userTUNSettings = Storage.UserTUNSettings()
+        self._userSingTUNSettings = Storage.UserSingTUNSettings()

    def _initializeControllers(self):
        """Create the process-lifetime application state authorities."""
@@ -34,6 +34,7 @@ from Furious.Backends.SocksURI import (
    parseSocksURI,
    serializeSocksURI,
 )
+from Furious.Frozenlib.SocksProxy import socksURL

 from typing import Union, Tuple

@@ -1575,6 +1576,49 @@ class ConfigXray(CoreConfiguration):

            return ''

+    def applicationTUNProxy(self) -> str:
+        """Validate Xray's selected listener and carry its optional SOCKS auth."""
+        endpoint = socksURL(self.socksProxy())
+
+        for inbound in self.get('inbounds', []):
+            if inbound.get('protocol') != 'socks':
+                continue
+
+            settings = inbound.get('settings', {})
+
+            if settings.get('udp', False) is not True:
+                raise ValueError(
+                    'Application TUN requires SOCKS UDP on the selected Xray listener'
+                )
+
+            auth = settings.get('auth', 'noauth')
+
+            if auth == 'noauth':
+                return endpoint
+            if auth != 'password':
+                raise ValueError('Unsupported Xray SOCKS authentication')
+
+            accounts = settings.get('accounts', [])
+
+            if not accounts or not isinstance(accounts[0], dict):
+                raise ValueError('Xray SOCKS authentication requires an account')
+
+            user, password = accounts[0].get('user', ''), accounts[0].get('pass', '')
+
+            if not isinstance(user, str) or not isinstance(password, str):
+                raise ValueError('Invalid Xray SOCKS authentication')
+
+            return socksURL(
+                'socks5://'
+                + quote(user, safe='')
+                + ':'
+                + quote(password, safe='')
+                + '@'
+                + endpoint.removeprefix('socks5://')
+            )
+
+        raise ValueError('A SOCKS listener is required for application TUN')
+
    def setHttpProxy(self, endpoint: str) -> bool:
        """Set the local HTTP proxy endpoint."""
        try:
@@ -1925,6 +1969,32 @@ class ConfigHysteria1(CoreConfiguration):

            return ''

+    def applicationTUNProxy(self) -> str:
+        """Validate the selected Hysteria SOCKS transit listener."""
+        listener = self.get('socks5', {})
+
+        if listener.get('disable_udp', False) is not False:
+            raise ValueError('Application TUN requires SOCKS UDP forwarding')
+
+        endpoint = socksURL(self.socksProxy())
+
+        user, password = listener.get('user', ''), listener.get('password', '')
+
+        if not user and not password:
+            return endpoint
+
+        if not isinstance(user, str) or not isinstance(password, str):
+            raise ValueError('Invalid SOCKS authentication')
+
+        return socksURL(
+            'socks5://'
+            + quote(user, safe='')
+            + ':'
+            + quote(password, safe='')
+            + '@'
+            + endpoint.removeprefix('socks5://')
+        )
+
    def setHttpProxy(self, endpoint: str) -> bool:
        """Set the local HTTP proxy endpoint."""
        try:
@@ -2285,6 +2355,32 @@ class ConfigHysteria2(CoreConfiguration):

            return ''

+    def applicationTUNProxy(self) -> str:
+        """Validate the selected Hysteria SOCKS transit listener."""
+        listener = self.get('socks5', {})
+
+        if listener.get('disableUDP', False) is not False:
+            raise ValueError('Application TUN requires SOCKS UDP forwarding')
+
+        endpoint = socksURL(self.socksProxy())
+
+        user, password = listener.get('username', ''), listener.get('password', '')
+
+        if not user and not password:
+            return endpoint
+
+        if not isinstance(user, str) or not isinstance(password, str):
+            raise ValueError('Invalid SOCKS authentication')
+
+        return socksURL(
+            'socks5://'
+            + quote(user, safe='')
+            + ':'
+            + quote(password, safe='')
+            + '@'
+            + endpoint.removeprefix('socks5://')
+        )
+
    def setHttpProxy(self, endpoint: str) -> bool:
        """Set the local HTTP proxy endpoint."""
        try:
@@ -50,6 +50,12 @@ SYSTEM_PROXY_MODE_OPTIONS = (
 )

 registerAppSettings('VPNMode', isBinary=True)
+# A missing preference, including upgrades from 0.8.2, deliberately selects
+# sing-tun. VPNMode remains independent: choosing an engine does not enable TUN.
+# 0.8.2 ignores this key and continues using application-managed tun2socks.
+registerAppSettings(
+    APPLICATION_TUN_BACKEND, validRange=list(TUN_BACKENDS), default='sing-tun'
+)
 registerAppSettings(
    APPLICATION_THEME_SETTING,
    validRange=[theme.value for theme in ApplicationTheme],
@@ -145,6 +151,7 @@ class SettingsController(QtCore.QObject):
    """Apply application settings independently from their presentation."""

    tunModeChanged = QtCore.Signal(bool)
+    tunBackendChanged = QtCore.Signal(str)
    systemProxyModeChanged = QtCore.Signal(str)

    def __init__(self, parent=None):
@@ -184,6 +191,20 @@ class SettingsController(QtCore.QObject):

        showMBoxNewChangesNextTime()

+    def setTUNBackend(self, backend: str):
+        """Select the application TUN implementation for the next connection."""
+        if backend not in TUN_BACKENDS:
+            raise ValueError('Unknown application TUN backend')
+
+        if AppSettings.get(APPLICATION_TUN_BACKEND) == backend:
+            return
+
+        AppSettings.set(APPLICATION_TUN_BACKEND, backend)
+
+        self.tunBackendChanged.emit(backend)
+
+        showMBoxNewChangesNextTime()
+
    @staticmethod
    def setApplicationTheme(theme: ApplicationTheme | str):
        """Persist and immediately apply one application theme preference."""
@@ -1,10 +1,10 @@
 # Embedded runtime guidance

 Inherit `Furious/AGENTS.md` and its root ancestor. Consult Interface for runtime contracts and Service for
-connection ownership. This scope owns reusable embedded execution machinery and application tun2socks;
+connection ownership. This scope owns reusable embedded execution machinery and application TUN engines;
 connection policy remains outside it.

- `Core` supplies shared multiprocessing runtime machinery, bounded output transport, and application tun2socks. External
+- `Core` supplies shared multiprocessing runtime machinery, bounded output transport, and application tun2socks/sing-tun. External
  Core owns its separate direct `subprocess.Popen`; neither layer owns controller, repository, UI, or protocol policy.
 - A launch spec describes prepared child construction, never semantic connection readiness. Serialization and launch
  arguments are prepared before execution starts; constructors may create owned timers/queues that still need
@@ -42,3 +42,10 @@ connection policy remains outside it.
  queues, or callbacks. Start with `tests/test_runtime_lifecycle.py` and `tests/test_connection_startup_async.py`;
  output/process stress lives in the tiers documented by `tests/README.md`. Review output admission and draining
  together when changing backpressure.
+- `SingTUN` imports the Go binding only in a spawned child. Its bounded status pipe establishes native readiness
+  and the actual device name independently of diagnostic output and process liveness. Cooperative stop is followed
+  by exact-child reap and attempt-local host recovery; retain the lease when either fails. Never infer Go cleanup
+  from forced termination. Its `SingTUNHostPlan` remains attached through host-worker drain and restoration; tests
+  in `tests/test_sing_tun.py` cover startup cancellation, validated status with binding-provided failure reasons,
+  and cleanup refusal/retry. Privileged
+  platform smoke tests remain separate from harmless source/compiled lifetime checks.
@@ -76,7 +76,9 @@ class MultiprocessingRuntime(CoreRuntime):

    StopJoinTimeout = 3

-    def __init__(self, launchFactory, *, exitCallback=None, msgCallback=None):
+    def __init__(
+        self, launchFactory, *, exitCallback=None, msgCallback=None, processContext=None
+    ):
        """Prepare an immutable launch specification around an owned queue."""
        super().__init__(exitCallback)

@@ -84,7 +86,8 @@ class MultiprocessingRuntime(CoreRuntime):
        self._lastExit = None
        self._stopRequested = False
        self._exitPublished = False
-        self._output = MsgQueue(msgCallback=msgCallback)
+        self._processContext = processContext
+        self._output = MsgQueue(msgCallback=msgCallback, context=processContext)

        try:
            launch = launchFactory(self._output)
@@ -144,7 +147,12 @@ class MultiprocessingRuntime(CoreRuntime):
        self.setState(RuntimeState.Starting)

        try:
-            process = multiprocessing.Process(**self._launch.processKeywordArguments())
+            if self._processContext is not None:
+                factory = self._processContext.Process
+            else:
+                factory = multiprocessing.Process
+
+            process = factory(**self._launch.processKeywordArguments())

            self._process = process

@@ -52,15 +52,14 @@ class MsgQueue(multiprocessing.queues.Queue):

    def __init__(self, **kwargs):
        """Initialize the bounded queue and its GUI-thread drain timer."""
-        callback = kwargs.pop('msgCallback', None)
-        maximum = kwargs.pop('maximumPendingMessages', self.MAXIMUM_PENDING_MESSAGES)
-
-        super().__init__(
-            maxsize=maximum,
-            **kwargs,
-            ctx=multiprocessing.get_context(),
+        callback, maximum, context = (
+            kwargs.pop('msgCallback', None),
+            kwargs.pop('maximumPendingMessages', self.MAXIMUM_PENDING_MESSAGES),
+            kwargs.pop('context', None) or multiprocessing.get_context(),
        )

+        super().__init__(maxsize=maximum, **kwargs, ctx=context)
+
        self.timer = QtCore.QTimer()
        self._timerConnection = connectWeakly(
            self.timer.timeout,
@@ -0,0 +1,373 @@
+# 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/>.
+
+"""Spawn-only sing-tun bridge with authoritative readiness and cooperative stop."""
+
+from __future__ import annotations
+
+from Furious.Core.MultiprocessingRuntime import (
+    MultiprocessingRuntime,
+    ProcessLaunchSpec,
+)
+from Furious.Core.ProcessOutput import ProcessOutputRedirector
+from Furious.Interface import RuntimeExit, RuntimeExitReason, RuntimeStartError
+
+import copy
+import importlib.metadata
+import json
+import multiprocessing
+import threading
+
+
+def _nativeError(engine):
+    try:
+        error = engine.status.get('error', '')
+    except (AttributeError, RuntimeError):
+        return ''
+
+    return error if isinstance(error, str) else ''
+
+
+def _runEngine(config, control, engineFactory=None):
+    """Child control remains responsive while native startup runs on its own thread."""
+    engine = None
+    worker = None
+    complete = threading.Event()
+    errors = []
+    requested = False
+
+    try:
+        if engineFactory is None:
+            binding = importlib.import_module('sing_tun')
+
+            engine = binding.Engine(binding.Config(**config))
+        else:
+            engine = engineFactory(config)
+
+        def initialize():
+            try:
+                engine.start()
+            except BaseException as ex:
+                errors.append(ex)
+            finally:
+                complete.set()
+
+        worker = threading.Thread(target=initialize, name='sing-tun-initialize')
+        worker.start()
+
+        announced = False
+
+        while True:
+            if control.poll(0.05):
+                try:
+                    command = control.recv_bytes(64)
+                except (EOFError, OSError):
+                    command = b'stop'
+
+                if command == b'stop':
+                    requested = True
+
+                    engine.stop()
+
+            if not complete.is_set():
+                continue
+
+            if errors and not requested:
+                raise errors[0]
+
+            if not announced and not requested:
+                if not engine.ready:
+                    raise RuntimeError('sing-tun did not become ready')
+
+                control.send_bytes(
+                    json.dumps(
+                        {'state': 'ready', 'device_name': engine.device_name}
+                    ).encode()
+                )
+
+                announced = True
+
+            if engine.wait(0):
+                break
+
+        worker.join()
+
+        engine.close(5)
+
+        try:
+            control.send_bytes(b'{"state":"stopped"}')
+        except (EOFError, OSError):
+            pass
+    except BaseException as ex:
+        if isinstance(ex, ImportError):
+            message = 'sing-tun package is unavailable or incompatible'
+        else:
+            message = str(ex) or 'sing-tun native startup or cleanup failed'
+
+        nativeMessage = _nativeError(engine)
+
+        if nativeMessage:
+            message = 'sing-tun: ' + nativeMessage
+
+        try:
+            control.send_bytes(
+                json.dumps({'state': 'failed', 'error': message}).encode()
+            )
+        except (OSError, EOFError):
+            pass
+
+        raise RuntimeError(message) from None
+    finally:
+        try:
+            if engine is not None:
+                try:
+                    engine.stop()
+                except BaseException:
+                    pass
+
+                if worker is not None:
+                    worker.join(5)
+
+                try:
+                    engine.close(5)
+                except BaseException:
+                    # The authoritative failure was already published above.
+                    # Preserve that failure if cleanup also fails; the parent
+                    # retains recovery for incomplete cleanup.
+                    pass
+        finally:
+            control.close()
+
+
+def startSingTUN(output, config, control):
+    """Import the Go binding only inside the spawned child."""
+    ProcessOutputRedirector.launch(output, lambda: _runEngine(config, control), True)
+
+
+class SingTUN(MultiprocessingRuntime):
+    """Own IPC, child execution, and registered attempt-local host restoration."""
+
+    CooperativeStopTimeout = 6
+
+    def __init__(self, config, *, hostPlan=None, **kwargs):
+        context = multiprocessing.get_context('spawn')
+
+        self._control, self._childControl = context.Pipe()
+        self._status = {'state': 'created'}
+        self._hostPlan = hostPlan
+        self._started = False
+        self._cleanExit = False
+        self._nativeWasReady = False
+        self.cleanup = None
+        self._configuration = copy.deepcopy(config)
+
+        try:
+            super().__init__(
+                lambda output: ProcessLaunchSpec(
+                    target=startSingTUN,
+                    args=(output, copy.deepcopy(config), self._childControl),
+                ),
+                processContext=context,
+                **kwargs,
+            )
+        except BaseException:
+            self._control.close()
+            self._childControl.close()
+
+            raise
+
+    @staticmethod
+    def name():
+        return 'sing-tun'
+
+    @staticmethod
+    def version():
+        try:
+            return importlib.metadata.version('sing-tun')
+        except importlib.metadata.PackageNotFoundError:
+            return 'unavailable'
+
+    def start(self):
+        if self._started:
+            raise RuntimeStartError('sing-tun runtimes are one-shot')
+
+        self._started = True
+
+        try:
+            if self._hostPlan is not None:
+                configuration = self._hostPlan.configuration
+            else:
+                configuration = self._configuration
+
+            self._launch = ProcessLaunchSpec(
+                target=startSingTUN,
+                args=(self._output, copy.deepcopy(configuration), self._childControl),
+            )
+
+            if self._hostPlan is not None:
+                self._hostPlan.markActivated()
+
+            super().start()
+
+            self._monitor.setInterval(50)
+        finally:
+            self._childControl.close()
+
+    def _readStatus(self):
+        try:
+            for _ in range(8):
+                if self._control.closed or not self._control.poll():
+                    break
+
+                status = json.loads(self._control.recv_bytes(4096))
+
+                if not isinstance(status, dict) or status.get('state') not in (
+                    'ready',
+                    'failed',
+                    'stopped',
+                ):
+                    raise ValueError()
+
+                if status['state'] == 'ready':
+                    name = status.get('device_name')
+
+                    if (
+                        not isinstance(name, str)
+                        or not 1 <= len(name) <= 32
+                        or not all(
+                            character.isascii()
+                            and (character.isalnum() or character in '_.-')
+                            for character in name
+                        )
+                    ):
+                        raise ValueError()
+
+                    self._nativeWasReady = True
+
+                elif status['state'] == 'failed':
+                    error = status.get('error')
+
+                    if not isinstance(error, str) or not error:
+                        raise ValueError()
+
+                if self._status.get('state') == 'failed':
+                    continue
+
+                self._status = {**self._status, **status}
+                self._cleanExit = status['state'] == 'stopped'
+        except EOFError:
+            pass
+        except (OSError, ValueError, TypeError) as ex:
+            if isinstance(ex, OSError) and str(ex) != 'bad message length':
+                # Windows can report a normal peer close as a broken pipe.
+                return
+
+            self._status = {
+                'state': 'failed',
+                'error': 'Invalid sing-tun status message',
+            }
+
+    @property
+    def ready(self):
+        self._readStatus()
+
+        return self._status.get('state') == 'ready' and self.isRunning()
+
+    @property
+    def deviceName(self):
+        self._readStatus()
+
+        return self._status.get('device_name', '')
+
+    @property
+    def startupError(self):
+        self._readStatus()
+
+        return self._status.get('error', '')
+
+    def _pollProcess(self):
+        self._readStatus()
+
+        super()._pollProcess()
+
+    def interpretExit(self, exitcode, *, requested=False):
+        self._readStatus()
+
+        if self.startupError:
+            if self._nativeWasReady:
+                reason = RuntimeExitReason.Unexpected
+            else:
+                reason = RuntimeExitReason.StartFailure
+
+            return RuntimeExit(exitcode, reason, message=self.startupError)
+
+        return super().interpretExit(exitcode, requested=requested)
+
+    def stop(self):
+        if self._hostPlan is not None:
+            self._hostPlan.drain()
+
+        process = self._process
+
+        if process is not None and process.is_alive():
+            self._stopRequested = True
+
+            try:
+                self._control.send_bytes(b'stop')
+            except (OSError, EOFError):
+                pass
+
+            # Native close has a five-second deadline; allow it to complete
+            # before falling back to the shared exact-child escalation path.
+            process.join(self.CooperativeStopTimeout)
+
+        self._readStatus()
+
+        errors = []
+
+        try:
+            super().stop()
+        except Exception as ex:
+            errors.append(str(ex))
+
+        # A stopped process is not proof of host restoration. Failure keeps the
+        # lease retryable, even after its process handle has been released.
+        if self._hostPlan is not None:
+            try:
+                if self._process is None or not self._process.is_alive():
+                    self._hostPlan.finishCleanup(self._cleanExit or not self._started)
+                else:
+                    # A retained live child prevents native recovery, but host
+                    # DNS snapshots can be restored independently.
+                    self._hostPlan.finishDNSCleanup()
+            except Exception as ex:
+                errors.append(str(ex))
+
+        if callable(self.cleanup):
+            try:
+                self.cleanup()
+            except Exception as ex:
+                errors.append(str(ex))
+
+        if errors:
+            raise RuntimeError('; '.join(errors))
+
+    def dispose(self):
+        super().dispose()
+
+        self._control.close()
+        self._childControl.close()
@@ -22,6 +22,7 @@ from __future__ import annotations
 from .MultiprocessingRuntime import MultiprocessingRuntime, ProcessLaunchSpec
 from .ProcessOutput import MsgQueue, ProcessOutputRedirector
 from .Tun2socks import Tun2socks
+from .SingTUN import SingTUN

 __all__ = [
    'MultiprocessingRuntime',
@@ -29,4 +30,5 @@ __all__ = [
    'ProcessLaunchSpec',
    'ProcessOutputRedirector',
    'Tun2socks',
+    'SingTUN',
 ]
@@ -44,6 +44,7 @@ __all__ = [
    'APPLICATION_TUN2SOCKS_IP_ADDRESS',
    'APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS',
    'APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME',
+    'APPLICATION_TUN_BACKEND',
    'APPLICATION_VERSION',
    'CORE_CHECK_ALIVE_INTERVAL',
    'CRASH_LOG_DIR',
@@ -65,6 +66,7 @@ __all__ = [
    'PYSIDE6_VERSION',
    'ROOT_DIR',
    'SYSTEM_LANGUAGE',
+    'TUN_BACKENDS',
    'URL_GEOIP',
    'URL_GEOIP_SHA256',
    'URL_GEOSITE',
@@ -138,6 +140,9 @@ if PLATFORM == 'Windows':

 CORE_CHECK_ALIVE_INTERVAL = 2500

+APPLICATION_TUN_BACKEND = 'ApplicationTUNBackend'
+TUN_BACKENDS = ('sing-tun', 'tun2socks')
+
 if PLATFORM == 'Windows':
    APPLICATION_TUN2SOCKS_DEVICE_NAME = APPLICATION_NAME
 elif PLATFORM == 'Darwin':
@@ -0,0 +1,95 @@
+# 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/>.
+
+"""Provide local SOCKS endpoint validation and normalization helpers."""
+
+from __future__ import annotations
+
+import ipaddress
+import urllib.parse
+
+
+def socksURL(endpoint):
+    """Normalize a local listener without permitting direct DNS or logging secrets."""
+    if (
+        not isinstance(endpoint, str)
+        or not endpoint
+        or any(c in endpoint for c in '\x00\r\n')
+    ):
+        raise ValueError('A local SOCKS endpoint is required')
+
+    # Xray's legacy accessor can produce an unbracketed IPv6 host:port.
+    if (
+        '://' not in endpoint
+        and not endpoint.startswith('[')
+        and endpoint.count(':') > 1
+    ):
+        host, port = endpoint.rsplit(':', 1)
+
+        endpoint = '[' + host + ']:' + port
+
+    try:
+        parsed = urllib.parse.urlsplit(
+            endpoint if '://' in endpoint else 'socks5://' + endpoint
+        )
+
+        if parsed.scheme != 'socks5' or parsed.path or parsed.query or parsed.fragment:
+            raise ValueError()
+
+        host = parsed.hostname
+
+        if host == 'localhost':
+            host = '127.0.0.1'
+
+        address = ipaddress.ip_address(host)
+
+        if address.is_unspecified:
+            address = ipaddress.ip_address(
+                '::1' if address.version == 6 else '127.0.0.1'
+            )
+
+        port = parsed.port
+
+        if port is None or not 1 <= port <= 65535:
+            raise ValueError()
+    except (ValueError, TypeError):
+        raise ValueError(
+            'Invalid local SOCKS endpoint (IP address and port required)'
+        ) from None
+
+    host = '[' + str(address) + ']' if address.version == 6 else str(address)
+    credentials = ''
+
+    if parsed.username is not None:
+        username = urllib.parse.unquote(parsed.username)
+        password = urllib.parse.unquote(parsed.password or '')
+
+        if (
+            not username
+            or not password
+            or max(len(username.encode()), len(password.encode())) > 255
+        ):
+            raise ValueError('Invalid SOCKS authentication credentials')
+
+        credentials = (
+            urllib.parse.quote(username, safe='')
+            + ':'
+            + urllib.parse.quote(password, safe='')
+            + '@'
+        )
+
+    return 'socks5://' + credentials + host + ':' + str(port)
@@ -37,6 +37,7 @@ from .Constants import (
    APPLICATION_TUN2SOCKS_IP_ADDRESS,
    APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS,
    APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME,
+    APPLICATION_TUN_BACKEND,
    APPLICATION_VERSION,
    CORE_CHECK_ALIVE_INTERVAL,
    CRASH_LOG_DIR,
@@ -58,6 +59,7 @@ from .Constants import (
    PYSIDE6_VERSION,
    ROOT_DIR,
    SYSTEM_LANGUAGE,
+    TUN_BACKENDS,
    URL_GEOIP,
    URL_GEOIP_SHA256,
    URL_GEOSITE,
@@ -123,6 +125,7 @@ __all__ = [
    'APPLICATION_TUN2SOCKS_IP_ADDRESS',
    'APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS',
    'APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME',
+    'APPLICATION_TUN_BACKEND',
    'APPLICATION_VERSION',
    'ApplicationTheme',
    'AppBinarySettings',
@@ -164,6 +167,7 @@ __all__ = [
    'PySide6Legacy',
    'ROOT_DIR',
    'SYSTEM_LANGUAGE',
+    'TUN_BACKENDS',
    'StartupOnBoot',
    'SystemProxy',
    'SystemRoutingTable',
@@ -219,6 +219,10 @@ class CoreConfiguration(dict):

        return ''

+    def applicationTUNProxy(self) -> str:
+        """Return the SOCKS transit endpoint; backends validate listener policy."""
+        return self.socksProxy()
+
    def remoteAddress(self) -> str:
        """Return the remote host that must remain reachable outside TUN."""
        return str(getattr(self, 'itemAddress', ''))
@@ -277,6 +277,10 @@ class ServerProfile(MutableMapping[str, Any]):
        """Return the connection's SOCKS proxy endpoint."""
        return self.connection.socksProxy()

+    def applicationTUNProxy(self) -> str:
+        """Delegate SOCKS transit policy to the selected backend document."""
+        return self.connection.applicationTUNProxy()
+
    def remoteAddress(self) -> str:
        """Return the connection's operational remote host."""
        return self.connection.remoteAddress()
@@ -0,0 +1,364 @@
+# 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/>.
+
+"""Pure configuration preparation for the application sing-tun bridge."""
+
+from __future__ import annotations
+
+import copy
+import functools
+import ipaddress
+import math
+import re
+
+SING_TUN_HOST_DEFAULTS = {
+    'primaryAdapterInterfaceName': '',
+    'primaryAdapterInterfaceIP': '',
+    'tunAdapterInterfaceDNS': '',
+    'bypassTUNAdapterInterfaceIP': '',
+    'disablePrimaryAdapterInterfaceDNS': True,
+}
+
+SING_TUN_DEFAULTS = {
+    'stack': 'go',
+    'log_level': 'error',
+    'network_interface': '',
+    'max_sessions': 1024,
+    'udp_timeout': 60,
+    'connect_timeout': 10,
+    'tcp_idle_timeout': 300,
+    'tun_options': {
+        'Name': '',
+        'MTU': 1500,
+        'Inet4Address': ['198.18.0.1/15'],
+        'Inet6Address': [],
+        'AutoRoute': True,
+        'EXP_ExternalConfiguration': False,
+        'DNSMode': 'disabled',
+        'EXP_DisableDNSHijack': True,
+    },
+    'stack_options': {'UDPTimeout': '1m'},
+    'host_options': SING_TUN_HOST_DEFAULTS,
+}
+_OWNED = {
+    'AutoRoute': True,
+    'EXP_ExternalConfiguration': False,
+    'DNSMode': 'disabled',
+    'EXP_DisableDNSHijack': True,
+}
+_PREFIXES = {
+    'Inet4Address': 4,
+    'Inet6Address': 6,
+    'Inet4RouteAddress': 4,
+    'Inet6RouteAddress': 6,
+    'Inet4RouteExcludeAddress': 4,
+    'Inet6RouteExcludeAddress': 6,
+}
+_TUN_FIELDS = {*_PREFIXES, *_OWNED, 'Name', 'MTU', 'StrictRoute', 'InterfaceScope'}
+_STACK_FIELDS = {
+    'UDPTimeout',
+    'ICMPTimeout',
+    'UDPNATMax',
+    'UDPMapping',
+    'UDPFiltering',
+    'ForwarderBindInterface',
+    'IncludeAllNetworks',
+    'TCPCongestionControl',
+}
+
+
+class SingTUNHostSettingsCallers:
+    """Bind sing-tun's named host settings callers to a captured host document."""
+
+    def __init__(self, settings):
+        if not isinstance(settings, dict):
+            raise ValueError('host_options must be an object')
+
+        (
+            self.userPrimaryAdapterInterfaceName,
+            self.userPrimaryAdapterInterfaceIP,
+            self.userTunAdapterInterfaceDNS,
+            self.userBypassTUNAdapterInterfaceIP,
+            self.userDisablePrimaryAdapterInterfaceDNS,
+        ) = (
+            functools.partial(
+                settings.get,
+                'primaryAdapterInterfaceName',
+                SING_TUN_HOST_DEFAULTS['primaryAdapterInterfaceName'],
+            ),
+            functools.partial(
+                settings.get,
+                'primaryAdapterInterfaceIP',
+                SING_TUN_HOST_DEFAULTS['primaryAdapterInterfaceIP'],
+            ),
+            functools.partial(
+                settings.get,
+                'tunAdapterInterfaceDNS',
+                SING_TUN_HOST_DEFAULTS['tunAdapterInterfaceDNS'],
+            ),
+            functools.partial(
+                settings.get,
+                'bypassTUNAdapterInterfaceIP',
+                SING_TUN_HOST_DEFAULTS['bypassTUNAdapterInterfaceIP'],
+            ),
+            functools.partial(
+                settings.get,
+                'disablePrimaryAdapterInterfaceDNS',
+                SING_TUN_HOST_DEFAULTS['disablePrimaryAdapterInterfaceDNS'],
+            ),
+        )
+
+
+def durationSeconds(value):
+    """Read Go duration strings (or integer nanoseconds), without native imports."""
+    if isinstance(value, int) and not isinstance(value, bool):
+        return value / 1e9
+
+    if not isinstance(value, str) or len(value) > 80:
+        raise ValueError('Use a Go duration such as 1m or 30s')
+
+    parts = re.findall(r'(\d+(?:\.\d+)?)(ns|us|µs|μs|ms|s|m|h)', value)
+
+    if not parts or ''.join(number + unit for number, unit in parts) != value:
+        raise ValueError('Use a Go duration such as 1m or 30s')
+
+    units = {
+        'ns': 1e-9,
+        'us': 1e-6,
+        'µs': 1e-6,
+        'μs': 1e-6,
+        'ms': 0.001,
+        's': 1,
+        'm': 60,
+        'h': 3600,
+    }
+
+    return sum(float(number) * units[unit] for number, unit in parts)
+
+
+def prepareSingTUNSettings(document, *, platform=None):
+    """Validate copied settings; the stored document is never normalized in place."""
+    if not isinstance(document, dict):
+        raise ValueError('sing-tun settings must be an object')
+
+    if set(document) - set(SING_TUN_DEFAULTS):
+        raise ValueError('Unsupported sing-tun settings fields')
+
+    result = copy.deepcopy(SING_TUN_DEFAULTS)
+
+    for key, value in document.items():
+        if key in ('tun_options', 'stack_options', 'host_options'):
+            if not isinstance(value, dict):
+                raise ValueError(key + ' must be an object')
+
+            result[key].update(copy.deepcopy(value))
+        else:
+            result[key] = copy.deepcopy(value)
+
+    if result['stack'] not in ('', 'go', 'gvisor', 'system', 'mixed'):
+        raise ValueError('Unsupported sing-tun stack')
+
+    if result['log_level'] not in ('trace', 'debug', 'info', 'warn', 'error', 'silent'):
+        raise ValueError('Invalid sing-tun log level')
+
+    for key in ('network_interface',):
+        if (
+            not isinstance(result[key], str)
+            or len(result[key]) > 256
+            or any(c in result[key] for c in '\x00\r\n')
+        ):
+            raise ValueError('Invalid ' + key)
+
+    sessions = result['max_sessions']
+
+    if type(sessions) is not int or not 1 <= sessions <= 16384:
+        raise ValueError('max_sessions must be between 1 and 16384')
+
+    for key in ('udp_timeout', 'connect_timeout', 'tcp_idle_timeout'):
+        value = result[key]
+
+        if (
+            type(value) not in (int, float)
+            or not math.isfinite(value)
+            or not 0.05 <= value <= 86400
+        ):
+            raise ValueError(key + ' must be between 0.05 and 86400 seconds')
+
+    tun, stack = result['tun_options'], result['stack_options']
+
+    if set(tun) - _TUN_FIELDS or set(stack) - _STACK_FIELDS:
+        raise ValueError('Unsupported or application-owned native option')
+
+    for key, expected in _OWNED.items():
+        if type(tun[key]) is not type(expected) or tun[key] != expected:
+            raise ValueError(key + ' is managed by Furious')
+
+    if not isinstance(tun['Name'], str) or not re.fullmatch(
+        r'[a-zA-Z0-9_.-]{0,32}', tun['Name']
+    ):
+        raise ValueError('Invalid TUN device name')
+
+    if type(tun['MTU']) is not int or not 68 <= tun['MTU'] <= 65535:
+        raise ValueError('MTU must be between 68 and 65535')
+
+    for key, family in _PREFIXES.items():
+        if key not in tun:
+            continue
+
+        if not isinstance(tun[key], list) or len(tun[key]) > 256:
+            raise ValueError(key + ' must be a bounded prefix list')
+
+        for prefix in tun[key]:
+            try:
+                if (
+                    not isinstance(prefix, str)
+                    or ipaddress.ip_interface(prefix).version != family
+                ):
+                    raise ValueError()
+            except (ValueError, TypeError):
+                raise ValueError('Invalid prefix in ' + key) from None
+
+    if not tun['Inet4Address'] and not tun['Inet6Address']:
+        raise ValueError('At least one TUN address prefix is required')
+
+    if tun['Inet6Address'] and tun['MTU'] < 1280:
+        raise ValueError('IPv6 requires an MTU of at least 1280 bytes')
+
+    if tun.get('StrictRoute') and (not tun['Inet4Address'] or not tun['Inet6Address']):
+        raise ValueError(
+            'StrictRoute requires both address families to preserve remote bypasses'
+        )
+
+    if platform == 'Darwin' and tun.get('InterfaceScope'):
+        raise ValueError(
+            'Scoped macOS routes require binding support for scoped route cleanup'
+        )
+
+    for options, keys in (
+        (tun, ('StrictRoute', 'InterfaceScope')),
+        (stack, ('ForwarderBindInterface', 'IncludeAllNetworks')),
+    ):
+        for key in keys:
+            if key in options and type(options[key]) is not bool:
+                raise ValueError(key + ' must be a boolean')
+
+    if stack.get('IncludeAllNetworks') and result['stack'] != 'gvisor':
+        raise ValueError('IncludeAllNetworks requires an explicit gvisor stack')
+
+    if stack.get('ForwarderBindInterface') and result['stack'] in ('', 'go'):
+        raise ValueError('ForwarderBindInterface is not used by the go stack')
+
+    congestion = stack.get('TCPCongestionControl', '')
+
+    if type(congestion) is not str or congestion not in ('', 'cubic', 'reno', 'bbr'):
+        raise ValueError('Invalid TCPCongestionControl')
+
+    if congestion and result['stack'] not in ('', 'go'):
+        raise ValueError('TCPCongestionControl requires go or automatic stack')
+
+    for key in ('UDPTimeout', 'ICMPTimeout'):
+        if key in stack:
+            seconds = durationSeconds(stack[key])
+
+            if (
+                not math.isfinite(seconds)
+                or seconds < 0
+                or (key == 'UDPTimeout' and seconds == 0)
+                or seconds > 86400
+            ):
+                raise ValueError('Invalid ' + key)
+
+    for key, maximum in (
+        ('UDPNATMax', 4294967295),
+        ('UDPMapping', 2),
+        ('UDPFiltering', 2),
+    ):
+        if key in stack and (
+            type(stack[key]) is not int or not 0 <= stack[key] <= maximum
+        ):
+            raise ValueError('Invalid ' + key)
+
+    result['host_options'] = prepareSingHostSettings(result['host_options'])
+
+    return result
+
+
+def prepareSingHostSettings(document):
+    """Validate sing-tun's own DNS, physical adapter and bypass preferences."""
+    if not isinstance(document, dict):
+        raise ValueError('sing-tun host options must be an object')
+
+    if set(document) - set(SING_TUN_HOST_DEFAULTS):
+        raise ValueError('Unsupported sing-tun host options')
+
+    result = copy.deepcopy(SING_TUN_HOST_DEFAULTS)
+    result.update(copy.deepcopy(document))
+
+    for key in (
+        'primaryAdapterInterfaceName',
+        'primaryAdapterInterfaceIP',
+        'tunAdapterInterfaceDNS',
+        'bypassTUNAdapterInterfaceIP',
+    ):
+        value = result[key]
+
+        if (
+            not isinstance(value, str)
+            or len(value) > 8192
+            or any(char in value for char in '\x00\r\n')
+        ):
+            raise ValueError('Invalid sing-tun host field: ' + key)
+
+        if key.endswith('IP') or key == 'tunAdapterInterfaceDNS':
+            try:
+                addresses = (
+                    value.split(',')
+                    if key == 'bypassTUNAdapterInterfaceIP'
+                    else [value]
+                )
+
+                for address in addresses if value else []:
+                    ipaddress.ip_address(address.strip())
+            except ValueError:
+                raise ValueError(
+                    'Invalid IP address in sing-tun host field: ' + key
+                ) from None
+
+    if type(result['disablePrimaryAdapterInterfaceDNS']) is not bool:
+        raise ValueError('sing-tun primary adapter DNS mitigation must be a boolean')
+
+    return result
+
+
+def addSingTUNExclusions(settings, addresses):
+    """Merge attempt-only outbound exclusions with explicit user ranges."""
+    prepared = copy.deepcopy(settings)
+
+    for text in addresses:
+        address = ipaddress.ip_address(text)
+        key = (
+            'Inet4RouteExcludeAddress'
+            if address.version == 4
+            else 'Inet6RouteExcludeAddress'
+        )
+        entries = prepared['tun_options'].setdefault(key, [])
+        prefix = str(address) + ('/32' if address.version == 4 else '/128')
+
+        if prefix not in entries:
+            entries.append(prefix)
+
+    return prepared
@@ -20,10 +20,12 @@
 from __future__ import annotations

 from Furious.Frozenlib import *
+from Furious.Frozenlib.SocksProxy import socksURL

 from PySide6.QtNetwork import *

 from typing import Union
+from urllib.parse import urlsplit, unquote

 import weakref
 import functools
@@ -85,3 +87,20 @@ class AppQNetworkAccessManager(QNetworkAccessManager):
            self.setProxy(QNetworkProxy.ProxyType.NoProxy)

        return useProxy
+
+    def configureSocksProxy(self, endpoint: str):
+        """Use validated SOCKS transit without a direct-network fallback."""
+
+        proxy = urlsplit(socksURL(endpoint))
+
+        self.setProxy(
+            QNetworkProxy(
+                QNetworkProxy.ProxyType.Socks5Proxy,
+                proxy.hostname,
+                proxy.port,
+                unquote(proxy.username or ''),
+                unquote(proxy.password or ''),
+            )
+        )
+
+        return True
@@ -0,0 +1,82 @@
+# 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/>.
+
+"""Persist sing-tun's independent native and host configuration."""
+
+from __future__ import annotations
+
+from Furious.Frozenlib import AppSettings, Mixins, registerAppSettings
+from Furious.Interface import StorageBackend
+from Furious.Models.Encoding import PyBase64Encoder, UJSONEncoder
+
+import logging
+
+__all__ = ['UserSingTUNSettings']
+
+logger = logging.getLogger(__name__)
+
+# Keep sing-tun customizations independent of the legacy CustomTUNSettings.
+# 0.8.2 ignores this key and preserves it during ordinary settings writes;
+# an explicit settings reset clears both engines' customizations.
+registerAppSettings('CustomSingTUNSettings')
+
+
+class UserSingTUNSettings(Mixins.CleanupOnExit, StorageBackend):
+    """Own only the sing-tun document, including its host options."""
+
+    def __init__(self, *args, **kwargs):
+        super().__init__(*args, **kwargs)
+
+        self._restoreFailed = False
+
+        raw = AppSettings.get('CustomSingTUNSettings')
+
+        self._data = {}
+
+        if raw is None:
+            return
+
+        try:
+            data = UJSONEncoder.decode(PyBase64Encoder.decode(raw))
+
+            if not isinstance(data, dict):
+                raise TypeError('sing-tun settings repository root must be an object')
+
+            self._data = data
+        except Exception:
+            self._restoreFailed = True
+
+            logger.exception('failed to restore persisted sing-tun settings')
+
+    def data(self) -> dict:
+        return self._data
+
+    def sync(self):
+        AppSettings.set(
+            'CustomSingTUNSettings',
+            PyBase64Encoder.encode(UJSONEncoder.encode(self._data).encode()),
+        )
+
+        self._restoreFailed = False
+
+    def cleanup(self):
+        if self._restoreFailed and not self._data:
+            logger.warning('preserving unreadable sing-tun settings during cleanup')
+
+            return
+
+        self.sync()
@@ -26,10 +26,16 @@ from Furious.Repository.Routings import UserRoutings
 from Furious.Repository.Servers import UserServers
 from Furious.Repository.Subscriptions import SubscriptionGroup, UserSubs
 from Furious.Repository.TunSettings import UserTUNSettings
+from Furious.Repository.SingTUNSettings import UserSingTUNSettings
+from Furious.Models.SingTUN import prepareSingTUNSettings
+
+from PySide6 import QtCore

 from typing import Union

 import functools
+import copy
+import json

 __all__ = ['Storage']

@@ -61,6 +67,44 @@ class Storage:

        return UserTUNSettings()

+    @staticmethod
+    @functools.lru_cache(None)
+    def _UserSingTUNSettingsStorage() -> UserSingTUNSettings:
+        """Return the single sing-tun customization repository."""
+        assert APP() is not None
+
+        return UserSingTUNSettings()
+
+    @staticmethod
+    def UserSingTUNSettings() -> dict:
+        """Expose the compatibility live view of sing-tun settings."""
+        return Storage._UserSingTUNSettingsStorage().data()
+
+    @staticmethod
+    def replaceSingTUNSettings(singSettings):
+        """Commit sing-tun's validated document, then report its flush result."""
+
+        prepareSingTUNSettings(singSettings)
+
+        # Serialization is a fallible pre-commit stage.
+        json.dumps(singSettings, allow_nan=False)
+
+        singCandidate = copy.deepcopy(singSettings)
+
+        sing = Storage._UserSingTUNSettingsStorage()
+        sing.data().clear()
+        sing.data().update(singCandidate)
+
+        sing.sync()
+
+        settings = QtCore.QSettings()
+        settings.sync()
+
+        if settings.status() != QtCore.QSettings.Status.NoError:
+            raise OSError(
+                'sing-tun settings committed in memory, but could not be flushed to disk'
+            )
+
    @staticmethod
    @functools.lru_cache(None)
    def _UserRoutingsStorage() -> UserRoutings:
@@ -29,6 +29,8 @@ __all__ = ['UserTUNSettings']

 logger = logging.getLogger(__name__)

+# Preserve the legacy tun2socks key and Base64/JSON format used by 0.8.2.
+# These values remain tun2socks-specific; sing-tun owns a separate document.
 registerAppSettings('CustomTUNSettings')


@@ -25,13 +25,26 @@ for execution, and Qt for lifetime primitives. This scope owns multi-stage workf
 ## Connection and network workflows

 - GUI connection startup is a generation-checked transaction over a runtime copy: prepare TUN policy, launch and
-  observe the primary runtime, acquire optional tun2socks/DNS resources, and mutate host networking in platform
-  order before commit. Preserve Windows runtime-before-device, Linux device-before-runtime, and macOS
+  observe the primary runtime, acquire optional application TUN/DNS resources, and mutate host networking in platform
+  order before commit. The tun2socks path preserves Windows runtime-before-device, Linux device-before-runtime, and macOS
  survival-before-DNS ordering. Failure/cancellation releases attempt-owned runtimes and registered host cleanup.
  The synchronous start path remains a compatibility boundary. Review GUI responsiveness at each stage, including
  factory preparation, host commands, and reverse cleanup: a scheduled start only defers the first call. Readiness
  timers and asynchronous DNS cannot preempt synchronous work. Keep cancellation checks at reentrant stage boundaries
  before acquiring the next resource, and audit shared route bookkeeping separately from attempt-local leases.
+- Native proxy-core TUN policy precedes application-engine selection. The legacy plugin opt-in
+  `usesApplicationTun2socks` still means application-TUN eligibility; it must not force the selected engine.
+  Snapshot the engine and relevant customization once per attempt; bind each backend's named settings callers to
+  that snapshot, preserving proxy-only and explicit native TUN
+  behavior. Each application engine reads only its own persisted document: sing-tun's `host_options` belong to
+  `CustomSingTUNSettings`, while tun2socks uses `CustomTUNSettings`; edits and missing defaults must never import
+  preferences from the other engine. Their repository/host policies remain independent while orchestration,
+  leases and general utilities may be reused. For sing-tun, resolve automatic and manual remote exclusions before native auto-routing, then require
+  native readiness and checked DNS application before commit. `SingTUNHostPlan` owns only its assigned identifiers
+  and DNS snapshots, independently of legacy `SystemRoutingTable.managedRoutes`; refuse ambiguous recovery and
+  retain ownership for retry. Host preparation/DNS commands run in its owned worker, while synchronous compatibility
+  startup and bounded cleanup joins remain explicit responsiveness limits. Consult `tests/test_sing_tun.py` and
+  `tests/test_native_tun_semantics.py`; mocked host tests do not establish privileged OS behavior.
 - Construct a runtime event router before asking a plugin to create its runtime. One lease owns the runtime/router from
  acquisition through attempt ownership, commit, and reverse-order release; commit changes logical delivery without
  replacing the runtime callback. Worker-thread exits are queued to the router's Qt thread, delivered at most once, and
@@ -74,6 +87,10 @@ for execution, and Qt for lifetime primitives. This scope owns multi-stage workf
 - Log cursors are opaque and filter-specific. A generation change requires a reset; retention-only eviction supplies
  a new first-retained sequence so presenters can prune their prefix without rebuilding history. Capture entries and
  the next cursor atomically, and coalesce notifications without losing producer updates.
+- Application TUN logs share an engine-neutral runtime category. Bind each producer's source to the connection
+  attempt's captured engine selection rather than a later preference read; proxy-core native TUN output stays with
+  the Core category. Preserve batching, export/rendering labels, and runtime clearing together. Tests in
+  `test_sing_tun.py`, `test_log_manager_generation.py`, and `test_ui_behavior.py` cover these boundaries.
 - Metrics sampling owns its worker/future generation and rejects results after disconnect, disablement, replacement,
  or shutdown. Normalize cumulative-counter resets before history aggregation; clearing usage must not erase speed
  history.
@@ -25,6 +25,8 @@ from Furious.Frozenlib import (
    APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS,
    APPLICATION_TUN2SOCKS_IP_ADDRESS,
    APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME,
+    APPLICATION_TUN_BACKEND,
+    AppSettings,
    Mixins,
    PLATFORM,
    PySide6Legacy,
@@ -45,17 +47,25 @@ from Furious.Plugins import (
 from Furious.Qt.Signals import connectWeakly, singleShotWeakly
 from Furious.Service.DnsResolver import DnsResolver
 from Furious.Core.Tun2socks import Tun2socks
+from Furious.Core.SingTUN import SingTUN
+from Furious.Models.SingTUN import SingTUNHostSettingsCallers, prepareSingTUNSettings
+from Furious.Frozenlib.SocksProxy import socksURL
+from Furious.Service.SingTUNHost import SingTUNHostPlan
 from Furious.Service.RuntimeLease import RuntimeEventRouter, RuntimeLease

+from PySide6 import QtCore, QtNetwork
+
 from typing import Callable, Tuple, Union
 from dataclasses import dataclass, field
 from enum import Enum

-from PySide6 import QtCore, QtNetwork
 import os
 import logging
 import tempfile
 import functools
+import copy
+import time
+import ipaddress

 __all__ = ['ConnectionManager', 'ConnectionStartOperation', 'ConnectionStartStage']

@@ -72,6 +82,37 @@ class _ConnectionStartAttempt:
    nativeTUNHandled: bool = False
    applicationTun2socks: bool = False
    committed: bool = False
+    tunBackend: str = field(
+        default_factory=lambda: AppSettings.get(APPLICATION_TUN_BACKEND)
+    )
+    singSettings: dict = field(default_factory=dict)
+    tun2socksSettings: dict = field(default_factory=dict)
+    singHostSettingsCallers: SingTUNHostSettingsCallers | None = field(
+        default=None, init=False, repr=False
+    )
+    tun2socksSettingsCallers: _Tun2socksSettingsCallers | None = field(
+        default=None, init=False, repr=False
+    )
+
+    def applicationTUNLogCallback(self, callback):
+        """Bind the engine captured by this attempt, preserving plain callbacks."""
+        copyWithSource = getattr(callback, 'copyWithSource', None)
+
+        return copyWithSource(self.tunBackend) if callable(copyWithSource) else callback
+
+    def snapshotApplicationTUN(self):
+        """Read engine settings only when the plugin requires application TUN."""
+        if self.applicationTun2socks:
+            if self.tunBackend == 'sing-tun':
+                self.singSettings = copy.deepcopy(Storage.UserSingTUNSettings())
+                self.singHostSettingsCallers = SingTUNHostSettingsCallers(
+                    self.singSettings.get('host_options', {})
+                )
+            else:
+                self.tun2socksSettings = copy.deepcopy(Storage.UserTUNSettings())
+                self.tun2socksSettingsCallers = _Tun2socksSettingsCallers(
+                    self.tun2socksSettings
+                )

    @property
    def runtimes(self):
@@ -120,32 +161,33 @@ class _ConnectionStartAttempt:
        self.committed = True


-def getUserTUNSettings(*args, **kwargs):
-    """Return one user-defined TUN setting or its fallback value."""
-    return Storage.UserTUNSettings().get(*args, **kwargs)
+class _Tun2socksSettingsCallers:
+    """Bind named tun2socks settings callers to one attempt's captured document."""

-
-(
-    userPrimaryAdapterInterfaceName,
-    userPrimaryAdapterInterfaceIP,
-    userDefaultPrimaryGatewayIP,
-    userTunAdapterInterfaceDNS,
-    userBypassTUNAdapterInterfaceIP,
-    userDisablePrimaryAdapterInterfaceDNS,
-    userTcpSendBufferSize,
-    userTcpReceiveBufferSize,
-    userTcpAutoTuning,
-) = (
-    functools.partial(getUserTUNSettings, 'primaryAdapterInterfaceName', ''),
-    functools.partial(getUserTUNSettings, 'primaryAdapterInterfaceIP', ''),
-    functools.partial(getUserTUNSettings, 'defaultPrimaryGatewayIP', ''),
-    functools.partial(getUserTUNSettings, 'tunAdapterInterfaceDNS', ''),
-    functools.partial(getUserTUNSettings, 'bypassTUNAdapterInterfaceIP', ''),
-    functools.partial(getUserTUNSettings, 'disablePrimaryAdapterInterfaceDNS', 'True'),
-    functools.partial(getUserTUNSettings, 'tcpSendBufferSize', 1),
-    functools.partial(getUserTUNSettings, 'tcpReceiveBufferSize', 1),
-    functools.partial(getUserTUNSettings, 'tcpAutoTuning', 'False'),
-)
+    def __init__(self, settings):
+        (
+            self.userPrimaryAdapterInterfaceName,
+            self.userPrimaryAdapterInterfaceIP,
+            self.userDefaultPrimaryGatewayIP,
+            self.userTunAdapterInterfaceDNS,
+            self.userBypassTUNAdapterInterfaceIP,
+            self.userDisablePrimaryAdapterInterfaceDNS,
+            self.userTcpSendBufferSize,
+            self.userTcpReceiveBufferSize,
+            self.userTcpAutoTuning,
+        ) = (
+            functools.partial(settings.get, 'primaryAdapterInterfaceName', ''),
+            functools.partial(settings.get, 'primaryAdapterInterfaceIP', ''),
+            functools.partial(settings.get, 'defaultPrimaryGatewayIP', ''),
+            functools.partial(settings.get, 'tunAdapterInterfaceDNS', ''),
+            functools.partial(settings.get, 'bypassTUNAdapterInterfaceIP', ''),
+            functools.partial(
+                settings.get, 'disablePrimaryAdapterInterfaceDNS', 'True'
+            ),
+            functools.partial(settings.get, 'tcpSendBufferSize', 1),
+            functools.partial(settings.get, 'tcpReceiveBufferSize', 1),
+            functools.partial(settings.get, 'tcpAutoTuning', 'False'),
+        )


 class ConnectionStartStage(Enum):
@@ -221,10 +263,21 @@ class _RuntimeReadinessProbe(QtCore.QObject):
        if self._terminal:
            return

+        if isinstance(self._runtime, SingTUN):
+            if self._runtime.startupError:
+                self._finishFailed(self._runtime.startupError)
+
+                return
+
+            if self._runtime.ready:
+                self._finishReady()
+
+                return
+
        timeout = max(int(self._startup.timeout), 1)

        if self._elapsed.isValid() and self._elapsed.elapsed() >= timeout:
-            if self._startup.endpoint:
+            if self._startup.endpoint or isinstance(self._runtime, SingTUN):
                self._finishFailed('core readiness check timed out')
            else:
                self._finishReady()
@@ -377,7 +430,6 @@ class ConnectionStartOperation(QtCore.QObject):
        self.routing = routing
        self.exitCallback = exitCallback
        self.msgCallbackCore = msgCallbackCore
-        self.msgCallbackTUN_ = msgCallbackTUN_
        self.proxyModeOnly = proxyModeOnly
        self.log = log
        self.options = dict(options or {})
@@ -385,6 +437,7 @@ class ConnectionStartOperation(QtCore.QObject):
        self.attempt = _ConnectionStartAttempt(
            manager, manager._runtimeConfiguration(config, deepcopy)
        )
+        self.msgCallbackTUN_ = self.attempt.applicationTUNLogCallback(msgCallbackTUN_)

        self.stage = ConnectionStartStage.Pending

@@ -392,6 +445,7 @@ class ConnectionStartOperation(QtCore.QObject):
        self._readinessProbe = None
        self._conditionProbe = None
        self._conditionContinuation = ''
+        self._conditionFailureMessage = 'TUN device did not become ready'
        self._dnsOperation = None
        self._tun = None
        self._startTUN = None
@@ -443,6 +497,8 @@ class ConnectionStartOperation(QtCore.QObject):
                self.attempt.nativeTUNHandled,
                self.attempt.applicationTun2socks,
            ) = self.manager._prepareTUNPolicy(configcopy, self.proxyModeOnly)
+
+            self.attempt.snapshotApplicationTUN()
        except TUNPreparationError as ex:
            self.manager._lastStartError = str(ex)

@@ -608,6 +664,11 @@ class ConnectionStartOperation(QtCore.QObject):
        if not self._isCurrent():
            return

+        if self.attempt.tunBackend == 'sing-tun':
+            self._resume('_beginSingTUN')
+
+            return
+
        self._setStage(ConnectionStartStage.PreparingTUN)

        if not self._isCurrent():
@@ -621,8 +682,9 @@ class ConnectionStartOperation(QtCore.QObject):
                APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS,
            )

-        userGateway = userDefaultPrimaryGatewayIP()
-        userInterfaceIP = userPrimaryAdapterInterfaceIP()
+        settings = self.attempt.tun2socksSettingsCallers
+        userGateway = settings.userDefaultPrimaryGatewayIP()
+        userInterfaceIP = settings.userPrimaryAdapterInterfaceIP()

        if userGateway and userInterfaceIP:
            logger.info(
@@ -663,9 +725,9 @@ class ConnectionStartOperation(QtCore.QObject):

                return

-        tcpSendBufferSize = userTcpSendBufferSize()
-        tcpReceiveBufferSize = userTcpReceiveBufferSize()
-        tcpAutoTuning = userTcpAutoTuning() == 'True'
+        tcpSendBufferSize = settings.userTcpSendBufferSize()
+        tcpReceiveBufferSize = settings.userTcpReceiveBufferSize()
+        tcpAutoTuning = settings.userTcpAutoTuning() == 'True'
        interfaceArg = (
            APPLICATION_TUN2SOCKS_NETWORK_INTERFACE_NAME
            if PLATFORM != 'Linux'
@@ -715,12 +777,132 @@ class ConnectionStartOperation(QtCore.QObject):
        else:
            self._resume('_prepareTunBypass')

+    def _beginSingTUN(self):
+        """Validate before activation, then resolve remote exclusions via the core."""
+        self._setStage(ConnectionStartStage.PreparingTUN)
+
+        if not self._isCurrent():
+            return
+
+        router = RuntimeEventRouter()
+
+        try:
+            self._tun = self.manager._createSingTUN(
+                self.attempt, router.publish, self.msgCallbackTUN_
+            )
+        except BaseException:
+            router.finishRelease()
+            router.deleteLater()
+
+            raise
+
+        router.attach(self._tun, self)
+
+        self.attempt.ownRuntime(self._tun, router)
+
+        address = self.attempt.runtimeConfiguration.remoteAddress()
+
+        if isValidIPAddress(address):
+            self._singAddressesResolved(False, [address])
+
+            return
+        if not address:
+            raise ValueError('A remote destination is required for sing-tun bypass')
+
+        resolver = self.manager._connectionDnsResolver()
+
+        self.manager._configureTunResolver(resolver, self.attempt.runtimeConfiguration)
+
+        self._setStage(ConnectionStartStage.ResolvingTUNAddress)
+
+        if not self._isCurrent():
+            return
+
+        operation = resolver.resolveAsync(address, parent=self)
+
+        self._dnsOperation = operation
+
+        connectWeakly(operation.finished, self, '_singAddressesResolved')
+
+        operation.start()
+
+    def _singAddressesResolved(self, error, addresses):
+        if not self._isCurrent():
+            return
+
+        operation, self._dnsOperation = self._dnsOperation, None
+
+        if operation is not None:
+            operation.deleteLater()
+
+        if error or not addresses:
+            self._fail(
+                'Failed to prepare sing-tun', 'Remote bypass DNS resolution failed'
+            )
+
+            return
+
+        try:
+            addresses = self.manager._singBypassAddresses(self.attempt, addresses)
+
+            self._tun._hostPlan.begin('prepare', addresses)
+            self._waitSingHost('_startSingTUN')
+        except Exception as ex:
+            self._fail('Failed to prepare sing-tun', str(ex))
+
+    def _waitSingHost(self, continuation):
+        if not self._isCurrent():
+            return
+
+        self._observeCondition(
+            self._tun._hostPlan.workCompleted,
+            'sing-tun host preparation',
+            continuation,
+            self.stage,
+            'sing-tun host preparation timed out',
+        )
+
+    def _startSingTUN(self):
+        if self._tun._hostPlan.error:
+            raise RuntimeStartError(self._tun._hostPlan.error)
+
+        self._setStage(ConnectionStartStage.StartingTUNRuntime)
+
+        if not self._isCurrent():
+            return
+
+        self._tun.start()
+
+        self._observeRuntime(
+            self._tun,
+            CoreRuntimeStartup(timeout=15000),
+            '_applySingDNS',
+            ConnectionStartStage.WaitingTUNRuntime,
+        )
+
+    def _applySingDNS(self):
+        self._setStage(ConnectionStartStage.ApplyingHostNetwork)
+
+        if not self._isCurrent():
+            return
+
+        self._tun._hostPlan.begin('applyDNS', self._tun.deviceName)
+        self._waitSingHost('_singHostReady')
+
+    def _singHostReady(self):
+        if self._tun._hostPlan.error:
+            raise RuntimeStartError(self._tun._hostPlan.error)
+
+        self._commit()
+
    def _prepareTunBypass(self):
        """Prepare remote-server bypass routes without blocking for DNS."""
        if not self._isCurrent():
            return

-        bypassTUN = userBypassTUNAdapterInterfaceIP()
+        bypassTUN = (
+            self.attempt.tun2socksSettingsCallers.userBypassTUNAdapterInterfaceIP()
+        )

        if bypassTUN:
            for bypass in bypassTUN.split(','):
@@ -814,19 +996,32 @@ class ConnectionStartOperation(QtCore.QObject):

    def _waitForTunDevice(self, predicate, continuation):
        """Wait for one platform TUN device through a Qt timer."""
+        self._observeCondition(
+            predicate,
+            f'TUN device {APPLICATION_TUN2SOCKS_DEVICE_NAME!r}',
+            continuation,
+            ConnectionStartStage.WaitingTUNDevice,
+            'TUN device did not become ready',
+        )
+
+    def _observeCondition(
+        self, predicate, description, continuation, stage, failureMessage
+    ):
+        """Observe an owned worker or device without inventing device readiness."""
        if not self._isCurrent():
            return

-        self._setStage(ConnectionStartStage.WaitingTUNDevice)
+        self._setStage(stage)

        if not self._isCurrent():
            return

        self._conditionContinuation = continuation
+        self._conditionFailureMessage = failureMessage

        probe = _ConditionProbe(
            predicate,
-            f'TUN device {APPLICATION_TUN2SOCKS_DEVICE_NAME!r}',
+            description,
            parent=self,
        )

@@ -850,7 +1045,7 @@ class ConnectionStartOperation(QtCore.QObject):
        probe.deleteLater()

        if not success:
-            self._fail('', 'TUN device did not become ready')
+            self._fail('', self._conditionFailureMessage)

            return

@@ -863,7 +1058,8 @@ class ConnectionStartOperation(QtCore.QObject):

        self._setStage(ConnectionStartStage.ApplyingHostNetwork)

-        userInterfaceName = userPrimaryAdapterInterfaceName()
+        settings = self.attempt.tun2socksSettingsCallers
+        userInterfaceName = settings.userPrimaryAdapterInterfaceName()
        alias = (
            userInterfaceName
            if userInterfaceName
@@ -878,7 +1074,7 @@ class ConnectionStartOperation(QtCore.QObject):

            self._tun.cleanup = functools.partial(_windowsCleanup, alias)

-            if userDisablePrimaryAdapterInterfaceDNS() != 'False':
+            if settings.userDisablePrimaryAdapterInterfaceDNS() != 'False':
                SystemRoutingTable.WIN32SetInterfaceDNS(
                    alias,
                    '127.0.0.1',
@@ -886,7 +1082,8 @@ class ConnectionStartOperation(QtCore.QObject):
                )

        interfaceDNS = (
-            userTunAdapterInterfaceDNS() or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS
+            settings.userTunAdapterInterfaceDNS()
+            or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS
        )

        SystemRoutingTable.addRelations()
@@ -928,7 +1125,8 @@ class ConnectionStartOperation(QtCore.QObject):
        self._tun.cleanup = functools.partial(_darwinCleanup, servers)

        interfaceDNS = (
-            userTunAdapterInterfaceDNS() or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS
+            self.attempt.tun2socksSettingsCallers.userTunAdapterInterfaceDNS()
+            or APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS
        )

        for service, _dnsserver in servers:
@@ -1199,12 +1397,12 @@ class ConnectionManager(Mixins.CleanupOnExit):

        if useAppTun2socks:
            logger.info(
-                f'application-managed tun2socks selected. Remote '
+                f'application-managed TUN selected. Remote '
                f'address: {config.remoteAddress()!r}'
            )
        else:
            logger.info(
-                'application-managed tun2socks skipped by the active '
+                'application-managed TUN skipped by the active '
                'core runtime configuration'
            )

@@ -1389,6 +1587,10 @@ class ConnectionManager(Mixins.CleanupOnExit):

        tunModeRequested = not proxyModeOnly and SystemRuntime.isTUNMode()

+        attempt.snapshotApplicationTUN()
+
+        msgCallbackTUN_ = attempt.applicationTUNLogCallback(msgCallbackTUN_)
+
        if not self._startPrimaryRuntime(
            attempt,
            routing,
@@ -1401,15 +1603,110 @@ class ConnectionManager(Mixins.CleanupOnExit):
            return abortStart()

        if tunModeRequested and attempt.applicationTun2socks:
-            if not self._startApplicationTun2socks(
-                attempt, exitCallback, msgCallbackTUN_
-            ):
+            startApplicationTUN = (
+                self._startApplicationSingTUN
+                if attempt.tunBackend == 'sing-tun'
+                else self._startApplicationTun2socks
+            )
+
+            if not startApplicationTUN(attempt, exitCallback, msgCallbackTUN_):
                return False

        attempt.commit()

        return True

+    @staticmethod
+    def _singBypassAddresses(attempt, addresses):
+        result = list(addresses)
+        manual = attempt.singHostSettingsCallers.userBypassTUNAdapterInterfaceIP()
+
+        if manual:
+            result.extend(value.strip() for value in manual.split(','))
+
+        try:
+            return list(
+                dict.fromkeys(str(ipaddress.ip_address(value)) for value in result)
+            )
+        except ValueError:
+            raise ValueError('Invalid application TUN bypass IP address') from None
+
+    @staticmethod
+    def _configureTunResolver(resolver, configuration):
+        # SOCKS transit carries the prepared listener's authentication policy.
+        # Never require HTTP or fall back to a direct bootstrap request.
+        getter = getattr(configuration, 'applicationTUNProxy', None)
+
+        resolver.configureSocksProxy(
+            getter() if callable(getter) else socksURL(configuration.socksProxy())
+        )
+
+    @staticmethod
+    def _createSingTUN(attempt, exitCallback, messageCallback):
+        configuration = prepareSingTUNSettings(attempt.singSettings, platform=PLATFORM)
+        getter = getattr(attempt.runtimeConfiguration, 'applicationTUNProxy', None)
+
+        configuration['proxy'] = socksURL(
+            getter() if callable(getter) else attempt.runtimeConfiguration.socksProxy()
+        )
+
+        plan = SingTUNHostPlan(configuration)
+
+        return SingTUN(
+            plan.configuration,
+            hostPlan=plan,
+            exitCallback=exitCallback,
+            msgCallback=messageCallback,
+        )
+
+    def _startApplicationSingTUN(self, attempt, exitCallback, messageCallback):
+        """Compatibility startup shares configuration, bypass and host ownership."""
+        try:
+            runtime = self._createSingTUN(attempt, exitCallback, messageCallback)
+
+            attempt.ownRuntime(runtime)
+
+            address = attempt.runtimeConfiguration.remoteAddress()
+
+            if isValidIPAddress(address):
+                addresses = [address]
+            else:
+                resolver = self._connectionDnsResolver()
+
+                self._configureTunResolver(resolver, attempt.runtimeConfiguration)
+
+                error, addresses = resolver.resolve(address)
+
+                if error or not addresses:
+                    raise ValueError('Remote bypass DNS resolution failed')
+
+            plan = runtime._hostPlan
+            plan.prepare(self._singBypassAddresses(attempt, addresses))
+
+            runtime.start()
+
+            deadline = time.monotonic() + 15
+
+            while not runtime.ready:
+                if (
+                    runtime.startupError
+                    or not runtime.isRunning()
+                    or time.monotonic() >= deadline
+                ):
+                    raise RuntimeStartError(
+                        runtime.startupError or 'sing-tun readiness timed out'
+                    )
+
+                time.sleep(0.02)
+
+            plan.applyDNS(runtime.deviceName)
+
+            return True
+        except Exception as ex:
+            self._lastStartError = str(ex)
+
+            return attempt.rollback('sing-tun startup failed: ' + self._lastStartError)
+
    def _startApplicationTun2socks(
        self,
        attempt: _ConnectionStartAttempt,
@@ -1427,9 +1724,10 @@ class ConnectionManager(Mixins.CleanupOnExit):
            SystemRoutingTable.delete('0.0.0.0', APPLICATION_TUN2SOCKS_GATEWAY_ADDRESS)

        # Handle user defined settings
+        settings = attempt.tun2socksSettingsCallers
        userGateway, userInterfaceIP = (
-            userDefaultPrimaryGatewayIP(),
-            userPrimaryAdapterInterfaceIP(),
+            settings.userDefaultPrimaryGatewayIP(),
+            settings.userPrimaryAdapterInterfaceIP(),
        )

        if userGateway and userInterfaceIP:
@@ -1470,9 +1768,9 @@ class ConnectionManager(Mixins.CleanupOnExit):
                return abortStart(f'unrecognized platform: {PLATFORM}')

        tcpSendBufferSize, tcpReceiveBufferSize, tcpAutoTuning = (
-            userTcpSendBufferSize(),
-            userTcpReceiveBufferSize(),
-            userTcpAutoTuning(),
+            settings.userTcpSendBufferSize(),
+            settings.userTcpReceiveBufferSize(),
+            settings.userTcpAutoTuning(),
        )

        if tcpSendBufferSize != 1:
@@ -1526,7 +1824,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
                return abortStart(f'core {Tun2socks.name()} start failed')

        # Handle user defined settings
-        bypassTUN = userBypassTUNAdapterInterfaceIP()
+        bypassTUN = settings.userBypassTUNAdapterInterfaceIP()

        if bypassTUN:
            try:
@@ -1589,7 +1887,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
                return abortStart()

            # Handle user defined settings
-            userInterfaceName = userPrimaryAdapterInterfaceName()
+            userInterfaceName = settings.userPrimaryAdapterInterfaceName()

            if userInterfaceName:
                logger.info(
@@ -1616,7 +1914,9 @@ class ConnectionManager(Mixins.CleanupOnExit):
                tun.cleanup = functools.partial(_windowsCleanup, alias)

                # Handle user defined settings
-                userDisableInterfaceDNS = userDisablePrimaryAdapterInterfaceDNS()
+                userDisableInterfaceDNS = (
+                    settings.userDisablePrimaryAdapterInterfaceDNS()
+                )

                logger.info(f'DisablePrimaryInterfaceDNS: {userDisableInterfaceDNS}')

@@ -1624,7 +1924,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
                    SystemRoutingTable.WIN32SetInterfaceDNS(alias, '127.0.0.1', False)

            # Handle user defined settings
-            userTunInterfaceDNS = userTunAdapterInterfaceDNS()
+            userTunInterfaceDNS = settings.userTunAdapterInterfaceDNS()

            if userTunInterfaceDNS == '':
                userTunInterfaceDNS = APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS
@@ -1667,7 +1967,7 @@ class ConnectionManager(Mixins.CleanupOnExit):
            tun.cleanup = functools.partial(_darwinCleanup, servers)

            # Handle user defined settings
-            userTunInterfaceDNS = userTunAdapterInterfaceDNS()
+            userTunInterfaceDNS = settings.userTunAdapterInterfaceDNS()

            if userTunInterfaceDNS == '':
                userTunInterfaceDNS = APPLICATION_TUN2SOCKS_INTERFACE_DNS_ADDRESS
@@ -0,0 +1,786 @@
+# 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/>.
+
+"""Attempt-local host preparation and recovery for the native auto-route engine."""
+
+from __future__ import annotations
+
+from Furious.Frozenlib import (
+    PLATFORM,
+    SystemRuntime,
+    runExternalCommand,
+    SystemRoutingTable,
+)
+from Furious.Models.SingTUN import (
+    SingTUNHostSettingsCallers,
+    addSingTUNExclusions,
+    prepareSingTUNSettings,
+)
+
+import copy
+import ipaddress
+import json
+import os
+import secrets
+import shutil
+import socket
+import subprocess
+import threading
+import base64
+
+
+def _command(arguments):
+    """Run a checked, bounded argument vector without shell interpolation."""
+    try:
+        return (
+            runExternalCommand(
+                arguments,
+                stdout=subprocess.PIPE,
+                stderr=subprocess.PIPE,
+                check=True,
+                timeout=5,
+            )
+            .stdout.decode('utf-8', 'replace')
+            .strip()
+        )
+    except (OSError, subprocess.SubprocessError):
+        # Command output may contain host configuration; keep the UI diagnostic
+        # useful without forwarding it or profile credentials to logs.
+        raise RuntimeError('sing-tun host operation failed: ' + arguments[0]) from None
+
+
+def _powershell(script):
+    script = (
+        '[Console]::OutputEncoding=[System.Text.Encoding]::UTF8; '
+        '$OutputEncoding=[Console]::OutputEncoding; ' + script
+    )
+
+    return _command(
+        [
+            'powershell',
+            '-NoProfile',
+            '-NonInteractive',
+            '-EncodedCommand',
+            base64.b64encode(script.encode('utf-16le')).decode(),
+        ]
+    )
+
+
+def _darwinRoutes(text, family):
+    """Normalize netstat's abbreviated destinations, never matching substrings."""
+    result = []
+
+    for line in text.splitlines():
+        fields = line.split()
+
+        if len(fields) < 4:
+            continue
+
+        destination = fields[0]
+
+        try:
+            if destination == 'default':
+                destination = '0.0.0.0/0' if family == 4 else '::/0'
+            elif family == 4:
+                address, separator, mask = destination.partition('/')
+                octets = address.split('.')
+                destination = (
+                    '.'.join(octets + ['0'] * (4 - len(octets)))
+                    + '/'
+                    + (mask if separator else str(8 * len(octets)))
+                )
+
+            network = ipaddress.ip_network(destination, strict=False)
+        except ValueError:
+            continue
+
+        if network.version == family:
+            result.append((network, fields[1], fields[3]))
+
+    return result
+
+
+def _routeRanges(tun):
+    """Mirror the pinned Darwin BuildAutoRouteRanges for exact conflict checks."""
+    result = []
+
+    for family in (4, 6):
+        addresses = tun['Inet' + str(family) + 'Address']
+
+        if not addresses:
+            continue
+
+        routes = tun.get('Inet' + str(family) + 'RouteAddress', [])
+
+        if routes:
+            ranges = [ipaddress.ip_network(value, strict=False) for value in routes]
+            ranges.extend(
+                ipaddress.ip_interface(value).network
+                for value in addresses
+                if ipaddress.ip_interface(value).network.prefixlen < 32
+            )
+        else:
+            bits = 32 if family == 4 else 128
+            ranges = [
+                ipaddress.ip_network(
+                    (ipaddress.ip_address(1 << (bits - length)), length)
+                )
+                for length in range(8, 0, -1)
+            ]
+
+        for value in tun.get('Inet' + str(family) + 'RouteExcludeAddress', []):
+            exclusion = ipaddress.ip_network(value, strict=False)
+            remaining = []
+
+            for network in ranges:
+                if network.subnet_of(exclusion):
+                    continue
+
+                remaining.extend(
+                    network.address_exclude(exclusion)
+                    if exclusion.subnet_of(network)
+                    else [network]
+                )
+
+            ranges = remaining
+
+        result.extend(ipaddress.collapse_addresses(ranges))
+
+    return result
+
+
+def _linuxRuleDelete(flag, rule, table, start, device):
+    """Match a pinned native rule's selectors, never delete a priority wildcard."""
+    allowed = {
+        'priority',
+        'src',
+        'dst',
+        'table',
+        'not',
+        'iif',
+        'oif',
+        'goto',
+        'suppress_prefixlength',
+        'dport',
+        'action',
+        'protocol',
+    }
+    priority = int(rule['priority'])
+    target = str(rule.get('table', ''))
+
+    if set(rule) - allowed or target not in ('', str(table), 'main', '254'):
+        raise RuntimeError('Conflicting Linux policy rule; refusing sing-tun recovery')
+
+    if rule.get('iif') not in (None, 'lo', device) or rule.get('oif') is not None:
+        raise RuntimeError(
+            'Conflicting Linux interface rule; refusing sing-tun recovery'
+        )
+
+    if 'goto' in rule and rule['goto'] != start + 10:
+        raise RuntimeError('Conflicting Linux goto rule; refusing sing-tun recovery')
+
+    if priority == start + 10 and rule.get('action') != 'nop':
+        raise RuntimeError(
+            'Conflicting Linux terminal rule; refusing sing-tun recovery'
+        )
+
+    if rule.get('protocol') not in (None, 0, 'unspec'):
+        raise RuntimeError(
+            'Conflicting Linux rule protocol; refusing sing-tun recovery'
+        )
+
+    args = ['ip', flag, 'rule', 'del', 'priority', str(priority)]
+
+    if rule.get('not'):
+        args.append('not')
+
+    for key, parameter in (('src', 'from'), ('dst', 'to'), ('iif', 'iif')):
+        if key in rule:
+            value = str(rule[key])
+
+            if key != 'iif' and value != 'all':
+                ipaddress.ip_network(value, strict=False)
+
+            args.extend([parameter, value])
+
+    if 'suppress_prefixlength' in rule:
+        if rule['suppress_prefixlength'] != 0:
+            raise RuntimeError('Conflicting Linux suppression rule')
+
+        args.extend(['suppress_prefixlength', '0'])
+
+    if 'dport' in rule:
+        if str(rule['dport']) not in ('53', '53-53'):
+            raise RuntimeError('Conflicting Linux DNS rule')
+
+        args.extend(['dport', '53'])
+
+    if target:
+        args.extend(['table', target])
+
+    if 'goto' in rule:
+        args.extend(['goto', str(rule['goto'])])
+    elif rule.get('action') == 'nop':
+        args.append('nop')
+    elif not target:
+        raise RuntimeError('Unrecognized Linux rule; cleanup ownership retained')
+
+    return args
+
+
+class SingTUNHostPlan:
+    """Owned by one SingTUN runtime, including any unfinished host worker."""
+
+    def __init__(self, configuration, *, platform=PLATFORM):
+        copied = copy.deepcopy(configuration)
+        proxy = copied.pop('proxy', None)
+
+        self.configuration = prepareSingTUNSettings(copied, platform=platform)
+        self.hostSettings = self.configuration.pop('host_options')
+        self.hostSettingsCallers = SingTUNHostSettingsCallers(self.hostSettings)
+
+        if proxy is not None:
+            self.configuration['proxy'] = proxy
+
+        self.platform = platform
+        self.deviceName = ''
+        self._worker = None
+        self._done = threading.Event()
+        self._cancel = threading.Event()
+        self.error = ''
+        self._dnsRestore = []
+        self._table = None
+        self._ruleStart = None
+        self._activated = False
+        self._ranges = []
+        self._cleanupComplete = False
+        self.egressIP = ''
+
+    def begin(self, method, *args):
+        """Schedule bounded host work; the Qt owner polls completion."""
+        if self._worker is not None and self._worker.is_alive():
+            raise RuntimeError('sing-tun host worker is already active')
+
+        self._done.clear()
+        self.error = ''
+
+        def run():
+            try:
+                if not self._cancel.is_set():
+                    getattr(self, method)(*args)
+            except Exception as ex:
+                self.error = str(ex)
+            finally:
+                self._done.set()
+
+        self._worker = threading.Thread(target=run, name='sing-tun-host')
+        self._worker.start()
+
+    @property
+    def workDone(self):
+        return self._done.is_set()
+
+    def workCompleted(self):
+        return self.workDone
+
+    def drain(self):
+        """Retain a worker on timeout; no late DNS mutation can escape cleanup."""
+        self._cancel.set()
+
+        if self._worker is not None:
+            self._worker.join(6)
+
+            if self._worker.is_alive():
+                raise RuntimeError('sing-tun host preparation is still stopping')
+
+            self._worker = None
+
+    def prepare(self, addresses):
+        if SystemRuntime.flatpakID():
+            raise RuntimeError('sing-tun application TUN is unavailable in Flatpak')
+
+        if self.platform == 'Linux':
+            # The legacy pkexec script does not elevate this spawned Go child.
+            if os.geteuid() != 0:
+                raise RuntimeError(
+                    'sing-tun auto-routing requires root on Linux; '
+                    'select tun2socks to use the existing privileged helper'
+                )
+
+            if not shutil.which('resolvectl'):
+                raise RuntimeError(
+                    'sing-tun DNS management requires systemd-resolved on Linux; select tun2socks on other resolver configurations'
+                )
+
+            try:
+                _command(['resolvectl', 'status'])
+            except RuntimeError:
+                raise RuntimeError(
+                    'sing-tun requires an active systemd-resolved service on Linux'
+                ) from None
+        elif not SystemRuntime.isAdmin():
+            raise RuntimeError('sing-tun requires networking administrator privileges')
+
+        self.configuration = addSingTUNExclusions(self.configuration, addresses)
+        self.captureEgress()
+
+        tun = self.configuration['tun_options']
+        names = self._interfaceNames()
+        name = tun['Name']
+
+        if name and name in names:
+            raise RuntimeError('The requested sing-tun interface already exists')
+
+        if not name:
+            candidates = ['utun' + str(number) for number in range(10, 1025)]
+            candidates = [value for value in candidates if value not in names]
+
+            if not candidates:
+                raise RuntimeError('No unused sing-tun interface name is available')
+
+            name = secrets.choice(candidates)
+
+        if self.platform == 'Linux' and len(name) > 15:
+            raise ValueError('Linux TUN names must be at most 15 characters')
+
+        if self.platform == 'Darwin' and (
+            not name.startswith('utun') or not name[4:].isdigit()
+        ):
+            raise ValueError('macOS requires a utun device name followed by a number')
+
+        tun['Name'] = self.deviceName = name
+
+        if self.platform == 'Linux':
+            rules = sum(
+                (
+                    json.loads(_command(['ip', flag, '-j', 'rule', 'show']))
+                    for flag in ('-4', '-6')
+                ),
+                [],
+            )
+            routes = sum(
+                (
+                    json.loads(
+                        _command(['ip', flag, '-j', 'route', 'show', 'table', 'all'])
+                    )
+                    for flag in ('-4', '-6')
+                ),
+                [],
+            )
+            tables = {str(item.get('table')) for item in rules + routes}
+            priorities = {int(item['priority']) for item in rules}
+
+            for _ in range(100):
+                table = secrets.randbelow(1000000000) + 100000
+                start = secrets.randbelow(1800) * 11 + 1000
+
+                if str(table) not in tables and not any(
+                    start <= value <= start + 10 for value in priorities
+                ):
+                    self._table, self._ruleStart = table, start
+
+                    break
+            else:
+                raise RuntimeError(
+                    'No isolated sing-tun routing identifiers are available'
+                )
+
+            tun['IPRoute2TableIndex'], tun['IPRoute2RuleIndex'] = (
+                self._table,
+                self._ruleStart,
+            )
+        elif self.platform == 'Darwin':
+            self._ranges = _routeRanges(tun)
+
+            # Upstream replaces EEXIST destinations. Refuse a conflict before
+            # activation rather than replacing another owner's exact route.
+            existing = self._darwinRouteSnapshot()
+
+            if any(network in {row[0] for row in existing} for network in self._ranges):
+                raise RuntimeError(
+                    'sing-tun route ranges conflict with existing macOS routes'
+                )
+
+    def markActivated(self):
+        # Install before process.start: partial native initialization also needs recovery.
+        self._activated = True
+
+    def applyDNS(self, actualName):
+        if actualName != self.deviceName:
+            raise RuntimeError('sing-tun reported an unexpected interface name')
+
+        tun = self.configuration['tun_options']
+        dns = self.hostSettingsCallers.userTunAdapterInterfaceDNS() or (
+            '1.1.1.1' if tun['Inet4Address'] else '2606:4700:4700::1111'
+        )
+        address = ipaddress.ip_address(dns)
+
+        if not tun['Inet' + str(address.version) + 'Address']:
+            raise ValueError(
+                'The configured TUN DNS address requires that address family on the interface'
+            )
+
+        if self._cancel.is_set():
+            return
+
+        if self.platform == 'Windows':
+            primaryName = self.hostSettingsCallers.userPrimaryAdapterInterfaceName()
+            primaryIP = self.hostSettingsCallers.userPrimaryAdapterInterfaceIP()
+            snapshot = json.loads(
+                _powershell(
+                    '$ErrorActionPreference="Stop"; '
+                    '$items=@(Get-WmiObject Win32_NetworkAdapterConfiguration | Where-Object {$_.IPEnabled}); '
+                    '@($items | ForEach-Object { $c=$_; '
+                    '$a=Get-WmiObject Win32_NetworkAdapter -Filter ("Index="+$c.Index); '
+                    '$p=Get-ItemProperty ("HKLM:\\SYSTEM\\CurrentControlSet\\Services\\Tcpip\\Parameters\\Interfaces\\"+$c.SettingID); '
+                    'New-Object PSObject -Property @{Name=$a.NetConnectionID; IP=$c.IPAddress; '
+                    'Static=$p.NameServer; DNS=$c.DNSServerSearchOrder}}) | ConvertTo-Json -Depth 4'
+                )
+            )
+
+            if isinstance(snapshot, dict):
+                snapshot = [snapshot]
+
+            physical = [
+                item
+                for item in snapshot
+                if item['Name'] != actualName
+                and (
+                    (primaryName and item['Name'] == primaryName)
+                    or (primaryIP and primaryIP in (item['IP'] or []))
+                )
+            ]
+
+            if not primaryName and not primaryIP:
+                # The physical default route was captured before native startup.
+                physical = [
+                    item for item in snapshot if self.egressIP in (item['IP'] or [])
+                ]
+
+            if self.hostSettingsCallers.userDisablePrimaryAdapterInterfaceDNS():
+                if len(physical) != 1:
+                    raise RuntimeError(
+                        'Unable to identify the physical adapter for DNS restoration'
+                    )
+
+                item = physical[0]
+
+                if not isinstance(item.get('Name'), str) or not item['Name']:
+                    raise RuntimeError(
+                        'Unable to snapshot the physical adapter DNS identity'
+                    )
+
+                item['DNS'] = [
+                    server
+                    for server in item.get('DNS') or []
+                    if ipaddress.ip_address(server).version == 4
+                ]
+
+                # Register restoration before the first host write.
+                self._dnsRestore.append(('Windows', item))
+                self._windowsDNS(item['Name'], ['127.0.0.1'])
+
+            self._windowsDNS(actualName, [str(address)])
+
+            _command(['ipconfig', '/flushdns'])
+        elif self.platform == 'Darwin':
+            listing = _command(
+                ['networksetup', '-listallnetworkservices']
+            ).splitlines()[1:]
+
+            for service in listing:
+                if not service or service.startswith('*') or self._cancel.is_set():
+                    continue
+
+                original = _command(
+                    ['networksetup', '-getdnsservers', service]
+                ).splitlines()
+
+                servers = []
+
+                for value in original:
+                    try:
+                        servers.append(str(ipaddress.ip_address(value)))
+                    except ValueError:
+                        if "aren't any DNS Servers" not in value:
+                            raise RuntimeError(
+                                'Cannot snapshot macOS DNS configuration'
+                            ) from None
+
+                self._dnsRestore.append(('Darwin', (service, servers)))
+
+                _command(['networksetup', '-setdnsservers', service, str(address)])
+        elif self.platform == 'Linux':
+            # Only the new, ephemeral interface is touched; never /etc/resolv.conf.
+            _command(['resolvectl', 'dns', actualName, str(address)])
+            _command(['resolvectl', 'domain', actualName, '~.'])
+
+    def captureEgress(self):
+        if self.platform == 'Windows':
+            gateways = SystemRoutingTable.DEFAULT_GATEWAY_WIN32.findall(
+                _command(['route', 'print', '0.0.0.0'])
+            )
+
+            if len(gateways) != 1:
+                raise RuntimeError('Unable to identify the physical default gateway')
+
+            self.egressIP = (
+                self.hostSettingsCallers.userPrimaryAdapterInterfaceIP()
+                or gateways[0][1]
+            )
+
+    @staticmethod
+    def _windowsDNS(name, servers):
+        family = (
+            'ipv4'
+            if not servers or ipaddress.ip_address(servers[0]).version == 4
+            else 'ipv6'
+        )
+
+        if not servers:
+            _command(
+                [
+                    'netsh',
+                    'interface',
+                    family,
+                    'set',
+                    'dnsservers',
+                    'name=' + name,
+                    'source=dhcp',
+                ]
+            )
+
+            return
+
+        _command(
+            [
+                'netsh',
+                'interface',
+                family,
+                'set',
+                'dnsservers',
+                'name=' + name,
+                'source=static',
+                'address=' + servers[0],
+                'validate=no',
+            ]
+        )
+
+        for index, server in enumerate(servers[1:], 2):
+            _command(
+                [
+                    'netsh',
+                    'interface',
+                    family,
+                    'add',
+                    'dnsservers',
+                    'name=' + name,
+                    'address=' + server,
+                    'index=' + str(index),
+                    'validate=no',
+                ]
+            )
+
+    def _interfaceNames(self):
+        if self.platform == 'Windows':
+            names = json.loads(
+                _powershell(
+                    '@([System.Net.NetworkInformation.NetworkInterface]::GetAllNetworkInterfaces() | ForEach-Object {$_.Name}) | ConvertTo-Json'
+                )
+            )
+
+            return set(names if isinstance(names, list) else [names] if names else [])
+
+        return {name for _, name in socket.if_nameindex()}
+
+    def _darwinRouteSnapshot(self):
+        result = []
+
+        for family, name in ((4, 'inet'), (6, 'inet6')):
+            result.extend(
+                _darwinRoutes(_command(['netstat', '-rn', '-f', name]), family)
+            )
+
+        return result
+
+    def restoreDNS(self):
+        """Restore only recorded host DNS, independently of native release."""
+        errors = []
+
+        for platform, snapshot in reversed(self._dnsRestore[:]):
+            try:
+                if platform == 'Windows':
+                    self._windowsDNS(
+                        snapshot['Name'], snapshot['DNS'] if snapshot['Static'] else []
+                    )
+                else:
+                    service, servers = snapshot
+
+                    _command(
+                        [
+                            'networksetup',
+                            '-setdnsservers',
+                            service,
+                            *(servers or ['Empty']),
+                        ]
+                    )
+
+                self._dnsRestore.remove((platform, snapshot))
+            except Exception as ex:
+                errors.append(str(ex))
+
+        if errors:
+            raise RuntimeError('; '.join(errors))
+
+    def cleanup(self, cleanExit):
+        """Restore independent DNS even if exact native recovery refuses cleanup."""
+        errors = []
+
+        try:
+            self.restoreDNS()
+        except Exception as ex:
+            errors.append(str(ex))
+
+        if self._activated:
+            try:
+                self._recoverNative()
+            except Exception as ex:
+                errors.append(str(ex))
+
+        if errors:
+            raise RuntimeError('; '.join(errors))
+
+        self._activated = False
+        self._cleanupComplete = True
+
+    def finishCleanup(self, cleanExit):
+        """Run host restoration outside Qt; retain the worker on a bounded drain."""
+        if self._cleanupComplete:
+            return
+
+        self._finishCleanupWork('cleanup', cleanExit)
+
+    def finishDNSCleanup(self):
+        """Restore DNS even when the child cannot be reaped; retain native ownership."""
+        if self._dnsRestore:
+            self._finishCleanupWork('restoreDNS')
+
+    def _finishCleanupWork(self, method, *args):
+        if self._worker is None:
+            # drain() cancelled preparation; restoration must still execute.
+            self._cancel.clear()
+            self.begin(method, *args)
+
+        self._worker.join(6)
+
+        if self._worker.is_alive():
+            raise RuntimeError('sing-tun host restoration is still running')
+
+        self._worker = None
+
+        if self.error:
+            raise RuntimeError(self.error)
+
+    def _recoverNative(self):
+        names = self._interfaceNames()
+
+        if self.deviceName in names:
+            # Closing the exact child should remove its ephemeral TUN. Do not
+            # destroy a same-named interface whose identity we cannot prove.
+            raise RuntimeError(
+                'sing-tun interface remains after child exit; cleanup ownership retained'
+            )
+
+        if self.platform == 'Linux' and self._table is not None:
+            for flag in ('-4', '-6'):
+                rules = json.loads(_command(['ip', flag, '-j', 'rule', 'show']))
+
+                for rule in rules:
+                    priority = int(rule['priority'])
+
+                    if not self._ruleStart <= priority <= self._ruleStart + 10:
+                        continue
+
+                    _command(
+                        _linuxRuleDelete(
+                            flag, rule, self._table, self._ruleStart, self.deviceName
+                        )
+                    )
+
+                remaining = json.loads(_command(['ip', flag, '-j', 'rule', 'show']))
+
+                if any(
+                    self._ruleStart <= int(rule['priority']) <= self._ruleStart + 10
+                    for rule in remaining
+                ):
+                    raise RuntimeError(
+                        'sing-tun Linux rules remain; cleanup ownership retained'
+                    )
+
+                routes = json.loads(
+                    _command(
+                        ['ip', flag, '-j', 'route', 'show', 'table', str(self._table)]
+                    )
+                )
+
+                if routes:
+                    # Routes referring to the now-closed ephemeral interface
+                    # should disappear in the kernel; never flush an unrelated table.
+                    raise RuntimeError(
+                        'sing-tun routing table remains; cleanup ownership retained'
+                    )
+        elif self.platform == 'Darwin':
+            # Routes may survive a partial native start before routeSet is set.
+            # Delete only an exact planned destination AND its attempt gateway.
+            table = self._darwinRouteSnapshot()
+
+            for network in self._ranges:
+                family = '4' if network.version == 4 else '6'
+                gateway = str(
+                    ipaddress.ip_interface(
+                        self.configuration['tun_options']['Inet' + family + 'Address'][
+                            0
+                        ]
+                    ).ip
+                )
+
+                if any(
+                    row[0] == network
+                    and row[1].split('%', 1)[0] == gateway
+                    and row[2] == self.deviceName
+                    for row in table
+                ):
+                    _command(
+                        [
+                            'route',
+                            '-n',
+                            'delete',
+                            '-inet' + ('6' if family == '6' else ''),
+                            str(network),
+                            gateway,
+                        ]
+                    )
+
+            remaining = self._darwinRouteSnapshot()
+
+            if any(
+                row[0] in self._ranges and row[2] == self.deviceName
+                for row in remaining
+            ):
+                raise RuntimeError(
+                    'sing-tun macOS routes remain; cleanup ownership retained'
+                )
@@ -24,7 +24,7 @@ from Furious.Application.DesktopApplication import (
    _ExistingInstanceResult,
    _SingletonStartupResult,
 )
-from Furious.Frozenlib import AppBuiltinCommand, PySide6Legacy
+from Furious.Frozenlib import AppBuiltinCommand, PySide6Legacy, AppSettings
 from Furious.Interface import ApplicationRunner, CoreRuntime
 from Furious.Qt import AppQMessageBox
 from Furious.Qt.AppStyleSheet import AppStyleSheet
@@ -34,6 +34,8 @@ from Furious.Service.RuntimeLease import RuntimeEventRouter, RuntimeLease
 from PySide6 import QtCore
 from PySide6.QtNetwork import QLocalServer

+from tests.support import isolatedSettings
+
 from types import SimpleNamespace
 from unittest import TestCase, mock

@@ -1551,6 +1553,15 @@ class ApplicationLifecycleTransactionTest(TestCase):
 class ConnectionStartupTransactionTest(TestCase):
    """Verify one failed attempt releases only its own exact runtimes."""

+    def setUp(self):
+        # These fixtures explicitly replace the legacy tun2socks host stage.
+        self._settings = isolatedSettings()
+        self._settings.__enter__()
+        AppSettings.set('ApplicationTUNBackend', 'tun2socks')
+
+    def tearDown(self):
+        self._settings.__exit__(None, None, None)
+
    @staticmethod
    def _commitExistingRuntime(manager, runtime):
        """Give one fixture runtime to the manager through the real lease boundary."""
@@ -22,21 +22,19 @@ from __future__ import annotations
 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,
-)
+from Furious.Service.ConnectionManager import ConnectionManager, ConnectionStartStage
 from Furious.Service.DnsResolver import DnsResolutionOperation, DnsResolver
+from Furious.Frozenlib import AppSettings

 from PySide6 import QtCore, QtNetwork

 from tests.support import application, processQtEvents, waitFor

+from unittest import TestCase, mock
+
 import importlib
 import unittest

-from unittest import TestCase, mock
-

 class _Configuration:
    """Provide the endpoint values used by manager-only tests."""
@@ -205,6 +203,8 @@ class ConnectionStartupAsyncTest(TestCase):
    def setUp(self):
        """Ensure a Qt application exists for real timer/socket delivery."""
        self.app = application()
+        self._previousTUNBackend = AppSettings.get('ApplicationTUNBackend')
+        AppSettings.set('ApplicationTUNBackend', 'tun2socks')
        self.servers = []
        self.managers = []

@@ -219,6 +219,8 @@ class ConnectionStartupAsyncTest(TestCase):

        processQtEvents()

+        AppSettings.set('ApplicationTUNBackend', self._previousTUNBackend)
+
    def _manager(self):
        """Return a manager with host TUN disabled at the test boundary."""
        manager = ConnectionManager()
@@ -20,6 +20,7 @@
 from __future__ import annotations

 from Furious.Backends.ExternalCore import ConfigExternalCore, ExternalCoreProcess
+from Furious.Frozenlib import AppSettings
 from Furious.Backends.ExternalCore.Plugin import (
    ExternalCorePlugin,
    ExternalCoreRuntimeFactory,
@@ -41,7 +42,7 @@ from Furious.Service.SubscriptionImporter import (

 from PySide6 import QtCore

-from tests.support import application, waitFor as waitForQt
+from tests.support import application, isolatedSettings, waitFor as waitForQt

 import os
 import sys
@@ -522,7 +523,10 @@ class ExternalCoreProcessTest(unittest.TestCase):
                """Return one successful process-free fixture launch."""
                return None, True

-        with tempfile.TemporaryDirectory(dir=Path.cwd()) as directory:
+        with tempfile.TemporaryDirectory(
+            dir=Path.cwd()
+        ) as directory, isolatedSettings():
+            AppSettings.set('ApplicationTUNBackend', 'tun2socks')
            config = self.configuration([], directory)
            config['useApplicationTun2socks'] = True
            config['tunRemoteAddress'] = 'actual-server.example.com'
@@ -551,28 +555,11 @@ class ExternalCoreProcessTest(unittest.TestCase):
                    return_value=registry,
                ),
                mock.patch(
-                    'Furious.Service.ConnectionManager.userDefaultPrimaryGatewayIP',
-                    return_value='192.168.50.1',
-                ),
-                mock.patch(
-                    'Furious.Service.ConnectionManager.userPrimaryAdapterInterfaceIP',
-                    return_value='192.168.50.20',
-                ),
-                mock.patch(
-                    'Furious.Service.ConnectionManager.userTcpSendBufferSize',
-                    return_value=1,
-                ),
-                mock.patch(
-                    'Furious.Service.ConnectionManager.userTcpReceiveBufferSize',
-                    return_value=1,
-                ),
-                mock.patch(
-                    'Furious.Service.ConnectionManager.userTcpAutoTuning',
-                    return_value='False',
-                ),
-                mock.patch(
-                    'Furious.Service.ConnectionManager.userBypassTUNAdapterInterfaceIP',
-                    return_value='',
+                    'Furious.Service.ConnectionManager.Storage.UserTUNSettings',
+                    return_value={
+                        'defaultPrimaryGatewayIP': '192.168.50.1',
+                        'primaryAdapterInterfaceIP': '192.168.50.20',
+                    },
                ),
                mock.patch(
                    'Furious.Service.ConnectionManager.SystemRoutingTable.delete'
@@ -178,11 +178,13 @@ class XrayNativeTUNTest(unittest.TestCase):
                return_value=registry,
            ),
            mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks,
+            mock.patch('Furious.Service.ConnectionManager.SingTUN') as singTUN,
        ):
            self.assertTrue(manager.start(original, 'Global'))

        registry.usesApplicationTun2socks.assert_not_called()
        tun2socks.assert_not_called()
+        singTUN.assert_not_called()

        self.assertEqual(manager.runtimeConfiguration, original)

@@ -205,6 +207,7 @@ class XrayNativeTUNTest(unittest.TestCase):
                return_value=registry,
            ),
            mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks,
+            mock.patch('Furious.Service.ConnectionManager.SingTUN') as singTUN,
        ):
            self.assertFalse(manager.start(original, 'Global'))

@@ -212,6 +215,7 @@ class XrayNativeTUNTest(unittest.TestCase):

        registry.usesApplicationTun2socks.assert_not_called()
        tun2socks.assert_not_called()
+        singTUN.assert_not_called()

        self.assertFalse(hasattr(manager, 'runtimeConfiguration'))

@@ -407,11 +411,13 @@ class Hysteria2NativeTUNTest(unittest.TestCase):
                return_value=registry,
            ),
            mock.patch('Furious.Service.ConnectionManager.Tun2socks') as tun2socks,
+            mock.patch('Furious.Service.ConnectionManager.SingTUN') as singTUN,
        ):
            self.assertTrue(manager.start(original, 'Global'))

        registry.usesApplicationTun2socks.assert_not_called()
        tun2socks.assert_not_called()
+        singTUN.assert_not_called()

        self.assertEqual(manager.runtimeConfiguration, original)

@@ -58,6 +58,7 @@ class RepositoryContractTest(unittest.TestCase):
            '_UserServersStorage',
            '_UserSubsStorage',
            '_UserTUNSettingsStorage',
+            '_UserSingTUNSettingsStorage',
            '_UserRoutingsStorage',
        ):
            getattr(Storage, name).cache_clear()