From 251ec3cec6ab782ca0b7397acf8a56a1fd5a0d11 Mon Sep 17 00:00:00 2001 From: Loren Eteval Date: Tue, 6 Oct 2026 00:37:54 +0800 Subject: [PATCH] Add sing-tun settings UI Signed-off-by: Loren Eteval --- Furious/Backends/ExternalCore/Editor.py | 4 +- Furious/Externals/GenTranslation.py | 317 +++- Furious/Window/HomePage.py | 5 + Furious/Window/MainWindow.py | 1 + Furious/Window/SettingsPage.py | 69 +- Furious/Window/SingTUNSettingsDialog.py | 424 ++++++ Furious/Window/TunSettingsDialog.py | 4 +- tests/README.md | 3 +- tests/fixtures/sing_tun_lifetime_probe.py | 116 ++ tests/test_sing_tun.py | 1603 +++++++++++++++++++++ tests/test_ui_behavior.py | 28 + 11 files changed, 2534 insertions(+), 40 deletions(-) create mode 100644 Furious/Window/SingTUNSettingsDialog.py create mode 100644 tests/fixtures/sing_tun_lifetime_probe.py create mode 100644 tests/test_sing_tun.py diff --git a/Furious/Backends/ExternalCore/Editor.py b/Furious/Backends/ExternalCore/Editor.py index f40a1a64..8e6f1e0e 100644 --- a/Furious/Backends/ExternalCore/Editor.py +++ b/Furious/Backends/ExternalCore/Editor.py @@ -309,7 +309,7 @@ class ExternalCoreApplicationTun2socksInput(GuiEditorItemTextSwitch): def __init__(self): """Initialize the application tun2socks opt-in control.""" - super().__init__(title=_('Use Application Tun2socks')) + super().__init__(title=_('Use Application TUN')) def connectToggled(self, receiver, methodName: str): """Connect a field-state callback without retaining its binding.""" @@ -536,7 +536,7 @@ class ExternalCoreEditor(GuiEditorWidgetQDialog): and not self._tunRemoteAddressInput.text() ): validationMessage = _( - 'TUN remote address is required when application tun2socks is enabled' + 'TUN remote address is required when application TUN is enabled' ) if validationMessage: diff --git a/Furious/Externals/GenTranslation.py b/Furious/Externals/GenTranslation.py index 03e30270..67339af6 100644 --- a/Furious/Externals/GenTranslation.py +++ b/Furious/Externals/GenTranslation.py @@ -1425,6 +1425,7 @@ TRANSLATION = { }, "Primary Adapter Interface Name": { "source": [ + "Furious.Window.SingTUNSettingsDialog", "Furious.Window.TunSettingsDialog" ], "RU": "Имя основного сетевого адаптера", @@ -1433,6 +1434,7 @@ TRANSLATION = { }, "Primary Adapter Interface IP": { "source": [ + "Furious.Window.SingTUNSettingsDialog", "Furious.Window.TunSettingsDialog" ], "RU": "IP-адрес основного сетевого адаптера", @@ -1447,24 +1449,9 @@ TRANSLATION = { "ZH": "默认主网关IP", "isReviewed": "True" }, - "Tun2socks Adapter Interface DNS": { - "source": [ - "Furious.Window.TunSettingsDialog" - ], - "RU": "DNS-адрес интерфейса адаптера Tun2socks", - "ZH": "Tun2socks适配器接口DNS", - "isReviewed": "True" - }, - "Bypass Tun2socks Adapter Interface IP (separated by commas)": { - "source": [ - "Furious.Window.TunSettingsDialog" - ], - "RU": "Обход IP-адреса интерфейса адаптера Tun2socks (через запятую)", - "ZH": "绕过Tun2socks适配器接口IP（以逗号分隔）", - "isReviewed": "True" - }, "Disable Primary Adapter Interface DNS (Mitigating DNS leaks on Windows)": { "source": [ + "Furious.Window.SingTUNSettingsDialog", "Furious.Window.TunSettingsDialog" ], "RU": "Отключить DNS основного сетевого адаптера (для предотвращения утечек DNS в Windows)", @@ -2760,14 +2747,6 @@ TRANSLATION = { "ZH": "SOCKS代理", "isReviewed": "True" }, - "Use Application Tun2socks": { - "source": [ - "Furious.Backends.ExternalCore.Editor" - ], - "RU": "Использовать Tun2socks приложения", - "ZH": "使用应用程序的Tun2socks", - "isReviewed": "True" - }, "TUN Remote Address": { "source": [ "Furious.Backends.ExternalCore.Editor" @@ -2936,14 +2915,6 @@ TRANSLATION = { "ZH": "代理出口信息与隐私", "isReviewed": "True" }, - "TUN remote address is required when application tun2socks is enabled": { - "source": [ - "Furious.Backends.ExternalCore.Editor" - ], - "RU": "При включённом Tun2socks приложения необходимо указать удалённый адрес TUN", - "ZH": "启用应用程序的Tun2socks时必须填写TUN远程地址", - "isReviewed": "True" - }, "Data usage": { "source": [ "Furious.Window.SettingsPage" @@ -3375,5 +3346,287 @@ TRANSLATION = { "RU": "Тесты", "ZH": "测试", "isReviewed": "True" + }, + "Application TUN Engine": { + "source": [ + "Furious.Window.SettingsPage" + ], + "RU": "Движок TUN приложения", + "ZH": "应用 TUN 引擎", + "isReviewed": "True" + }, + "Native proxy-core TUN takes priority. This engine is used when Furious provides application TUN. Changes apply to the next connection.": { + "source": [ + "Furious.Window.SettingsPage" + ], + "RU": "Встроенный TUN прокси-ядра имеет приоритет. Этот движок используется, когда TUN предоставляет Furious. Изменения применяются при следующем подключении.", + "ZH": "代理内核的原生 TUN 优先。由 Furious 提供应用 TUN 时才使用此引擎。更改将在下次连接时生效。", + "isReviewed": "True" + }, + "Customize sing-tun Settings...": { + "source": [ + "Furious.Window.SettingsPage" + ], + "RU": "Настроить sing-tun...", + "ZH": "自定义 sing-tun 设置...", + "isReviewed": "True" + }, + "Customize sing-tun Settings": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Настройка sing-tun", + "ZH": "自定义 sing-tun 设置", + "isReviewed": "True" + }, + "SOCKS Transit": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Передача через SOCKS", + "ZH": "SOCKS 转发", + "isReviewed": "True" + }, + "Stack": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Сетевой стек", + "ZH": "网络协议栈", + "isReviewed": "True" + }, + "Log Level": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Уровень журналирования", + "ZH": "日志级别", + "isReviewed": "True" + }, + "SOCKS Interface Binding (empty for automatic)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Интерфейс SOCKS (пусто — автоматически)", + "ZH": "SOCKS 绑定接口（留空自动选择）", + "isReviewed": "True" + }, + "Maximum Sessions": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Максимум сеансов", + "ZH": "最大会话数", + "isReviewed": "True" + }, + "Interface and Stack": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Интерфейс и сетевой стек", + "ZH": "接口与协议栈", + "isReviewed": "True" + }, + "Device Name (empty for automatic)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Имя устройства (пусто — автоматически)", + "ZH": "设备名称（留空自动选择）", + "isReviewed": "True" + }, + "MTU (bytes)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "MTU (байты)", + "ZH": "MTU（字节）", + "isReviewed": "True" + }, + "IPv4 Address Prefixes (comma separated)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Префиксы IPv4 (через запятую)", + "ZH": "IPv4 地址前缀（以逗号分隔）", + "isReviewed": "True" + }, + "IPv6 Address Prefixes (comma separated)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Префиксы IPv6 (через запятую)", + "ZH": "IPv6 地址前缀（以逗号分隔）", + "isReviewed": "True" + }, + "Native UDP Timeout (Go duration, e.g. 1m)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Тайм-аут UDP стека (длительность Go, например 1m)", + "ZH": "原生 UDP 超时（Go 时长，例如 1m）", + "isReviewed": "True" + }, + "Native ICMP Timeout (Go duration, 0 for default)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Тайм-аут ICMP стека (длительность Go, 0 — по умолчанию)", + "ZH": "原生 ICMP 超时（Go 时长，0 使用默认值）", + "isReviewed": "True" + }, + "Native UDP NAT Limit (0 for default)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Лимит UDP NAT стека (0 — по умолчанию)", + "ZH": "原生 UDP NAT 上限（0 使用默认值）", + "isReviewed": "True" + }, + "Advanced Options": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Дополнительные параметры", + "ZH": "高级选项", + "isReviewed": "True" + }, + "Use exact native JSON field names. Supported: route address/exclusion lists, StrictRoute, InterfaceScope, ForwarderBindInterface and IncludeAllNetworks. AutoRoute, external configuration and DNS ownership are managed. Unsupported fields are rejected.": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Используйте точные имена полей JSON. Поддерживаются списки маршрутов и исключений, StrictRoute, InterfaceScope, ForwarderBindInterface и IncludeAllNetworks. AutoRoute, внешняя настройка и управление DNS контролируются приложением. Неподдерживаемые поля отклоняются.", + "ZH": "请使用准确的原生 JSON 字段名。支持路由地址及排除列表、StrictRoute、InterfaceScope、ForwarderBindInterface 和 IncludeAllNetworks。AutoRoute、外部配置模式和 DNS 归属由应用管理。不支持的字段会被拒绝。", + "isReviewed": "True" + }, + "Connect Timeout (seconds)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Тайм-аут подключения (секунды)", + "ZH": "连接超时（秒）", + "isReviewed": "True" + }, + "SOCKS UDP Timeout (seconds)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Тайм-аут UDP SOCKS (секунды)", + "ZH": "SOCKS UDP 超时（秒）", + "isReviewed": "True" + }, + "TCP Idle Timeout (seconds)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Тайм-аут простоя TCP (секунды)", + "ZH": "TCP 空闲超时（秒）", + "isReviewed": "True" + }, + "UDP Mapping": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Сопоставление UDP", + "ZH": "UDP 映射方式", + "isReviewed": "True" + }, + "UDP Filtering": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Фильтрация UDP", + "ZH": "UDP 过滤方式", + "isReviewed": "True" + }, + "Automatic": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Автоматический выбор", + "ZH": "自动选择", + "isReviewed": "True" + }, + "Endpoint Independent": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Независимо от конечной точки", + "ZH": "与端点无关", + "isReviewed": "True" + }, + "Address Dependent": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Зависит от адреса", + "ZH": "依赖地址", + "isReviewed": "True" + }, + "Address and Port Dependent": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Зависит от адреса и порта", + "ZH": "依赖地址和端口", + "isReviewed": "True" + }, + "TUN remote address is required when application TUN is enabled": { + "source": [ + "Furious.Backends.ExternalCore.Editor" + ], + "RU": "При включённом TUN приложения требуется адрес удалённого сервера", + "ZH": "启用应用程序 TUN 时必须填写 TUN 远程地址", + "isReviewed": "True" + }, + "Use Application TUN": { + "source": [ + "Furious.Backends.ExternalCore.Editor" + ], + "RU": "Использовать TUN приложения", + "ZH": "使用应用程序 TUN", + "isReviewed": "True" + }, + "Configure sing-tun interface, stack, SOCKS transit and host settings.": { + "source": [ + "Furious.Window.SettingsPage" + ], + "RU": "Настройте интерфейс, стек, SOCKS-передачу и параметры хоста sing-tun.", + "ZH": "配置 sing-tun 的接口、协议栈、SOCKS 转发和主机设置。", + "isReviewed": "True" + }, + "Host Settings": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Настройки хоста", + "ZH": "主机设置", + "isReviewed": "True" + }, + "TCP Congestion Control (Go stack)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog" + ], + "RU": "Управление перегрузкой TCP (стек Go)", + "ZH": "TCP 拥塞控制（Go 协议栈）", + "isReviewed": "True" + }, + "TUN Adapter Interface DNS": { + "source": [ + "Furious.Window.SingTUNSettingsDialog", + "Furious.Window.TunSettingsDialog" + ], + "RU": "DNS-адрес интерфейса адаптера TUN", + "ZH": "TUN适配器接口DNS", + "isReviewed": "True" + }, + "Bypass TUN Adapter Interface IP (separated by commas)": { + "source": [ + "Furious.Window.SingTUNSettingsDialog", + "Furious.Window.TunSettingsDialog" + ], + "RU": "IP-адреса в обход интерфейса TUN (через запятую)", + "ZH": "绕过TUN适配器接口的IP地址（以逗号分隔）", + "isReviewed": "True" } } diff --git a/Furious/Window/HomePage.py b/Furious/Window/HomePage.py index 16168cdc..11f9d69a 100644 --- a/Furious/Window/HomePage.py +++ b/Furious/Window/HomePage.py @@ -49,6 +49,7 @@ from Furious.Window.ProxyBypassDialog import * from Furious.Window.QRCodeWindow import QRCodeWindow from Furious.Window.TextEditorWindow import TextEditorWindow from Furious.Window.TunSettingsDialog import * +from Furious.Window.SingTUNSettingsDialog import SingTUNSettingsDialog from PySide6 import QtCore from PySide6.QtGui import * @@ -1130,6 +1131,10 @@ class HomePage(Mixins.QTranslatable, QMainWindow): self.updatesManager.configureHttpProxy(Storage.Extras.UserHttpProxy()) self.updatesManager.checkForUpdates(**kwargs) + def getGuiSingTUNSettings(self, **kwargs): + """Transfer a fresh managed transient editor to the Settings page.""" + return SingTUNSettingsDialog(parent=kwargs.pop('parent', self), **kwargs) + def resetNetworkState(self): """Reset network state.""" self.networkState.clearStatus() diff --git a/Furious/Window/MainWindow.py b/Furious/Window/MainWindow.py index a89513cf..6c95e110 100644 --- a/Furious/Window/MainWindow.py +++ b/Furious/Window/MainWindow.py @@ -94,6 +94,7 @@ class MainWindow(AppQMainWindow): self.logPage = AppLogPage() self.settingsPage = SettingsPage( tunSettingsDialogFactory=self.homePage.getGuiTUNSettings, + singTunSettingsDialogFactory=self.homePage.getGuiSingTUNSettings, proxyBypassDialog=self.homePage.customizeProxyBypassDialog, networkTestDialog=self.homePage.customizeNetworkTestDialog, checkForUpdates=self.homePage.checkForUpdates, diff --git a/Furious/Window/SettingsPage.py b/Furious/Window/SettingsPage.py index b29185c4..5efd93df 100644 --- a/Furious/Window/SettingsPage.py +++ b/Furious/Window/SettingsPage.py @@ -39,6 +39,7 @@ from Furious.Service.TrafficStatsManager import ( METRICS_COLLECTION_SETTING, ) from Furious.Window.TunSettingsDialog import tunModeTitle +from Furious.Qt.Signals import connectWeakly from PySide6 import QtCore from PySide6.QtWidgets import * @@ -555,6 +556,44 @@ class _SystemProxySettingsCard(_SettingsCard): AppSettingsController().setSystemProxyMode(mode) +class _TUNBackendSettingsCard(_SettingsCard): + """Select an application fallback without changing native-core policy.""" + + def __init__(self): + self.comboBox = AppQComboBox() + self.comboBox.setObjectName('SettingsComboBox') + self.comboBox.addItem('sing-tun', 'sing-tun') + self.comboBox.addItem('tun2socks', 'tun2socks') + + super().__init__( + 'diagram-3.svg', + self.comboBox, + _('Application TUN Engine'), + _( + 'Native proxy-core TUN takes priority. This engine is used when Furious provides application TUN. Changes apply to the next connection.' + ), + ) + + self.sync() + + connectWeakly(self.comboBox.currentIndexChanged, self, '_selectionChanged') + connectWeakly(AppSettingsController().tunBackendChanged, self, 'sync') + + def sync(self, backend=None): + blocker = QtCore.QSignalBlocker(self.comboBox) + value = ( + backend + if isinstance(backend, str) + else AppSettings.get(APPLICATION_TUN_BACKEND) + ) + + self.comboBox.setCurrentIndex(max(self.comboBox.findData(value), 0)) + del blocker + + def _selectionChanged(self, _index): + AppSettingsController().setTUNBackend(self.comboBox.currentData()) + + class _SettingsSection(QWidget): """Group a translated heading and a stack of settings cards.""" @@ -601,6 +640,7 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): self, *, tunSettingsDialogFactory, + singTunSettingsDialogFactory=None, proxyBypassDialog, networkTestDialog, checkForUpdates, @@ -613,6 +653,7 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): super().__init__(parent) self.setObjectName('SettingsPage') + self._singTunSettingsDialogFactory = singTunSettingsDialogFactory ( self._tunSettingsDialogFactory, @@ -636,6 +677,7 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): self.pageTitleLabel.setObjectName('SettingsPageTitle') self.generalSection = _SettingsSection(_('General')) + self.tunSection = _SettingsSection('TUN', translatable=False) self.connectionSection = _SettingsSection(_('Connection and Interface')) self.applicationSection = _SettingsSection(_('Application')) self.pluginSettingsTitleLabel = AppQLabel(_('Plugin Settings')) @@ -644,6 +686,14 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): self._pluginActions = [] self._tunModeAvailable = AppSettingsController().tunModeAvailable() + self.tunBackendCard = _TUNBackendSettingsCard() + self.singTunSettingsCard = _ActionSettingsCard( + None, + self._openSingTUNSettings, + _('Customize sing-tun Settings...'), + _('Configure sing-tun interface, stack, SOCKS transit and host settings.'), + _('Open'), + ) self.tunModeCard = _ToggleSettingsCard( 'shield-check.svg', @@ -729,7 +779,7 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): self.editorWhitespaceCard, ) = ( _ActionSettingsCard( - 'diagram-3.svg', + None, self._openTUNSettings, _('Customize Tun2socks Settings...'), _('Configure the external Tun2socks network interface and routing.'), @@ -802,8 +852,10 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): self.systemProxyCard.sync ) - if not SystemRuntime.flatpakID(): - self.connectionSection.addCard(self.tunSettingsCard) + self.tunSection.addCard(self.tunBackendCard) + self.tunSection.addCard(self.singTunSettingsCard) + self.tunSection.addCard(self.tunSettingsCard) + self.tunSection.setVisible(not SystemRuntime.flatpakID()) self.connectionSection.addCard(self.systemProxyCard) self.connectionSection.addCard(self.proxyBypassCard) @@ -884,6 +936,7 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): contentLayout.setSpacing(22) contentLayout.addWidget(self.pageTitleLabel) contentLayout.addWidget(self.generalSection) + contentLayout.addWidget(self.tunSection) contentLayout.addWidget(self.connectionSection) if self.pluginSections: @@ -1124,10 +1177,19 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): """Show the end-user disclosure for the active endpoint providers.""" _endpointPrivacyMessageBox(self).open() + def _openSingTUNSettings(self): + if self._singTunSettingsDialogFactory is not None: + self._singTunSettingsDialogFactory(parent=self).open() + def setConnectionControlsEnabled(self, enabled: bool): """Disable connection-sensitive settings during a transition.""" self.tunModeCard.setEnabled(bool(enabled) and self._tunModeAvailable) self.systemProxyCard.comboBox.setEnabled(bool(enabled)) + self.tunBackendCard.setEnabled(bool(enabled) and not SystemRuntime.flatpakID()) + self.tunSettingsCard.setEnabled(bool(enabled)) + self.singTunSettingsCard.setEnabled( + bool(enabled) and self._singTunSettingsDialogFactory is not None + ) def showEvent(self, event): """Synchronize controls in case a legacy action changed a setting.""" @@ -1135,6 +1197,7 @@ class SettingsPage(Mixins.QTranslatable, QMainWindow): for section in ( self.generalSection, + self.tunSection, self.connectionSection, self.applicationSection, *self.pluginSections, diff --git a/Furious/Window/SingTUNSettingsDialog.py b/Furious/Window/SingTUNSettingsDialog.py new file mode 100644 index 00000000..3b9451f1 --- /dev/null +++ b/Furious/Window/SingTUNSettingsDialog.py @@ -0,0 +1,424 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# This file is part of Furious. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""A staged, transient editor for sing-tun's native and host configuration.""" + +from __future__ import annotations + +from Furious.Models.SingTUN import ( + SING_TUN_DEFAULTS, + SING_TUN_HOST_DEFAULTS, + prepareSingTUNSettings, +) +from Furious.Repository import Storage +from Furious.Frozenlib import PLATFORM, GOLDEN_RATIO, AppFontName +from Furious.Qt import ( + AppQTransientDialog, + AppQLabel, + AppQLineEdit, + AppQComboBox, + AppQSpinBox, + AppQSwitch, + AppQTabWidget, + AppQDialogButtonBox, + DraculaJSONTextEditor, +) +from Furious.Qt.Signals import connectWeakly +from Furious.Qt import gettext as _ + +from PySide6 import QtCore +from PySide6.QtWidgets import ( + QFormLayout, + QHBoxLayout, + QVBoxLayout, + QWidget, + QScrollArea, +) + +import copy +import json + + +class SingTUNSettingsDialog(AppQTransientDialog): + """Never import the native engine or mutate storage while loading widgets.""" + + FIXED_DIALOG_SIZE = ( + QtCore.QSize(int(690 * GOLDEN_RATIO), 690) + if PLATFORM in ('Darwin', 'Linux') + else QtCore.QSize(int(620 * GOLDEN_RATIO), 620) + ) + + def __init__(self, parent=None): + super().__init__(parent) + + self.setWindowTitle(_('Customize sing-tun Settings')) + self._original = copy.deepcopy(Storage.UserSingTUNSettings()) + self.fields = {} + + self.tabs = AppQTabWidget(self) + + layout = QVBoxLayout(self) + layout.addWidget(self.tabs) + + common = copy.deepcopy(SING_TUN_DEFAULTS) + + for key, value in self._original.items(): + if key in ('tun_options', 'stack_options', 'host_options') and isinstance( + value, dict + ): + common[key].update(value) + elif key in common: + common[key] = value + + bridge = self._page(_('SOCKS Transit')) + + self._choice( + bridge, + 'stack', + _('Stack'), + [ + ('go', 'go'), + ('gvisor', 'gvisor'), + ('system', 'system'), + ('mixed', 'mixed'), + (_('Automatic'), ''), + ], + common.get('stack'), + ) + self._choice( + bridge, + 'log_level', + _('Log Level'), + [ + (value, value) + for value in ('trace', 'debug', 'info', 'warn', 'error', 'silent') + ], + common.get('log_level'), + ) + self._text( + bridge, + 'network_interface', + _('SOCKS Interface Binding (empty for automatic)'), + common.get('network_interface'), + ) + self._number( + bridge, + 'max_sessions', + _('Maximum Sessions'), + common.get('max_sessions'), + 1, + 16384, + ) + + for key, label in ( + ('connect_timeout', _('Connect Timeout (seconds)')), + ('udp_timeout', _('SOCKS UDP Timeout (seconds)')), + ('tcp_idle_timeout', _('TCP Idle Timeout (seconds)')), + ): + self._text(bridge, key, label, common.get(key)) + + native = self._page(_('Interface and Stack')) + tun = ( + common['tun_options'] + if isinstance(common.get('tun_options'), dict) + else SING_TUN_DEFAULTS['tun_options'] + ) + + self._text( + native, 'Name', _('Device Name (empty for automatic)'), tun.get('Name', '') + ) + self._number(native, 'MTU', _('MTU (bytes)'), tun.get('MTU', 1500), 68, 65535) + self._text( + native, + 'Inet4Address', + _('IPv4 Address Prefixes (comma separated)'), + ( + ', '.join(str(value) for value in tun.get('Inet4Address', [])) + if isinstance(tun.get('Inet4Address'), list) + else '' + ), + ) + self._text( + native, + 'Inet6Address', + _('IPv6 Address Prefixes (comma separated)'), + ( + ', '.join(str(value) for value in tun.get('Inet6Address', [])) + if isinstance(tun.get('Inet6Address'), list) + else '' + ), + ) + + stack = ( + common['stack_options'] + if isinstance(common.get('stack_options'), dict) + else {} + ) + + self._choice( + native, + 'TCPCongestionControl', + _('TCP Congestion Control (Go stack)'), + [ + (_('Automatic'), ''), + ('cubic', 'cubic'), + ('reno', 'reno'), + ('bbr', 'bbr'), + ], + stack.get('TCPCongestionControl', ''), + ) + self._text( + native, + 'UDPTimeout', + _('Native UDP Timeout (Go duration, e.g. 1m)'), + stack.get('UDPTimeout', '1m'), + ) + self._text( + native, + 'ICMPTimeout', + _('Native ICMP Timeout (Go duration, 0 for default)'), + stack.get('ICMPTimeout', '0'), + ) + self._text( + native, + 'UDPNATMax', + _('Native UDP NAT Limit (0 for default)'), + stack.get('UDPNATMax', 0), + ) + for key, label in ( + ('UDPMapping', _('UDP Mapping')), + ('UDPFiltering', _('UDP Filtering')), + ): + self._choice( + native, + key, + label, + [ + (_('Endpoint Independent'), 0), + (_('Address Dependent'), 1), + (_('Address and Port Dependent'), 2), + ], + stack.get(key, 0), + ) + + host = ( + common['host_options'] + if isinstance(common.get('host_options'), dict) + else SING_TUN_HOST_DEFAULTS + ) + hostForm = self._page(_('Host Settings')) + + for key, label in ( + ('primaryAdapterInterfaceName', _('Primary Adapter Interface Name')), + ('primaryAdapterInterfaceIP', _('Primary Adapter Interface IP')), + ('tunAdapterInterfaceDNS', _('TUN Adapter Interface DNS')), + ( + 'bypassTUNAdapterInterfaceIP', + _('Bypass TUN Adapter Interface IP (separated by commas)'), + ), + ): + self._text(hostForm, key, label, host.get(key, '')) + + self.disableDNS = AppQSwitch() + self.disableDNS.setChecked( + host.get('disablePrimaryAdapterInterfaceDNS', True) is True + ) + + disableDNSRow = QHBoxLayout() + disableDNSRow.setContentsMargins(0, 0, 0, 0) + disableDNSRow.addWidget( + AppQLabel( + _( + 'Disable Primary Adapter Interface DNS (Mitigating DNS leaks on Windows)' + ) + ) + ) + disableDNSRow.addWidget(self.disableDNS) + disableDNSRow.addStretch() + hostForm.addRow(disableDNSRow) + + advanced = self._page(_('Advanced Options')) + + self.advanced = DraculaJSONTextEditor(fontFamily=AppFontName()) + self.advanced.setLineWrapMode(DraculaJSONTextEditor.LineWrapMode.NoWrap) + + extra = copy.deepcopy(self._original) + + for key in SING_TUN_DEFAULTS: + if key not in ('tun_options', 'stack_options', 'host_options'): + extra.pop(key, None) + + for key, represented in ( + ('tun_options', ('Name', 'MTU', 'Inet4Address', 'Inet6Address')), + ('host_options', tuple(SING_TUN_HOST_DEFAULTS)), + ( + 'stack_options', + ( + 'UDPTimeout', + 'ICMPTimeout', + 'UDPNATMax', + 'UDPMapping', + 'UDPFiltering', + 'TCPCongestionControl', + ), + ), + ): + if isinstance(extra.get(key), dict): + for field in represented: + extra[key].pop(field, None) + + self.advanced.setPlainText(json.dumps(extra, indent=2, ensure_ascii=False)) + + advanced.addRow(self.advanced) + instructions = AppQLabel( + _( + 'Use exact native JSON field names. Supported: route address/exclusion lists, StrictRoute, InterfaceScope, ForwarderBindInterface and IncludeAllNetworks. AutoRoute, external configuration and DNS ownership are managed. Unsupported fields are rejected.' + ) + ) + instructions.setWordWrap(True) + advanced.addRow(instructions) + + self.errorLabel = AppQLabel('') + self.errorLabel.setWordWrap(True) + + layout.addWidget(self.errorLabel) + + buttons = AppQDialogButtonBox( + AppQDialogButtonBox.StandardButton.Ok + | AppQDialogButtonBox.StandardButton.Cancel + ) + connectWeakly(buttons.accepted, self, 'accept') + connectWeakly(buttons.rejected, self, 'reject') + + layout.addWidget(buttons) + + def _page(self, title): + page = QWidget() + + form = QFormLayout(page) + form.setFieldGrowthPolicy(QFormLayout.FieldGrowthPolicy.ExpandingFieldsGrow) + + scroll = QScrollArea() + scroll.setWidgetResizable(True) + scroll.setWidget(page) + + self.tabs.addTab(scroll, title) + + return form + + def _text(self, form, key, label, value): + widget = AppQLineEdit() + widget.setText(str(value) if value is not None else '') + + self.fields[key] = widget + + form.addRow(AppQLabel(label), widget) + + def _number(self, form, key, label, value, minimum, maximum): + widget = AppQSpinBox() + widget.setRange(minimum, maximum) + widget.setValue(value if type(value) is int else minimum) + + self.fields[key] = widget + + form.addRow(AppQLabel(label), widget) + + def _choice(self, form, key, label, choices, value): + widget = AppQComboBox() + + for text, data in choices: + widget.addItem(text, data) + + widget.setCurrentIndex(max(widget.findData(value), 0)) + + self.fields[key] = widget + + form.addRow(AppQLabel(label), widget) + + def document(self): + """Prepare sing-tun's complete candidate before its live collection changes.""" + document = json.loads(self.advanced.toPlainText()) + + if not isinstance(document, dict): + raise ValueError('Advanced options must be a JSON object') + + tun = document.setdefault('tun_options', {}) + stack = document.setdefault('stack_options', {}) + + if not isinstance(tun, dict) or not isinstance(stack, dict): + raise ValueError('Native options must be JSON objects') + + for key in ('stack', 'log_level'): + document[key] = self.fields[key].currentData() + + document['network_interface'] = self.fields['network_interface'].text().strip() + document['max_sessions'] = self.fields['max_sessions'].value() + + for key in ('connect_timeout', 'udp_timeout', 'tcp_idle_timeout'): + document[key] = float(self.fields[key].text()) + + tun['Name'], tun['MTU'] = ( + self.fields['Name'].text().strip(), + self.fields['MTU'].value(), + ) + + for key in ('Inet4Address', 'Inet6Address'): + tun[key] = [ + value.strip() + for value in self.fields[key].text().split(',') + if value.strip() + ] + + for key in ('UDPTimeout', 'ICMPTimeout'): + value = self.fields[key].text().strip() + stack[key] = int(value) if value.isdigit() else value + + stack['UDPNATMax'] = int(self.fields['UDPNATMax'].text()) + + for key in ('UDPMapping', 'UDPFiltering', 'TCPCongestionControl'): + stack[key] = self.fields[key].currentData() + + host = document.setdefault('host_options', {}) + + if not isinstance(host, dict): + raise ValueError('sing-tun host options must be a JSON object') + + for key in ( + 'primaryAdapterInterfaceName', + 'primaryAdapterInterfaceIP', + 'tunAdapterInterfaceDNS', + 'bypassTUNAdapterInterfaceIP', + ): + host[key] = self.fields[key].text().strip() + + host['disablePrimaryAdapterInterfaceDNS'] = self.disableDNS.isChecked() + + prepareSingTUNSettings(document, platform=PLATFORM) + + return document + + def accept(self): + try: + document = self.document() + + Storage.replaceSingTUNSettings(document) + except (ValueError, TypeError, OSError) as ex: + self.errorLabel.setText(str(ex)) + + return + + super().accept() diff --git a/Furious/Window/TunSettingsDialog.py b/Furious/Window/TunSettingsDialog.py index 75fb2275..6412d4b7 100644 --- a/Furious/Window/TunSettingsDialog.py +++ b/Furious/Window/TunSettingsDialog.py @@ -264,11 +264,11 @@ class GuiTUNSettingsGroupBoxBasic(GuiEditorWidgetQGroupBox): key='defaultPrimaryGatewayIP', ), GuiTUNSettingsItemXXX( - title=_('Tun2socks Adapter Interface DNS'), + title=_('TUN Adapter Interface DNS'), key='tunAdapterInterfaceDNS', ), GuiTUNSettingsItemXXX( - title=_('Bypass Tun2socks Adapter Interface IP (separated by commas)'), + title=_('Bypass TUN Adapter Interface IP (separated by commas)'), key='bypassTUNAdapterInterfaceIP', ), GuiTUNSettingsItemSwitchXXX( diff --git a/tests/README.md b/tests/README.md index 8e714f5b..cc80e940 100644 --- a/tests/README.md +++ b/tests/README.md @@ -94,6 +94,7 @@ worker. Choose tests by the changed contract rather than by filename alone. | [test_external_core.py](test_external_core.py) | Harmless real process launch/output/shutdown, partial thread-start rollback, non-finite timeout rejection, failed reap/thread-join retry, readiness/TUN metadata, Windows paths with spaces, subscription rejection of executable profiles, bounded DNS references. | | [test_frozenlib.py](test_frozenlib.py) | Nested state guards, cleanup isolation, bounded caches/throttling, dual-stack probe selection, mocked proxy/DNS/routes/startup/session boundaries and failure handling. | | [test_native_tun_semantics.py](test_native_tun_semantics.py) | Xray/Hysteria2 runtime-copy TUN preservation/replacement, managed-TUN failures, download-test stripping, prevention of a second tun2socks owner. | +| [test_sing_tun.py](test_sing_tun.py) | Engine choice/Go default, development-stack and congestion-control compatibility, independent backend settings and host defaults, selected-only snapshots, captured TUN log sources in both startup paths, staged settings and malformed storage, SOCKS authentication/UDP, harmless real spawn/status/stop, native readiness, cancellation, synchronous compatibility, mocked DNS/routing ownership including development IPv6 rule ordering and refused cleanup. | | [test_subscription_sync.py](test_subscription_sync.py) | Group-local preparation/commit, stable duplicate identity, atomic failure, preservation of newer local metadata, rejection of changed source state. | | [test_subscription_manager.py](test_subscription_manager.py) | Stable request generations, provider metadata, batch/partial failures, timer policy, targeted updates, real Stop Updates input, worker responsiveness, cancellation and synchronous shutdown ownership. | | [test_subscription_scalability.py](test_subscription_scalability.py) | Deterministic 1/3/8-group preparation and commit with 1,500 profiles per group and bounded workers; uses the offline benchmark helper. | @@ -140,7 +141,7 @@ regular cases run by default while its heavy class remains gated. python -m unittest tests.test_interface tests.test_models_and_services tests.test_repository_contracts tests.test_plugin_architecture tests.test_public_api tests.test_hysteria1_protocol tests.test_hysteria2_compatibility tests.test_backend_editor_contract tests.test_shadowsocks_uri tests.test_socks_uri -v # Application, connection, runtime, and mocked host boundaries -python -m unittest tests.test_application_process tests.test_architecture_refactors tests.test_connection_startup_async tests.test_controllers tests.test_runtime_lifecycle tests.test_external_core tests.test_frozenlib tests.test_native_tun_semantics -v +python -m unittest tests.test_application_process tests.test_architecture_refactors tests.test_connection_startup_async tests.test_controllers tests.test_runtime_lifecycle tests.test_external_core tests.test_frozenlib tests.test_native_tun_semantics tests.test_sing_tun -v # Subscription, profile tests, HTTP/assets, endpoint information, metrics, and logs python -m unittest tests.test_subscription_sync tests.test_subscription_manager tests.test_subscription_scalability tests.test_profile_test_jobs tests.test_service_runtime tests.test_xray_asset_download tests.test_endpoint_info tests.test_metrics_behavior tests.test_log_manager_generation tests.test_log_manager_benchmark -v diff --git a/tests/fixtures/sing_tun_lifetime_probe.py b/tests/fixtures/sing_tun_lifetime_probe.py new file mode 100644 index 00000000..da641953 --- /dev/null +++ b/tests/fixtures/sing_tun_lifetime_probe.py @@ -0,0 +1,116 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# This file is part of Furious. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""Harmless source/Nuitka probe for the new transient settings editor.""" + +from __future__ import annotations + +from Furious.Qt import AppQDialog +from Furious.Repository import Storage +from Furious.Window.SingTUNSettingsDialog import SingTUNSettingsDialog + +from PySide6.QtWidgets import QWidget +from shiboken6 import isValid + +from tests.support import application, isolatedSettings, processQtEvents + +import argparse +import copy +import gc +import json +import sys +import weakref +import importlib + + +def run(iterations, method): + application() + destroyed = [] + references = [] + errors = [] + + originalHook = sys.excepthook + sys.excepthook = lambda kind, error, traceback: errors.append( + (kind.__name__, str(error)) + ) + singGetter, replace = ( + Storage.UserSingTUNSettings, + Storage.replaceSingTUNSettings, + ) + sing = {} + + Storage.UserSingTUNSettings = staticmethod(lambda: copy.deepcopy(sing)) + Storage.replaceSingTUNSettings = staticmethod(lambda document: None) + baseline = len(AppQDialog._openDialogs) + + try: + with isolatedSettings(): + for index in range(iterations): + owner = QWidget() + dialog = SingTUNSettingsDialog(owner) + dialog.destroyed.connect(lambda *_args: destroyed.append(True)) + references.append(weakref.ref(dialog)) + + dialog.open() + if method == 'owner': + owner.deleteLater() + else: + getattr(dialog, method)() + processQtEvents() + + assert not isValid(dialog), index + + if isValid(owner): + owner.deleteLater() + processQtEvents() + del dialog, owner + + gc.collect() + processQtEvents() + + assert len(destroyed) == iterations + assert all(reference() is None for reference in references) + assert len(AppQDialog._openDialogs) == baseline + assert not errors, errors + finally: + sys.excepthook = originalHook + Storage.UserSingTUNSettings = staticmethod(singGetter) + Storage.replaceSingTUNSettings = staticmethod(replace) + + return { + 'method': method, + 'destroyed': len(destroyed), + 'wrapperReleased': iterations, + 'callbackErrors': len(errors), + 'compiledDialog': hasattr( + importlib.import_module('Furious.Window.SingTUNSettingsDialog'), + '__compiled__', + ), + 'compiledSignals': hasattr( + importlib.import_module('Furious.Qt.Signals'), '__compiled__' + ), + } + + +if __name__ == '__main__': + parser = argparse.ArgumentParser() + parser.add_argument('--iterations', type=int, default=20) + parser.add_argument( + '--method', choices=('accept', 'reject', 'close', 'owner'), default='reject' + ) + arguments = parser.parse_args() + print(json.dumps(run(arguments.iterations, arguments.method))) diff --git a/tests/test_sing_tun.py b/tests/test_sing_tun.py new file mode 100644 index 00000000..7b98ac7c --- /dev/null +++ b/tests/test_sing_tun.py @@ -0,0 +1,1603 @@ +# Copyright (C) 2024–present Loren Eteval & contributors +# +# This file is part of Furious. +# +# This program is free software: you can redistribute it and/or modify +# it under the terms of the GNU General Public License as published by +# the Free Software Foundation, either version 3 of the License, or +# (at your option) any later version. +# +# This program is distributed in the hope that it will be useful, +# but WITHOUT ANY WARRANTY; without even the implied warranty of +# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the +# GNU General Public License for more details. +# +# You should have received a copy of the GNU General Public License +# along with this program. If not, see . + +"""Isolated sing-tun policy, UI, native-child and host ownership contracts.""" + +from __future__ import annotations + +from Furious.Core.SingTUN import SingTUN, _runEngine +from Furious.Core.MultiprocessingRuntime import ( + ProcessLaunchSpec, + MultiprocessingRuntime, +) +from Furious.Interface import CoreRuntime, RuntimeState, RuntimeExitReason +from Furious.Frozenlib import AppSettings +from Furious.Models.SingTUN import prepareSingTUNSettings, addSingTUNExclusions +from Furious.Frozenlib.SocksProxy import socksURL +from Furious.Repository import Storage +from Furious.Repository.SingTUNSettings import UserSingTUNSettings +from Furious.Repository.TunSettings import UserTUNSettings +from Furious.Service.ConnectionManager import ( + ConnectionManager, + ConnectionStartStage, + _ConnectionStartAttempt, +) +from Furious.Service.SingTUNHost import ( + SingTUNHostPlan, + _routeRanges, + _linuxRuleDelete, + _darwinRoutes, +) +from Furious.Plugins import PreparedRuntime, CoreRuntimeStartup, TUNPreparationError +from Furious.Window.SingTUNSettingsDialog import SingTUNSettingsDialog +from Furious.Window.TunSettingsDialog import TunSettingsDialog +from Furious.Backends.Configuration import ConfigXray, ConfigHysteria1, ConfigHysteria2 +from Furious.Controllers.SettingsController import SettingsController +from Furious.Qt import gettext as _ +from Furious.Service.RuntimeLease import RuntimeEventRouter, RuntimeLeaseState +from Furious.Service.LogManager import LogManager, TUN_LOG_CATEGORY + +from PySide6 import QtCore +from PySide6.QtWidgets import QLabel, QWidget +from shiboken6 import isValid + +from tests.support import application, isolatedSettings, processQtEvents, waitFor +from tests.test_connection_startup_async import _Configuration, _Registry, _Runtime + +from unittest import mock + +import copy +import importlib +import json +import multiprocessing +import subprocess +import sys +import threading +import tempfile +import unittest +import ipaddress + + +class _Engine: + """Harmless native stand-in, never acquires an interface or route.""" + + def __init__(self, configuration): + if configuration.get('missing', False): + raise ImportError('unavailable native library with sensitive loader data') + if configuration.get('invalid', False): + raise ValueError('binding configuration validation failed') + + self.ready = False + self.status = {'error': ''} + self.device_name = 'utun101' + self.stopped = threading.Event() + self.block = configuration.get('block', False) + self.fail = configuration.get('fail', False) + self.failWithoutStatus = configuration.get('fail_without_status', False) + self.runtimeFail = configuration.get('runtime_fail', False) + + def start(self): + if self.block: + self.stopped.wait(2) + raise RuntimeError('cancelled') + if self.fail: + self.status['error'] = 'new binding startup failure wording' + raise RuntimeError('do not expose socks5://secret:password@127.0.0.1:1080') + if self.failWithoutStatus: + raise RuntimeError('binding initialization exception without status') + + self.ready = True + + def stop(self): + self.stopped.set() + + def wait(self, timeout): + if self.runtimeFail: + self.status['error'] = 'new binding runtime failure wording' + raise RuntimeError('do not expose runtime secrets') + return self.stopped.wait(timeout) + + def close(self, _timeout): + self.stop() + + +def _fakeChild(_output, configuration, control): + _runEngine(configuration, control, _Engine) + + +class _ChildRuntime(SingTUN): + """Use the actual spawn/IPC owner with a harmless child entry point.""" + + def start(self): + self._started = True + self._launch = ProcessLaunchSpec( + target=_fakeChild, + args=(self._output, self._configuration, self._childControl), + ) + try: + MultiprocessingRuntime.start(self) + self._monitor.setInterval(10) + finally: + self._childControl.close() + + +class _PreparedSing(SingTUN): + """Drive service readiness without multiprocessing or host commands.""" + + def __init__(self, configuration, *, hostPlan, exitCallback=None, **_kwargs): + CoreRuntime.__init__(self, exitCallback) + self._hostPlan = hostPlan + self._configuration = copy.deepcopy(configuration) + self.alive = False + self.nativeReady = False + self.released = False + + @property + def startupError(self): + return '' + + @property + def deviceName(self): + return 'utun101' + + @property + def ready(self): + return self.nativeReady and self.alive + + def start(self): + self.alive = True + self.setState(RuntimeState.Alive) + + def stop(self): + self.alive = False + self.released = True + + def dispose(self): + self.stop() + + def isRunning(self): + return self.alive + + +class _Plan: + def __init__(self, configuration): + self.configuration = copy.deepcopy(configuration) + self.hostSettings = self.configuration.pop('host_options') + self.error = '' + self.deviceName = 'utun101' + self.addresses = [] + self.applied = [] + + def begin(self, method, *args): + if method == 'prepare': + self.addresses = args[0] + elif method == 'applyDNS': + self.applied.append(args[0]) + + def workCompleted(self): + return True + + def prepare(self, addresses): + self.addresses = addresses + + def applyDNS(self, device): + self.applied.append(device) + + +class SingTUNConfigurationTest(unittest.TestCase): + def testHostOptionsHaveIndependentDefaultsAndNeverEnterNativeConfig(self): + stored = {'host_options': {'tunAdapterInterfaceDNS': '9.9.9.9'}} + prepared = prepareSingTUNSettings(stored) + plan = SingTUNHostPlan(prepared) + + self.assertEqual(plan.hostSettings['tunAdapterInterfaceDNS'], '9.9.9.9') + self.assertTrue(plan.hostSettings['disablePrimaryAdapterInterfaceDNS']) + self.assertNotIn('host_options', plan.configuration) + plan.hostSettings['tunAdapterInterfaceDNS'] = '1.0.0.1' + self.assertEqual( + stored, {'host_options': {'tunAdapterInterfaceDNS': '9.9.9.9'}} + ) + self.assertEqual( + prepareSingTUNSettings({})['host_options']['tunAdapterInterfaceDNS'], '' + ) + + def testMalformedHostOptionsAreRejectedBeforeActivation(self): + for host in ( + None, + [], + {'tunAdapterInterfaceDNS': 'invalid'}, + {'disablePrimaryAdapterInterfaceDNS': 'False'}, + {'tcpSendBufferSize': 4}, + {'bypassTUNAdapterInterfaceIP': '192.0.2.1,invalid'}, + ): + with self.subTest(host=host), self.assertRaises(ValueError): + prepareSingTUNSettings({'host_options': host}) + + def testDefaultsCopyAndOwnedFields(self): + original = {'tun_options': {'MTU': 1400}} + prepared = prepareSingTUNSettings(original) + + self.assertEqual(original, {'tun_options': {'MTU': 1400}}) + self.assertEqual(prepared['stack'], 'go') + self.assertEqual(prepared['tun_options']['Inet4Address'], ['198.18.0.1/15']) + + for field, value in [ + ('AutoRoute', False), + ('DNSMode', 'native'), + ('EXP_ExternalConfiguration', True), + ('EXP_DisableDNSHijack', False), + ('FileDescriptor', 3), + ('IPRoute2TableIndex', 99), + ]: + with self.subTest(field=field), self.assertRaises(ValueError): + prepareSingTUNSettings({'tun_options': {field: value}}) + + def testValidationRejectsUnsupportedAndMalformedValues(self): + for value in [ + {'stack': 'unsupported'}, + {'unknown': True}, + {'max_sessions': True}, + {'udp_timeout': float('nan')}, + {'tun_options': {'MTU': 0}}, + {'tun_options': {'Inet4Address': ['::1/128']}}, + {'stack_options': {'UDPTimeout': '0s'}}, + {'stack': 'system', 'stack_options': {'IncludeAllNetworks': True}}, + {'stack': '', 'stack_options': {'IncludeAllNetworks': True}}, + {'stack': 'go', 'stack_options': {'IncludeAllNetworks': True}}, + {'stack_options': {'ForwarderBindInterface': True}}, + {'stack_options': {'TCPCongestionControl': 'invalid'}}, + {'stack_options': {'TCPCongestionControl': True}}, + {'stack': 'gvisor', 'stack_options': {'TCPCongestionControl': 'bbr'}}, + ]: + with self.subTest(value=value), self.assertRaises(ValueError): + prepareSingTUNSettings(value) + + def testGoAndPreviousStacksPreserveExplicitChoicesAndCongestionControl(self): + for stack in ('', 'go', 'gvisor', 'system', 'mixed'): + original = {'stack': stack} + self.assertEqual(prepareSingTUNSettings(original)['stack'], stack) + self.assertEqual(original, {'stack': stack}) + + for stack in ('', 'go'): + for congestion in ('', 'cubic', 'reno', 'bbr'): + original = { + 'stack': stack, + 'stack_options': {'TCPCongestionControl': congestion}, + } + self.assertEqual( + prepareSingTUNSettings(original)['stack_options'][ + 'TCPCongestionControl' + ], + congestion, + ) + + self.assertTrue( + prepareSingTUNSettings( + {'stack': 'gvisor', 'stack_options': {'IncludeAllNetworks': True}} + )['stack_options']['IncludeAllNetworks'] + ) + + def testEndpointsIncludingAuthentication(self): + expected = { + '0.0.0.0:1080': 'socks5://127.0.0.1:1080', + 'localhost:1080': 'socks5://127.0.0.1:1080', + ':::1080': 'socks5://[::1]:1080', + '[::1]:1080': 'socks5://[::1]:1080', + 'socks5://u:p%40ss@[::1]:1080': 'socks5://u:p%40ss@[::1]:1080', + } + for value, normalized in expected.items(): + self.assertEqual(socksURL(value), normalized) + + for value in ( + '', + 'http://127.0.0.1:1080', + 'example.com:1080', + '127.0.0.1:0', + '127.0.0.1:65536', + 'socks5://u:@127.0.0.1:1080', + ): + with self.assertRaises(ValueError): + socksURL(value) + + def testDualStackBypassesPreserveStoredExclusions(self): + original = prepareSingTUNSettings( + {'tun_options': {'Inet4RouteExcludeAddress': ['192.0.2.0/24']}} + ) + prepared = addSingTUNExclusions( + original, ['192.0.2.1', '2001:db8::1', '192.0.2.1'] + ) + + self.assertEqual( + original['tun_options']['Inet4RouteExcludeAddress'], ['192.0.2.0/24'] + ) + self.assertEqual( + prepared['tun_options']['Inet6RouteExcludeAddress'], ['2001:db8::1/128'] + ) + self.assertEqual(len(prepared['tun_options']['Inet4RouteExcludeAddress']), 2) + + def testBackendAuthenticationAndUDPPolicy(self): + xray = ConfigXray( + { + 'inbounds': [ + { + 'listen': '::', + 'port': 1080, + 'protocol': 'socks', + 'settings': { + 'udp': True, + 'auth': 'password', + 'accounts': [{'user': 'u', 'pass': 'p@ss'}], + }, + } + ] + } + ) + self.assertEqual(xray.applicationTUNProxy(), 'socks5://u:p%40ss@[::1]:1080') + + xray['inbounds'][0]['settings']['accounts'][0]['pass'] = 'p/@ss' + self.assertEqual(xray.applicationTUNProxy(), 'socks5://u:p%2F%40ss@[::1]:1080') + + xray['inbounds'][0]['settings']['udp'] = False + with self.assertRaises(ValueError): + xray.applicationTUNProxy() + + for cls, user, disable in ( + (ConfigHysteria1, 'user', 'disable_udp'), + (ConfigHysteria2, 'username', 'disableUDP'), + ): + configuration = cls( + {'socks5': {'listen': '127.0.0.1:1080', user: 'u', 'password': 'p'}} + ) + self.assertEqual( + configuration.applicationTUNProxy(), 'socks5://u:p@127.0.0.1:1080' + ) + configuration['socks5'][disable] = True + with self.assertRaises(ValueError): + configuration.applicationTUNProxy() + + def testPublishedNativeAPIWithoutTUNAcquisition(self): + for settings in ( + {}, + {'stack': 'gvisor'}, + {'stack': 'system'}, + {'stack': 'mixed'}, + {'stack': ''}, + {'stack_options': {'TCPCongestionControl': 'bbr'}}, + { + 'max_sessions': 512, + 'tun_options': {'MTU': 1450}, + 'stack_options': { + 'UDPTimeout': '15s', + 'ICMPTimeout': 0, + 'UDPNATMax': 64, + }, + }, + ): + configuration = prepareSingTUNSettings(settings) + configuration['proxy'] = 'socks5://127.0.0.1:1080' + configuration = SingTUNHostPlan(configuration).configuration + with tempfile.TemporaryDirectory() as directory: + result = subprocess.run( + [ + sys.executable, + '-c', + 'import json,sys,importlib.metadata; ' + 'from sing_tun import Config, Engine, capabilities; ' + 'assert importlib.metadata.version("sing-tun") == "0.9.7.dev0"; ' + 'assert Config(proxy="socks5://127.0.0.1:1080").stack == "go"; ' + 'assert "go" in capabilities()["stacks"]; ' + 'e=Engine(Config(**json.load(sys.stdin))); assert not e.ready; e.close(5); ' + 'assert capabilities(); print("Installed API OK")', + ], + input=json.dumps(configuration).encode(), + capture_output=True, + cwd=directory, + timeout=20, + ) + self.assertEqual( + result.returncode, 0, result.stderr.decode(errors='replace') + ) + + +class SingTUNUIAndStorageTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + cls.app = application() + + def testTwoChoiceSelectorRetranslatesWithoutChangingStablePreference(self): + module = importlib.import_module('Furious.Window.SettingsPage') + with isolatedSettings(), mock.patch( + 'Furious.Controllers.SettingsController.showMBoxNewChangesNextTime' + ): + controller = SettingsController() + with mock.patch.object( + module, 'AppSettingsController', return_value=controller + ): + card = module._TUNBackendSettingsCard() + self.assertEqual( + [ + card.comboBox.itemData(index) + for index in range(card.comboBox.count()) + ], + ['sing-tun', 'tun2socks'], + ) + + for backend in ('tun2socks', 'sing-tun', 'tun2socks'): + card.comboBox.setCurrentIndex(card.comboBox.findData(backend)) + self.assertEqual(AppSettings.get('ApplicationTUNBackend'), backend) + + for language in ('ZH', 'RU', 'EN'): + AppSettings.set('Language', language) + card.titleLabel.retranslate() + card.descriptionLabel.retranslate() + card.comboBox.retranslate() + self.assertEqual( + card.titleLabel.text(), _('Application TUN Engine', language) + ) + self.assertEqual(card.comboBox.currentData(), 'tun2socks') + + self.assertFalse(AppSettings.isStateON_('VPNMode')) + + card.deleteLater() + processQtEvents() + + def testIndependentRoundTripAndMalformedPreservation(self): + with isolatedSettings(): + AppSettings.set('CustomSingTUNSettings', 'malformed') + with self.assertLogs('Furious.Repository.SingTUNSettings', level='ERROR'): + repository = UserSingTUNSettings() + repository.cleanup() + self.assertEqual(AppSettings.get('CustomSingTUNSettings'), 'malformed') + + repository.data()['unknown'] = {'future': 1} + repository.sync() + self.assertEqual(UserSingTUNSettings().data(), {'unknown': {'future': 1}}) + + def _dialog(self, sing): + patches = ( + mock.patch.object(Storage, 'UserSingTUNSettings', return_value=sing), + mock.patch.object( + Storage, + 'UserTUNSettings', + side_effect=AssertionError('tun2socks settings must not be read'), + ), + ) + with patches[0], patches[1]: + return SingTUNSettingsDialog() + + def testGoDefaultAndCongestionControlRoundTripPreservePreviousStackChoices(self): + for stack in (None, '', 'go', 'gvisor', 'system', 'mixed'): + saved = {} if stack is None else {'stack': stack} + baseline = copy.deepcopy(saved) + dialog = self._dialog(saved) + expected = 'go' if stack is None else stack + self.assertEqual(dialog.fields['stack'].currentData(), expected) + self.assertEqual(dialog.document()['stack'], expected) + self.assertEqual(saved, baseline) + dialog.reject() + processQtEvents() + + saved = {'stack_options': {'TCPCongestionControl': 'bbr'}} + dialog = self._dialog(saved) + self.assertEqual(dialog.fields['TCPCongestionControl'].currentData(), 'bbr') + self.assertEqual( + dialog.document()['stack_options']['TCPCongestionControl'], 'bbr' + ) + self.assertNotIn('TCPCongestionControl', dialog.advanced.toPlainText()) + for language in ('ZH', 'RU'): + self.assertNotEqual( + _('TCP Congestion Control (Go stack)', language), + 'TCP Congestion Control (Go stack)', + ) + dialog.reject() + processQtEvents() + + def testCancelAndInvalidAreObservational(self): + sing = {'tun_options': {'MTU': 1450}} + baseline = copy.deepcopy(sing) + dialog = self._dialog(sing) + dialog.open() + + dialog.fields['udp_timeout'].setText('invalid') + with mock.patch.object(Storage, 'replaceSingTUNSettings') as commit: + dialog.accept() + commit.assert_not_called() + + self.assertEqual(sing, baseline) + self.assertTrue(dialog.errorLabel.text()) + + dialog.reject() + processQtEvents() + self.assertFalse(isValid(dialog)) + + def testDialogPreparesNativeAndHostOptionsInOneIndependentDocument(self): + sing = {'host_options': {'tunAdapterInterfaceDNS': '9.9.9.9'}} + dialog = self._dialog(sing) + dialog.fields['MTU'].setValue(1400) + dialog.fields['bypassTUNAdapterInterfaceIP'].setText('192.0.2.4,2001:db8::4') + document = dialog.document() + + self.assertEqual(document['tun_options']['MTU'], 1400) + self.assertEqual(document['host_options']['tunAdapterInterfaceDNS'], '9.9.9.9') + self.assertEqual( + document['host_options']['bypassTUNAdapterInterfaceIP'], + '192.0.2.4,2001:db8::4', + ) + self.assertNotIn('defaultPrimaryGatewayIP', dialog.fields) + + dialog.fields['tunAdapterInterfaceDNS'].setText('invalid') + with self.assertRaises(ValueError): + dialog.document() + self.assertEqual(sing, {'host_options': {'tunAdapterInterfaceDNS': '9.9.9.9'}}) + + dialog.reject() + processQtEvents() + + def testCommitStagesAliasedLiveViewAndRejectsInvalidSingFields(self): + sing = {'max_sessions': 512} + singRepository = mock.Mock() + singRepository.data.return_value = sing + with isolatedSettings(), mock.patch.object( + Storage, '_UserSingTUNSettingsStorage', return_value=singRepository + ), mock.patch.object( + Storage, + '_UserTUNSettingsStorage', + side_effect=AssertionError('tun2socks storage must not be acquired'), + ): + Storage.replaceSingTUNSettings(sing) + self.assertEqual(sing, {'max_sessions': 512}) + + with self.assertRaises(ValueError): + Storage.replaceSingTUNSettings( + { + 'max_sessions': 256, + 'host_options': {'tunAdapterInterfaceDNS': object()}, + } + ) + + self.assertEqual(sing, {'max_sessions': 512}) + + def testSingFlushFailureReportsItsCompletedCommitWithoutAcquiringTun2socks(self): + sing = {'max_sessions': 512} + repository = mock.Mock() + repository.data.return_value = sing + settings = mock.Mock() + status = QtCore.QSettings.Status + settings.status.return_value = status.AccessError + + with mock.patch.object( + Storage, '_UserSingTUNSettingsStorage', return_value=repository + ), mock.patch.object( + Storage, + '_UserTUNSettingsStorage', + side_effect=AssertionError('inactive backend'), + ), mock.patch.object( + QtCore, 'QSettings', return_value=settings + ) as factory: + factory.Status = status + with self.assertRaisesRegex(OSError, 'committed in memory'): + Storage.replaceSingTUNSettings({'max_sessions': 256}) + + self.assertEqual(sing, {'max_sessions': 256}) + repository.sync.assert_called_once_with() + settings.sync.assert_called_once_with() + + def testOwnHostSettingsLabelsAreTranslatedInBothSupportedLocales(self): + for language, title in ( + ('EN', 'Host Settings'), + ('ZH', '主机设置'), + ('RU', 'Настройки хоста'), + ): + with self.subTest(language=language), isolatedSettings(): + AppSettings.set('Language', language) + dialog = self._dialog({}) + with mock.patch.object(Storage, 'UserTUNSettings', return_value={}): + legacy = TunSettingsDialog() + self.assertEqual(_('Host Settings'), title) + self.assertIn( + title, + [ + dialog.tabs.tabText(index) + for index in range(dialog.tabs.count()) + ], + ) + form = dialog.disableDNS.parentWidget().layout() + legacyLabels = {label.text() for label in legacy.findChildren(QLabel)} + for key, source in ( + ('primaryAdapterInterfaceName', 'Primary Adapter Interface Name'), + ('primaryAdapterInterfaceIP', 'Primary Adapter Interface IP'), + ('tunAdapterInterfaceDNS', 'TUN Adapter Interface DNS'), + ( + 'bypassTUNAdapterInterfaceIP', + 'Bypass TUN Adapter Interface IP (separated by commas)', + ), + ): + expected = _(source) + self.assertEqual( + form.labelForField(dialog.fields[key]).text(), expected + ) + self.assertIn(expected, legacyLabels) + if language != 'EN': + self.assertNotEqual(expected, source) + + dnsLabel = _( + 'Disable Primary Adapter Interface DNS (Mitigating DNS leaks on Windows)' + ) + self.assertIn( + dnsLabel, + { + label.text() + for label in dialog.disableDNS.parentWidget().findChildren( + QLabel + ) + }, + ) + self.assertIn(dnsLabel, legacyLabels) + self.assertEqual(form.rowCount(), 5) + legacy.reject() + legacy.deleteLater() + dialog.reject() + processQtEvents() + + def testSavingEitherBackendPreservesTheOthersLiveAndPersistedSettings(self): + with isolatedSettings(): + tun2socks = UserTUNSettings() + tun2socks.data().update( + { + 'tunAdapterInterfaceDNS': '8.8.8.8', + 'defaultPrimaryGatewayIP': '192.0.2.254', + 'tcpSendBufferSize': 4, + } + ) + tun2socks.sync() + legacyBefore = copy.deepcopy(tun2socks.data()) + legacyRaw = AppSettings.get('CustomTUNSettings') + sing = UserSingTUNSettings() + + with mock.patch.object( + Storage, '_UserSingTUNSettingsStorage', return_value=sing + ), mock.patch.object( + Storage, + '_UserTUNSettingsStorage', + side_effect=AssertionError('sing-tun must not touch tun2socks storage'), + ): + dialog = SingTUNSettingsDialog() + dialog.open() + self.assertEqual(dialog.fields['tunAdapterInterfaceDNS'].text(), '') + dialog.fields['tunAdapterInterfaceDNS'].setText('9.9.9.9') + dialog.fields['bypassTUNAdapterInterfaceIP'].setText('2001:db8::1') + dialog.accept() + processQtEvents() + self.assertFalse(isValid(dialog)) + + self.assertEqual(tun2socks.data(), legacyBefore) + self.assertEqual(AppSettings.get('CustomTUNSettings'), legacyRaw) + self.assertEqual( + UserSingTUNSettings().data()['host_options']['tunAdapterInterfaceDNS'], + '9.9.9.9', + ) + singBefore = copy.deepcopy(sing.data()) + singRaw = AppSettings.get('CustomSingTUNSettings') + + tun2socks.data()['tunAdapterInterfaceDNS'] = '1.0.0.1' + tun2socks.sync() + + self.assertEqual(sing.data(), singBefore) + self.assertEqual(AppSettings.get('CustomSingTUNSettings'), singRaw) + self.assertNotIsInstance(sing, UserTUNSettings) + + def testRepeatedOpenAndOwnerFirstDestruction(self): + for _ in range(5): + owner = QWidget() + with mock.patch.object( + Storage, 'UserSingTUNSettings', return_value={} + ), mock.patch.object( + Storage, + 'UserTUNSettings', + side_effect=AssertionError('inactive backend'), + ): + dialog = SingTUNSettingsDialog(owner) + + dialog.open() + owner.deleteLater() + processQtEvents() + self.assertFalse(isValid(dialog)) + + def testPreferenceDefaultInvalidAndSignal(self): + with isolatedSettings(), mock.patch( + 'Furious.Controllers.SettingsController.showMBoxNewChangesNextTime' + ): + self.assertEqual(AppSettings.get('ApplicationTUNBackend'), 'sing-tun') + QtCore.QSettings().setValue('ApplicationTUNBackend', 'invalid') + self.assertEqual(AppSettings.get('ApplicationTUNBackend'), 'sing-tun') + + controller = SettingsController() + changes = [] + controller.tunBackendChanged.connect(changes.append) + + controller.setTUNBackend('tun2socks') + controller.setTUNBackend('tun2socks') + with self.assertRaises(ValueError): + controller.setTUNBackend('invalid') + + self.assertEqual(changes, ['tun2socks']) + self.assertFalse(AppSettings.isStateON_('VPNMode')) + + +class SingTUNChildTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + application() + + def testSpawnReadyCooperativeStopAndFinalHandleDisposal(self): + runtime = _ChildRuntime({}) + try: + self.assertEqual(runtime._processContext.get_start_method(), 'spawn') + runtime.start() + self.assertTrue(waitFor(lambda: runtime.ready, timeout=10)) + self.assertEqual(runtime.deviceName, 'utun101') + + child = runtime.process + runtime.stop() + + self.assertEqual(runtime._status['state'], 'stopped') + self.assertIsNone(runtime.process) + self.assertTrue(child._closed) + + runtime.dispose() + self.assertTrue(runtime._control.closed) + finally: + runtime.dispose() + + def testInvalidAndOversizeStatusCannotSupplyReadiness(self): + for payload in ( + b'{"state":"ready","device_name":false}', + b'{"state":"failed"}', + b'{"state":"failed","error":""}', + b'{"state":"failed","error":null}', + b'{"state":"failed","error":false}', + b'{"state":"failed","error":["failure"]}', + b'{"state":"failed","error":{"reason":"failure"}}', + b'x' * 4097, + ): + with self.subTest(payloadSize=len(payload)): + runtime = SingTUN({}) + try: + runtime._childControl.send_bytes(payload) + self.assertEqual( + runtime.startupError, 'Invalid sing-tun status message' + ) + self.assertFalse(runtime.ready) + self.assertEqual(runtime.deviceName, '') + finally: + runtime.dispose() + + def testStopDuringBlockedInitialization(self): + runtime = _ChildRuntime({'block': True}) + try: + runtime.start() + runtime.stop() + self.assertFalse(runtime.isRunning()) + finally: + runtime.dispose() + + def testBindingStatusReasonIsForwardedWithoutLiteralMatching(self): + runtime = _ChildRuntime({'fail': True}) + try: + runtime.start() + self.assertTrue(waitFor(lambda: bool(runtime.startupError), timeout=10)) + self.assertFalse(runtime.ready) + self.assertEqual( + runtime.startupError, 'sing-tun: new binding startup failure wording' + ) + event = runtime.interpretExit(1) + self.assertEqual(event.reason, RuntimeExitReason.StartFailure) + self.assertEqual(event.message, runtime.startupError) + finally: + runtime.dispose() + + def testBindingExceptionsAreForwardedWhenStatusHasNoReason(self): + for configuration, reason in ( + ({'invalid': True}, 'binding configuration validation failed'), + ( + {'fail_without_status': True}, + 'binding initialization exception without status', + ), + ): + with self.subTest(configuration=configuration): + runtime = _ChildRuntime(configuration) + try: + runtime.start() + self.assertTrue( + waitFor(lambda: bool(runtime.startupError), timeout=10) + ) + self.assertEqual(runtime.startupError, reason) + self.assertFalse(runtime.ready) + finally: + runtime.dispose() + + def testUnavailableBindingFailsVisiblyWithoutFallback(self): + runtime = _ChildRuntime({'missing': True}) + try: + runtime.start() + self.assertTrue(waitFor(lambda: bool(runtime.startupError), timeout=10)) + self.assertEqual( + runtime.startupError, 'sing-tun package is unavailable or incompatible' + ) + self.assertFalse(runtime.ready) + finally: + runtime.dispose() + + def testFailureAfterNativeReadinessKeepsRuntimeExitMeaning(self): + runtime = _ChildRuntime({'runtime_fail': True}) + try: + runtime.start() + self.assertTrue(waitFor(lambda: bool(runtime.startupError), timeout=10)) + event = runtime.interpretExit(1) + self.assertEqual(event.reason, RuntimeExitReason.Unexpected) + self.assertEqual( + event.message, 'sing-tun: new binding runtime failure wording' + ) + finally: + runtime.dispose() + + def testHostCleanupRefusalKeepsRuntimeRetryable(self): + plan = mock.Mock() + plan.finishCleanup.side_effect = [RuntimeError('retained'), None, None] + runtime = SingTUN({}, hostPlan=plan) + try: + with self.assertRaisesRegex(RuntimeError, 'retained'): + runtime.stop() + + self.assertFalse(runtime._control.closed) + + runtime.stop() + finally: + runtime.dispose() + + def testFailedHandleCloseStillRestoresHostAndRetainsRetryOwnership(self): + runtime = _ChildRuntime({}) + plan = mock.Mock() + runtime._hostPlan = plan + try: + runtime.start() + self.assertTrue(waitFor(lambda: runtime.ready, timeout=10)) + + child = runtime.process + with mock.patch.object( + child, 'close', side_effect=OSError('retained handle') + ): + with self.assertRaisesRegex(RuntimeError, 'handle could not be closed'): + runtime.stop() + + plan.finishCleanup.assert_called_once() + self.assertIs(runtime.process, child) + self.assertFalse(child.is_alive()) + self.assertFalse(runtime._control.closed) + + runtime.stop() + self.assertTrue(child._closed) + finally: + runtime.dispose() + + def testFailedReapRestoresDNSWithoutRecoveringLiveNativeResources(self): + runtime = SingTUN({}) + plan, child = mock.Mock(), mock.Mock() + runtime._hostPlan, runtime._process = plan, child + child.is_alive.return_value = True + try: + with mock.patch.object( + MultiprocessingRuntime, + 'stop', + side_effect=RuntimeError('retained child'), + ): + with self.assertRaisesRegex(RuntimeError, 'retained child'): + runtime.stop() + + plan.finishDNSCleanup.assert_called_once() + plan.finishCleanup.assert_not_called() + self.assertIs(runtime.process, child) + self.assertFalse(runtime._control.closed) + finally: + runtime._process = None + runtime.dispose() + + +class SingTUNHostTest(unittest.TestCase): + def testLinuxResolverPrerequisiteFailsBeforeNativeActivation(self): + module = importlib.import_module('Furious.Service.SingTUNHost') + plan = SingTUNHostPlan(prepareSingTUNSettings({}), platform='Linux') + with mock.patch.object( + module.os, 'geteuid', return_value=0, create=True + ), mock.patch.object( + module.SystemRuntime, 'flatpakID', return_value='' + ), mock.patch.object( + module.shutil, 'which', return_value=None + ), mock.patch.object( + module, '_command' + ) as command: + with self.assertRaisesRegex(RuntimeError, 'systemd-resolved'): + plan.prepare([]) + command.assert_not_called() + self.assertFalse(plan._activated) + + def testUnsupportedOwnershipCombinationsFailBeforeHostCommands(self): + with self.assertRaisesRegex(ValueError, 'both address families'): + prepareSingTUNSettings({'tun_options': {'StrictRoute': True}}) + with self.assertRaisesRegex(ValueError, 'scoped route cleanup'): + prepareSingTUNSettings( + {'tun_options': {'InterfaceScope': True}}, platform='Darwin' + ) + with self.assertRaisesRegex(ValueError, 'Unsupported sing-tun host'): + SingTUNHostPlan( + {'host_options': {'defaultPrimaryGatewayIP': '192.0.2.254'}} + ) + + def testLinuxRecoveryMatchesSelectorsRatherThanDeletingPriorityAlone(self): + rule = { + 'priority': 2002, + 'src': 'all', + 'table': 'main', + 'not': True, + 'dport': '53-53', + 'suppress_prefixlength': 0, + } + command = _linuxRuleDelete('-4', rule, 123456, 2000, 'utun101') + self.assertIn('not', command) + self.assertIn('dport', command) + self.assertIn('suppress_prefixlength', command) + self.assertIn('main', command) + with self.assertRaisesRegex(RuntimeError, 'Conflicting'): + _linuxRuleDelete('-4', {**rule, 'dport': 443}, 123456, 2000, 'utun101') + + def testDevelopmentIPv6RuleOrderingKeepsExactCleanupSelectors(self): + # The snapshot moves the TUN goto before the loopback source rules. + rule = { + 'priority': 2001, + 'src': 'all', + 'iif': 'utun101', + 'goto': 2010, + } + self.assertEqual( + _linuxRuleDelete('-6', rule, 123456, 2000, 'utun101'), + [ + 'ip', + '-6', + 'rule', + 'del', + 'priority', + '2001', + 'from', + 'all', + 'iif', + 'utun101', + 'goto', + '2010', + ], + ) + sourceRule = { + 'priority': 2002, + 'src': 'fd00::/64', + 'iif': 'lo', + 'table': 123456, + } + command = _linuxRuleDelete('-6', sourceRule, 123456, 2000, 'utun101') + self.assertIn('fd00::/64', command) + self.assertEqual(command[-2:], ['table', '123456']) + + def testWindowsRegistersStaticDNSRestorationBeforeFailedWrite(self): + module = importlib.import_module('Furious.Service.SingTUNHost') + plan = SingTUNHostPlan(prepareSingTUNSettings({}), platform='Windows') + plan.deviceName = 'utun101' + plan.egressIP = '192.0.2.10' + snapshot = [ + { + 'Name': 'Ethernet', + 'IP': ['192.0.2.10'], + 'Static': '8.8.8.8', + 'DNS': ['2606:4700:4700::1111', '8.8.8.8'], + } + ] + with mock.patch.object( + module, '_powershell', return_value=json.dumps(snapshot) + ), mock.patch.object( + plan, '_windowsDNS', side_effect=RuntimeError('write failed') + ): + with self.assertRaisesRegex(RuntimeError, 'write failed'): + plan.applyDNS('utun101') + + self.assertEqual(plan._dnsRestore[0][1]['DNS'], ['8.8.8.8']) + + with mock.patch.object(plan, '_windowsDNS') as restore: + plan.cleanup(True) + restore.assert_called_once_with('Ethernet', ['8.8.8.8']) + + def testDarwinAbbreviatedRouteDestinationsAreNormalized(self): + rows = _darwinRoutes( + 'Destination Gateway Flags Netif\n1 198.18.0.1 UGSc utun101\n192.0.2 192.0.2.254 UGSc en0', + 4, + ) + + self.assertEqual( + rows[0], (ipaddress.ip_network('1.0.0.0/8'), '198.18.0.1', 'utun101') + ) + self.assertEqual(rows[1][0], ipaddress.ip_network('192.0.2.0/24')) + + def testLinuxPrivilegesFailBeforeCommands(self): + module = importlib.import_module('Furious.Service.SingTUNHost') + plan = SingTUNHostPlan(prepareSingTUNSettings({}), platform='Linux') + with mock.patch.object( + module.os, 'geteuid', return_value=1000, create=True + ), mock.patch.object( + module.SystemRuntime, 'flatpakID', return_value='' + ), mock.patch.object( + module, '_command' + ) as commands: + with self.assertRaisesRegex(RuntimeError, 'requires root'): + plan.prepare(['192.0.2.1']) + commands.assert_not_called() + + def testLinuxAssignedIdentifiersAndDualStackExclusions(self): + module = importlib.import_module('Furious.Service.SingTUNHost') + plan = SingTUNHostPlan(prepareSingTUNSettings({}), platform='Linux') + with mock.patch.object( + module.os, 'geteuid', return_value=0, create=True + ), mock.patch.object( + module.SystemRuntime, 'flatpakID', return_value='' + ), mock.patch.object( + module.shutil, 'which', return_value='resolvectl' + ), mock.patch.object( + module.socket, 'if_nameindex', return_value=[] + ), mock.patch.object( + module, '_command', return_value='[]' + ): + plan.prepare(['192.0.2.1', '2001:db8::1']) + + tun = plan.configuration['tun_options'] + self.assertGreater(tun['IPRoute2TableIndex'], 100000) + self.assertGreater(tun['IPRoute2RuleIndex'], 999) + self.assertEqual(tun['Inet6RouteExcludeAddress'], ['2001:db8::1/128']) + + def testDNSRestorationIndependentOfNativeRecoveryFailure(self): + module = importlib.import_module('Furious.Service.SingTUNHost') + plan = SingTUNHostPlan(prepareSingTUNSettings({}), platform='Windows') + snapshot = { + 'Name': 'Ethernet', + 'Static': '8.8.8.8', + 'DNS': ['8.8.8.8', '8.8.4.4'], + } + plan._dnsRestore = [('Windows', snapshot)] + plan._activated = True + with mock.patch.object(plan, '_windowsDNS') as restore, mock.patch.object( + plan, '_recoverNative', side_effect=RuntimeError('retained') + ): + with self.assertRaisesRegex(RuntimeError, 'retained'): + plan.cleanup(False) + + restore.assert_called_once_with('Ethernet', ['8.8.8.8', '8.8.4.4']) + self.assertEqual(plan._dnsRestore, []) + self.assertTrue(plan._activated) + + def testRecoveryRefusesConflictingPolicyRules(self): + module = importlib.import_module('Furious.Service.SingTUNHost') + plan = SingTUNHostPlan(prepareSingTUNSettings({}), platform='Linux') + plan._table, plan._ruleStart = 123456, 2000 + with mock.patch.object( + module.socket, 'if_nameindex', return_value=[] + ), mock.patch.object( + module, '_command', return_value='[{"priority":2000,"table":987}]' + ) as commands: + with self.assertRaisesRegex(RuntimeError, 'Conflicting'): + plan._recoverNative() + self.assertEqual(commands.call_count, 1) + + def testDarwinRangesMirrorSubRangesAndExclusions(self): + tun = prepareSingTUNSettings({})['tun_options'] + ranges = _routeRanges(tun) + self.assertEqual(len(ranges), 8) + self.assertNotIn(ipaddress.ip_address('0.0.0.1'), ranges[0]) + + tun['Inet4RouteExcludeAddress'] = ['192.0.2.1/32'] + self.assertFalse( + any( + ipaddress.ip_address('192.0.2.1') in network + for network in _routeRanges(tun) + ) + ) + + +class SingTUNStartupTest(unittest.TestCase): + @classmethod + def setUpClass(cls): + application() + + def testBothStartupPathsKeepCapturedTUNLogSourceAfterPreferenceChanges(self): + """Changing the next engine cannot relabel a previously captured attempt.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + for backend in ('sing-tun', 'tun2socks'): + other = 'tun2socks' if backend == 'sing-tun' else 'sing-tun' + for asynchronous in (True, False): + with self.subTest(backend=backend, asynchronous=asynchronous): + with isolatedSettings(): + AppSettings.set('ApplicationTUNBackend', backend) + manager = ConnectionManager() + logs = LogManager() + route = logs.callback(TUN_LOG_CATEGORY) + + def switchPreference(*_args, **_kwargs): + AppSettings.set('ApplicationTUNBackend', other) + return True + + def emitTUN(_attempt, _exitCallback, callback): + callback.appendMany(('first', 'second')) + return True + + try: + if asynchronous: + operation = manager.startAsync( + _Configuration(), + 'Global', + msgCallbackTUN_=route, + deepcopy=False, + ) + switchPreference() + operation.msgCallbackTUN_.appendMany( + ('first', 'second') + ) + else: + starter = ( + '_startApplicationSingTUN' + if backend == 'sing-tun' + else '_startApplicationTun2socks' + ) + with mock.patch.object( + manager, + '_prepareTUNPolicy', + return_value=(False, True), + ), mock.patch.object( + module.SystemRuntime, 'isTUNMode', return_value=True + ), mock.patch.object( + Storage, 'UserSingTUNSettings', return_value={} + ), mock.patch.object( + Storage, 'UserTUNSettings', return_value={} + ), mock.patch.object( + manager, + '_startPrimaryRuntime', + side_effect=switchPreference, + ), mock.patch.object( + manager, starter, side_effect=emitTUN + ): + self.assertTrue( + manager.start( + _Configuration(), + 'Global', + msgCallbackTUN_=route, + deepcopy=False, + ) + ) + + self.assertEqual( + AppSettings.get('ApplicationTUNBackend'), other + ) + self.assertEqual(route.source, '') + self.assertEqual( + [ + entry.source + for entry in logs.entries(TUN_LOG_CATEGORY) + ], + [backend, backend], + ) + finally: + manager.cleanup() + logs.deleteLater() + processQtEvents() + + def testSnapshotReadsOnlyTheSelectedBackendAndKeepsItsOwnCopy(self): + for backend, getter, field in ( + ('sing-tun', 'UserSingTUNSettings', 'singSettings'), + ('tun2socks', 'UserTUNSettings', 'tun2socksSettings'), + ): + with self.subTest(backend=backend): + stored = ( + {'host_options': {'tunAdapterInterfaceDNS': '9.9.9.9'}} + if backend == 'sing-tun' + else {'tunAdapterInterfaceDNS': '8.8.8.8'} + ) + inactive = ( + 'UserTUNSettings' + if backend == 'sing-tun' + else 'UserSingTUNSettings' + ) + attempt = _ConnectionStartAttempt( + mock.Mock(), + _Configuration(), + applicationTun2socks=True, + tunBackend=backend, + ) + + with mock.patch.object( + Storage, getter, return_value=stored + ) as selected, mock.patch.object( + Storage, + inactive, + side_effect=AssertionError('inactive backend read'), + ) as other: + attempt.snapshotApplicationTUN() + + selected.assert_called_once_with() + other.assert_not_called() + expected = copy.deepcopy(stored) + stored.clear() + self.assertEqual(getattr(attempt, field), expected) + if backend == 'sing-tun': + callers = attempt.singHostSettingsCallers + self.assertIsNone(attempt.tun2socksSettingsCallers) + self.assertEqual(callers.userTunAdapterInterfaceDNS(), '9.9.9.9') + self.assertIs(callers.userDisablePrimaryAdapterInterfaceDNS(), True) + else: + callers = attempt.tun2socksSettingsCallers + self.assertIsNone(attempt.singHostSettingsCallers) + self.assertEqual(callers.userTunAdapterInterfaceDNS(), '8.8.8.8') + self.assertEqual( + callers.userDisablePrimaryAdapterInterfaceDNS(), 'True' + ) + self.assertEqual(callers.userTcpSendBufferSize(), 1) + + self.assertEqual( + getattr( + attempt, + ( + 'tun2socksSettings' + if backend == 'sing-tun' + else 'singSettings' + ), + ), + {}, + ) + + def testBothStartupPathsUseNamedCallersBoundBeforePrimaryLaunch(self): + """Settings edited during core launch only affect the next attempt.""" + module = importlib.import_module('Furious.Service.ConnectionManager') + for backend in ('tun2socks', 'sing-tun'): + for asynchronous in (True, False): + with self.subTest(backend=backend, asynchronous=asynchronous): + with isolatedSettings(): + AppSettings.set('ApplicationTUNBackend', backend) + host = {'tunAdapterInterfaceDNS': '9.9.9.9'} + stored = ( + {'host_options': host} if backend == 'sing-tun' else host + ) + getter = ( + 'UserSingTUNSettings' + if backend == 'sing-tun' + else 'UserTUNSettings' + ) + inactive = ( + 'UserTUNSettings' + if backend == 'sing-tun' + else 'UserSingTUNSettings' + ) + manager = ConnectionManager() + primary = _Runtime() + observed = [] + + def changeSettings(*_args, **_kwargs): + host['tunAdapterInterfaceDNS'] = '1.0.0.1' + return True + + def observeSettings(attempt): + callers = ( + attempt.singHostSettingsCallers + if backend == 'sing-tun' + else attempt.tun2socksSettingsCallers + ) + observed.append(callers.userTunAdapterInterfaceDNS()) + + def finishAsync(operation): + observeSettings(operation.attempt) + operation._commit() + + def finishSync(attempt, *_args): + observeSettings(attempt) + return True + + try: + with mock.patch.object( + Storage, getter, return_value=stored + ) as selected, mock.patch.object( + Storage, + inactive, + side_effect=AssertionError('inactive backend read'), + ), mock.patch.object( + manager, '_prepareTUNPolicy', return_value=(False, True) + ), mock.patch.object( + module.SystemRuntime, 'isTUNMode', return_value=True + ): + if asynchronous: + startPrimary = primary.start + + def launchPrimary(): + changeSettings() + startPrimary() + + with mock.patch.object( + module, + 'getPluginRegistry', + return_value=_Registry( + [PreparedRuntime(primary)] + ), + ), mock.patch.object( + primary, 'start', side_effect=launchPrimary + ), mock.patch.object( + module.ConnectionStartOperation, + '_beginApplicationTun', + autospec=True, + side_effect=finishAsync, + ): + operation = manager.startAsync( + _Configuration(), 'Global', deepcopy=False + ) + self.assertTrue( + waitFor(lambda: operation._terminal) + ) + self.assertTrue(operation.attempt.committed) + else: + starter = ( + '_startApplicationSingTUN' + if backend == 'sing-tun' + else '_startApplicationTun2socks' + ) + with mock.patch.object( + manager, + '_startPrimaryRuntime', + side_effect=changeSettings, + ), mock.patch.object( + manager, starter, side_effect=finishSync + ): + self.assertTrue( + manager.start( + _Configuration(), + 'Global', + deepcopy=False, + ) + ) + + self.assertEqual(observed, ['9.9.9.9']) + self.assertEqual( + host['tunAdapterInterfaceDNS'], '1.0.0.1' + ) + nextAttempt = _ConnectionStartAttempt( + manager, + _Configuration(), + applicationTun2socks=True, + tunBackend=backend, + ) + nextAttempt.snapshotApplicationTUN() + observeSettings(nextAttempt) + self.assertEqual(observed, ['9.9.9.9', '1.0.0.1']) + self.assertEqual(selected.call_count, 2) + finally: + manager.cleanup() + primary.dispose() + processQtEvents() + + def testBothPreferencesPreserveNativePriorityAndTypedFailure(self): + module = importlib.import_module('Furious.Service.ConnectionManager') + for backend in ('sing-tun', 'tun2socks'): + for failure in (False, True): + with self.subTest(backend=backend, failure=failure), isolatedSettings(): + AppSettings.set('ApplicationTUNBackend', backend) + manager, primary = ConnectionManager(), _Runtime() + registry = _Registry([PreparedRuntime(primary)]) + registry.prepareTUN = mock.Mock( + return_value=True, + side_effect=( + TUNPreparationError('native failed') if failure else None + ), + ) + registry.usesApplicationTun2socks = mock.Mock() + with mock.patch.object( + module.SystemRuntime, 'isTUNMode', return_value=True + ), mock.patch.object( + module, 'getPluginRegistry', return_value=registry + ), mock.patch.object( + module, 'SingTUN' + ) as sing, mock.patch.object( + module, 'Tun2socks' + ) as tun2socks, mock.patch.object( + Storage, 'UserSingTUNSettings' + ) as stored: + operation = manager.startAsync( + _Configuration(), 'Global', deepcopy=False + ) + self.assertTrue(waitFor(lambda: operation._terminal)) + + sing.assert_not_called() + tun2socks.assert_not_called() + stored.assert_not_called() + registry.usesApplicationTun2socks.assert_not_called() + self.assertEqual(primary.isRunning(), not failure) + + manager.cleanup() + processQtEvents() + + def testInvalidSettingsReleaseDetachedRouterAndPrimaryRuntime(self): + module = importlib.import_module('Furious.Service.ConnectionManager') + primaryRouter, router = RuntimeEventRouter(), RuntimeEventRouter() + primary = _Runtime() + manager = ConnectionManager() + with isolatedSettings(), mock.patch.object( + Storage, + 'UserTUNSettings', + side_effect=AssertionError('inactive tun2socks backend'), + ), mock.patch.object( + Storage, 'UserSingTUNSettings', return_value={'stack': 'go'} + ), mock.patch.object( + manager, '_prepareTUNPolicy', return_value=(False, True) + ), mock.patch.object( + module, + 'getPluginRegistry', + return_value=_Registry([PreparedRuntime(primary)]), + ): + operation = manager.startAsync(_Configuration(), 'Global', deepcopy=False) + with mock.patch.object( + module, 'RuntimeEventRouter', side_effect=[primaryRouter, router] + ): + self.assertTrue(waitFor(lambda: operation._terminal)) + self.assertEqual(router.state, RuntimeLeaseState.Released) + self.assertFalse(primary.isRunning()) + self.assertEqual(manager.runtimes, []) + + manager.cleanup() + processQtEvents() + self.assertFalse(isValid(router)) + + def testAliveDoesNotCommitUntilAuthoritativeNativeReadyAndDNSComplete(self): + module = importlib.import_module('Furious.Service.ConnectionManager') + manager = ConnectionManager() + primary = _Runtime() + registry = _Registry([PreparedRuntime(primary)]) + customized = { + 'tun_options': {'MTU': 1400}, + 'host_options': {'bypassTUNAdapterInterfaceIP': '2001:db8::1'}, + } + runtimes = [] + + def create(*args, **kwargs): + runtime = _PreparedSing(*args, **kwargs) + runtimes.append(runtime) + return runtime + + with isolatedSettings(), mock.patch.object( + Storage, + 'UserTUNSettings', + side_effect=AssertionError('inactive tun2socks backend'), + ), mock.patch.object( + Storage, 'UserSingTUNSettings', return_value=customized + ), mock.patch.object( + manager, '_prepareTUNPolicy', return_value=(False, True) + ), mock.patch.object( + module, 'getPluginRegistry', return_value=registry + ), mock.patch.object( + module, 'SingTUN', _PreparedSing + ), mock.patch.object( + module, 'SingTUNHostPlan', _Plan + ), mock.patch.object( + manager, '_createSingTUN', wraps=manager._createSingTUN + ): + succeeded = [] + operation = manager.startAsync(_Configuration(), 'Global', deepcopy=False) + operation.succeeded.connect(succeeded.append) + self.assertTrue( + waitFor(lambda: operation._tun is not None and operation._tun.alive) + ) + runtime = operation._tun + self.assertFalse(succeeded) + self.assertEqual(manager.runtimes, []) + + AppSettings.set('ApplicationTUNBackend', 'tun2socks') + customized['host_options']['bypassTUNAdapterInterfaceIP'] = '192.0.2.99' + customized['tun_options']['MTU'] = 1300 + self.assertEqual(operation.attempt.tunBackend, 'sing-tun') + self.assertEqual(runtime._configuration['tun_options']['MTU'], 1400) + + runtime.nativeReady = True + self.assertTrue(waitFor(lambda: bool(succeeded))) + self.assertEqual(runtime._hostPlan.applied, ['utun101']) + self.assertEqual(runtime._hostPlan.addresses, ['192.0.2.1', '2001:db8::1']) + + manager.cleanup() + processQtEvents() + + def testCancellationBeforeNativeReadyReleasesBothOwners(self): + module = importlib.import_module('Furious.Service.ConnectionManager') + manager = ConnectionManager() + primary = _Runtime() + with isolatedSettings(), mock.patch.object( + Storage, + 'UserTUNSettings', + side_effect=AssertionError('inactive tun2socks backend'), + ), mock.patch.object( + Storage, 'UserSingTUNSettings', return_value={} + ), mock.patch.object( + manager, '_prepareTUNPolicy', return_value=(False, True) + ), mock.patch.object( + module, + 'getPluginRegistry', + return_value=_Registry([PreparedRuntime(primary)]), + ), mock.patch.object( + module, 'SingTUN', _PreparedSing + ), mock.patch.object( + module, 'SingTUNHostPlan', _Plan + ): + operation = manager.startAsync(_Configuration(), 'Global', deepcopy=False) + self.assertTrue( + waitFor(lambda: operation._tun is not None and operation._tun.alive) + ) + runtime = operation._tun + + operation.cancel() + + self.assertTrue(runtime.released) + self.assertFalse(primary.alive) + self.assertEqual(manager.runtimes, []) + + manager.cleanup() + processQtEvents() + + def testSynchronousCompatibilityPathSelectsSingAndCommitsAfterHostDNS(self): + module = importlib.import_module('Furious.Service.ConnectionManager') + manager = ConnectionManager() + primary = _Runtime() + primary.start() + with isolatedSettings(), mock.patch.object( + Storage, + 'UserTUNSettings', + side_effect=AssertionError('inactive tun2socks backend'), + ), mock.patch.object( + Storage, 'UserSingTUNSettings', return_value={} + ), mock.patch.object( + manager, '_prepareTUNPolicy', return_value=(False, True) + ), mock.patch.object( + manager, '_startCoreRuntime', return_value=(primary, True) + ), mock.patch.object( + module.SystemRuntime, 'isTUNMode', return_value=True + ), mock.patch.object( + module, 'SingTUN', _PreparedSing + ), mock.patch.object( + module, 'SingTUNHostPlan', _Plan + ), mock.patch.object( + _PreparedSing, 'ready', new_callable=mock.PropertyMock, return_value=True + ): + self.assertTrue(manager.start(_Configuration(), 'Global', deepcopy=False)) + self.assertEqual(manager.runtimes[1]._hostPlan.applied, ['utun101']) + + manager.cleanup() + + def testSocksOnlyResolverHasNoHTTPOrDirectFallback(self): + resolver = mock.Mock() + configuration = _Configuration() + configuration.httpProxy = lambda: '' + + ConnectionManager._configureTunResolver(resolver, configuration) + + resolver.configureSocksProxy.assert_called_once_with('socks5://127.0.0.1:18081') + resolver.configureHttpProxy.assert_not_called() + + +if __name__ == '__main__': + unittest.main() diff --git a/tests/test_ui_behavior.py b/tests/test_ui_behavior.py index 0393c254..5a625e11 100644 --- a/tests/test_ui_behavior.py +++ b/tests/test_ui_behavior.py @@ -426,6 +426,7 @@ class SettingsPageOrganizationTest(unittest.TestCase): ) page = SettingsPage( tunSettingsDialogFactory=mock.Mock(), + singTunSettingsDialogFactory=mock.Mock(), proxyBypassDialog=proxyBypassDialog, networkTestDialog=networkTestDialog, **callbacks, @@ -449,6 +450,20 @@ class SettingsPageOrganizationTest(unittest.TestCase): ] self.assertEqual(page.generalSection.cards, expectedGeneralCards) + expectedTUNCards = [ + page.tunBackendCard, + page.singTunSettingsCard, + page.tunSettingsCard, + ] + self.assertEqual(page.tunSection.titleLabel.text(), 'TUN') + self.assertEqual(page.tunSection.cards, expectedTUNCards) + self.assertFalse(page.tunSection.isHidden()) + self.assertTrue( + all(card.parent() is page.tunSection for card in expectedTUNCards) + ) + self.assertTrue( + all(card not in page.connectionSection.cards for card in expectedTUNCards) + ) self.assertEqual( page.applicationSection.cards, [page.updateCard, page.aboutCard], @@ -473,11 +488,17 @@ class SettingsPageOrganizationTest(unittest.TestCase): page.restartCard.button.click() page.openFolderCard.button.click() + page.tunSettingsCard.button.click() + page.singTunSettingsCard.button.click() callbacks['restartAsAdmin'].assert_called_once_with() callbacks['openApplicationFolder'].assert_called_once_with() callbacks['checkForUpdates'].assert_not_called() callbacks['openAboutPage'].assert_not_called() + page._tunSettingsDialogFactory.assert_called_once_with(parent=page) + page._tunSettingsDialogFactory.return_value.open.assert_called_once_with() + page._singTunSettingsDialogFactory.assert_called_once_with(parent=page) + page._singTunSettingsDialogFactory.return_value.open.assert_called_once_with() page.close() page.deleteLater() @@ -494,6 +515,13 @@ class SettingsPageOrganizationTest(unittest.TestCase): self.assertIn(page.tunModeCard, page.generalSection.cards) self.assertFalse(page.tunModeCard.checkBox.isEnabled()) + self.assertTrue(page.tunSection.isHidden()) + self.assertTrue( + all( + card.parent() is page.tunSection + for card in page.tunSection.cards + ) + ) page.setConnectionControlsEnabled(True) self.assertFalse(page.tunModeCard.checkBox.isEnabled())