platform: modularize api/gui, add docs-tests-web foundation, and refresh root config

This commit is contained in:
beckline
2026-03-26 22:40:54 +03:00
parent 0e2d7f61ea
commit 6a56d734c2
562 changed files with 70151 additions and 16423 deletions

View File

@@ -0,0 +1,11 @@
from .cards_mixin import SingBoxCardsMixin
from .editor_mixin import SingBoxEditorMixin
from .links_mixin import SingBoxLinksMixin
from .runtime_mixin import SingBoxRuntimeMixin
__all__ = [
"SingBoxCardsMixin",
"SingBoxEditorMixin",
"SingBoxLinksMixin",
"SingBoxRuntimeMixin",
]

View File

@@ -0,0 +1,205 @@
from __future__ import annotations
from PySide6.QtCore import QSize, Qt
from PySide6.QtWidgets import QFrame, QLabel, QListWidgetItem, QVBoxLayout
from main_window.constants import SINGBOX_STATUS_ROLE
from transport_protocol_summary import transport_protocol_summary
class SingBoxCardsMixin:
def _singbox_client_protocol_summary(self, client) -> str:
protocol_txt = transport_protocol_summary(client)
if protocol_txt == "n/a":
cid = str(getattr(client, "id", "") or "").strip()
if (
cid
and cid == str(self._singbox_editor_profile_client_id or "").strip()
and str(self._singbox_editor_protocol or "").strip()
):
protocol_txt = str(self._singbox_editor_protocol).strip().lower()
return protocol_txt
def _make_singbox_profile_card_widget(
self,
*,
name: str,
protocol_txt: str,
status: str,
latency_txt: str,
cid: str,
) -> QFrame:
frame = QFrame()
frame.setObjectName("singboxProfileCard")
lay = QVBoxLayout(frame)
lay.setContentsMargins(10, 8, 10, 8)
lay.setSpacing(2)
lbl_name = QLabel(name)
lbl_name.setObjectName("cardName")
lbl_name.setAlignment(Qt.AlignHCenter)
lay.addWidget(lbl_name)
lbl_proto = QLabel(protocol_txt)
lbl_proto.setObjectName("cardProto")
lbl_proto.setAlignment(Qt.AlignHCenter)
lay.addWidget(lbl_proto)
lbl_state = QLabel(f"{str(status or '').upper()} · {latency_txt}")
lbl_state.setObjectName("cardState")
lbl_state.setAlignment(Qt.AlignHCenter)
lay.addWidget(lbl_state)
frame.setToolTip(f"{cid}\n{protocol_txt}\nstatus={status}")
return frame
def _style_singbox_profile_card_widget(
self,
card: QFrame,
*,
active: bool,
selected: bool,
) -> None:
if active and selected:
bg = "#c7f1d5"
border = "#208f47"
name_color = "#11552e"
meta_color = "#1f6f43"
elif active:
bg = "#eafaf0"
border = "#2f9e44"
name_color = "#14532d"
meta_color = "#1f6f43"
elif selected:
bg = "#e8f1ff"
border = "#2f80ed"
name_color = "#1b2f50"
meta_color = "#28568a"
else:
bg = "#f7f7f7"
border = "#c9c9c9"
name_color = "#202020"
meta_color = "#666666"
card.setStyleSheet(
f"""
QFrame#singboxProfileCard {{
border: 1px solid {border};
border-radius: 6px;
background: {bg};
}}
QLabel#cardName {{
color: {name_color};
font-weight: 600;
}}
QLabel#cardProto {{
color: {meta_color};
}}
QLabel#cardState {{
color: {meta_color};
}}
"""
)
def _refresh_singbox_profile_card_styles(self) -> None:
current_id = self._selected_transport_engine_id()
for i in range(self.lst_singbox_profile_cards.count()):
item = self.lst_singbox_profile_cards.item(i)
cid = str(item.data(Qt.UserRole) or "").strip()
status = str(item.data(SINGBOX_STATUS_ROLE) or "").strip().lower()
card = self.lst_singbox_profile_cards.itemWidget(item)
if not isinstance(card, QFrame):
continue
self._style_singbox_profile_card_widget(
card,
active=(status == "up"),
selected=bool(current_id and cid == current_id),
)
def _render_singbox_profile_cards(self) -> None:
current_id = self._selected_transport_engine_id()
self.lst_singbox_profile_cards.blockSignals(True)
self.lst_singbox_profile_cards.clear()
selected_item = None
if not self._transport_api_supported:
item = QListWidgetItem("Transport API unavailable")
item.setFlags(item.flags() & ~Qt.ItemIsEnabled & ~Qt.ItemIsSelectable)
self.lst_singbox_profile_cards.addItem(item)
self.lst_singbox_profile_cards.blockSignals(False)
return
if not self._transport_clients:
item = QListWidgetItem("No SingBox profiles configured")
item.setFlags(item.flags() & ~Qt.ItemIsEnabled & ~Qt.ItemIsSelectable)
self.lst_singbox_profile_cards.addItem(item)
self.lst_singbox_profile_cards.blockSignals(False)
return
for c in self._transport_clients:
cid = str(getattr(c, "id", "") or "").strip()
if not cid:
continue
name = str(getattr(c, "name", "") or "").strip() or cid
status, latency, _last_error, _last_check = self._transport_live_health_for_client(c)
latency_txt = f"{latency}ms" if latency > 0 else "no ping"
protocol_txt = self._singbox_client_protocol_summary(c)
item = QListWidgetItem("")
item.setData(Qt.UserRole, cid)
item.setData(SINGBOX_STATUS_ROLE, status)
item.setSizeHint(QSize(228, 78))
self.lst_singbox_profile_cards.addItem(item)
self.lst_singbox_profile_cards.setItemWidget(
item,
self._make_singbox_profile_card_widget(
name=name,
protocol_txt=protocol_txt,
status=status,
latency_txt=latency_txt,
cid=cid,
),
)
if current_id and cid == current_id:
selected_item = item
if selected_item is not None:
self.lst_singbox_profile_cards.setCurrentItem(selected_item)
elif self.lst_singbox_profile_cards.count() > 0:
self.lst_singbox_profile_cards.setCurrentRow(0)
self.lst_singbox_profile_cards.blockSignals(False)
self._refresh_singbox_profile_card_styles()
def _sync_singbox_profile_card_selection(self, cid: str) -> None:
if self._syncing_singbox_selection:
return
self._syncing_singbox_selection = True
try:
self.lst_singbox_profile_cards.blockSignals(True)
self.lst_singbox_profile_cards.clearSelection()
target = str(cid or "").strip()
if target:
for i in range(self.lst_singbox_profile_cards.count()):
item = self.lst_singbox_profile_cards.item(i)
if str(item.data(Qt.UserRole) or "").strip() == target:
self.lst_singbox_profile_cards.setCurrentItem(item)
break
self.lst_singbox_profile_cards.blockSignals(False)
finally:
self._syncing_singbox_selection = False
self._refresh_singbox_profile_card_styles()
def _select_transport_engine_by_id(self, cid: str) -> bool:
target = str(cid or "").strip()
if not target:
return False
idx = self.cmb_transport_engine.findData(target)
if idx < 0:
return False
if idx != self.cmb_transport_engine.currentIndex():
self.cmb_transport_engine.setCurrentIndex(idx)
else:
self._sync_singbox_profile_card_selection(target)
self._sync_selected_singbox_profile_link(silent=True)
self._load_singbox_editor_for_selected(silent=True)
self._update_transport_engine_view()
return True

View File

@@ -0,0 +1,633 @@
from __future__ import annotations
import json
from typing import Any
from main_window.constants import SINGBOX_EDITOR_PROTOCOL_IDS, SINGBOX_PROTOCOL_SEED_SPEC
class SingBoxEditorMixin:
def _selected_singbox_profile_id(self) -> str:
selected = self._selected_transport_client()
if selected is not None:
selected_cid = str(getattr(selected, "id", "") or "").strip()
if (
selected_cid
and self._singbox_editor_profile_id
and selected_cid == str(self._singbox_editor_profile_client_id or "").strip()
):
return str(self._singbox_editor_profile_id).strip()
if selected_cid:
# Desktop SingBox tab keeps one deterministic profile per engine card.
return selected_cid
return self._selected_transport_engine_id()
def _set_singbox_editor_enabled(self, enabled: bool) -> None:
widgets = [
self.ent_singbox_proto_name,
self.chk_singbox_proto_enabled,
self.cmb_singbox_proto_protocol,
self.ent_singbox_vless_server,
self.spn_singbox_vless_port,
self.ent_singbox_vless_uuid,
self.ent_singbox_proto_password,
self.cmb_singbox_vless_flow,
self.cmb_singbox_vless_packet_encoding,
self.cmb_singbox_ss_method,
self.ent_singbox_ss_plugin,
self.spn_singbox_hy2_up_mbps,
self.spn_singbox_hy2_down_mbps,
self.ent_singbox_hy2_obfs,
self.ent_singbox_hy2_obfs_password,
self.cmb_singbox_tuic_congestion,
self.cmb_singbox_tuic_udp_mode,
self.chk_singbox_tuic_zero_rtt,
self.ent_singbox_wg_private_key,
self.ent_singbox_wg_peer_public_key,
self.ent_singbox_wg_psk,
self.ent_singbox_wg_local_address,
self.ent_singbox_wg_reserved,
self.spn_singbox_wg_mtu,
self.btn_singbox_wg_paste_private,
self.btn_singbox_wg_copy_private,
self.btn_singbox_wg_paste_peer,
self.btn_singbox_wg_copy_peer,
self.btn_singbox_wg_paste_psk,
self.btn_singbox_wg_copy_psk,
self.cmb_singbox_vless_transport,
self.ent_singbox_vless_path,
self.ent_singbox_vless_grpc_service,
self.cmb_singbox_vless_security,
self.ent_singbox_vless_sni,
self.ent_singbox_tls_alpn,
self.cmb_singbox_vless_utls_fp,
self.ent_singbox_vless_reality_pk,
self.ent_singbox_vless_reality_sid,
self.chk_singbox_vless_insecure,
self.chk_singbox_vless_sniff,
]
for w in widgets:
w.setEnabled(bool(enabled))
def _clear_singbox_editor(self) -> None:
self._singbox_editor_loading = True
try:
self._singbox_editor_profile_id = ""
self._singbox_editor_profile_client_id = ""
self._singbox_editor_protocol = "vless"
self._singbox_editor_source_raw = {}
self.ent_singbox_proto_name.setText("")
self.chk_singbox_proto_enabled.setChecked(True)
self.cmb_singbox_proto_protocol.setCurrentIndex(0)
self.ent_singbox_vless_server.setText("")
self.spn_singbox_vless_port.setValue(443)
self.ent_singbox_vless_uuid.setText("")
self.ent_singbox_proto_password.setText("")
self.cmb_singbox_vless_flow.setCurrentIndex(0)
self.cmb_singbox_vless_flow.setEditText("")
self.cmb_singbox_vless_packet_encoding.setCurrentIndex(0)
self.cmb_singbox_ss_method.setCurrentIndex(0)
self.ent_singbox_ss_plugin.setText("")
self.spn_singbox_hy2_up_mbps.setValue(0)
self.spn_singbox_hy2_down_mbps.setValue(0)
self.ent_singbox_hy2_obfs.setText("")
self.ent_singbox_hy2_obfs_password.setText("")
self.cmb_singbox_tuic_congestion.setCurrentIndex(0)
self.cmb_singbox_tuic_udp_mode.setCurrentIndex(0)
self.chk_singbox_tuic_zero_rtt.setChecked(False)
self.ent_singbox_wg_private_key.setText("")
self.ent_singbox_wg_peer_public_key.setText("")
self.ent_singbox_wg_psk.setText("")
self.ent_singbox_wg_local_address.setText("")
self.ent_singbox_wg_reserved.setText("")
self.spn_singbox_wg_mtu.setValue(0)
self.cmb_singbox_vless_transport.setCurrentIndex(0)
self.ent_singbox_vless_path.setText("")
self.ent_singbox_vless_grpc_service.setText("")
self.cmb_singbox_vless_security.setCurrentIndex(0)
self.ent_singbox_vless_sni.setText("")
self.ent_singbox_tls_alpn.setText("")
self.cmb_singbox_vless_utls_fp.setCurrentIndex(0)
self.ent_singbox_vless_reality_pk.setText("")
self.ent_singbox_vless_reality_sid.setText("")
self.chk_singbox_vless_insecure.setChecked(False)
self.chk_singbox_vless_sniff.setChecked(True)
finally:
self._singbox_editor_loading = False
self.on_singbox_vless_editor_changed()
def _load_singbox_editor_for_selected(self, *, silent: bool = True) -> None:
client = self._selected_transport_client()
if client is None:
self._clear_singbox_editor()
self._set_singbox_editor_enabled(False)
return
try:
cid = str(getattr(client, "id", "") or "").strip()
profile = self.ctrl.singbox_profile_get_for_client(
client,
profile_id=self._selected_singbox_profile_id(),
)
self._apply_singbox_editor_profile(profile, fallback_name=str(getattr(client, "name", "") or "").strip())
self._singbox_editor_profile_client_id = cid
self._set_singbox_editor_enabled(True)
except Exception as e:
if not silent:
raise
self._append_transport_log(f"[profile] editor load failed: {e}")
self._clear_singbox_editor()
self._set_singbox_editor_enabled(False)
def _find_editor_proxy_outbound(self, outbounds: list[Any]) -> dict[str, Any]:
proxy = {}
for row in outbounds:
if not isinstance(row, dict):
continue
t = str(row.get("type") or "").strip().lower()
tag = str(row.get("tag") or "").strip().lower()
if self._is_supported_editor_protocol(t):
proxy = row
break
if tag == "proxy":
proxy = row
return dict(proxy) if isinstance(proxy, dict) else {}
def _find_editor_sniff_inbound(self, inbounds: list[Any]) -> dict[str, Any]:
inbound = {}
for row in inbounds:
if not isinstance(row, dict):
continue
tag = str(row.get("tag") or "").strip().lower()
t = str(row.get("type") or "").strip().lower()
if tag == "socks-in" or t == "socks":
inbound = row
break
return dict(inbound) if isinstance(inbound, dict) else {}
def _apply_singbox_editor_profile(self, profile, *, fallback_name: str = "") -> None:
raw = getattr(profile, "raw_config", {}) or {}
if not isinstance(raw, dict):
raw = {}
protocol = str(getattr(profile, "protocol", "") or "").strip().lower() or "vless"
outbounds = raw.get("outbounds") or []
if not isinstance(outbounds, list):
outbounds = []
inbounds = raw.get("inbounds") or []
if not isinstance(inbounds, list):
inbounds = []
proxy = self._find_editor_proxy_outbound(outbounds)
inbound = self._find_editor_sniff_inbound(inbounds)
proxy_type = str(proxy.get("type") or "").strip().lower()
if self._is_supported_editor_protocol(proxy_type):
protocol = proxy_type
tls = proxy.get("tls") if isinstance(proxy.get("tls"), dict) else {}
reality = tls.get("reality") if isinstance(tls.get("reality"), dict) else {}
utls = tls.get("utls") if isinstance(tls.get("utls"), dict) else {}
transport = proxy.get("transport") if isinstance(proxy.get("transport"), dict) else {}
security = "none"
if bool(tls.get("enabled", False)):
security = "tls"
if bool(reality.get("enabled", False)):
security = "reality"
transport_type = str(transport.get("type") or "").strip().lower() or "tcp"
path = str(transport.get("path") or "").strip()
grpc_service = str(transport.get("service_name") or "").strip()
alpn_vals = tls.get("alpn") or []
if not isinstance(alpn_vals, list):
alpn_vals = []
alpn_text = ",".join([str(x).strip() for x in alpn_vals if str(x).strip()])
self._singbox_editor_loading = True
try:
self._singbox_editor_profile_id = str(getattr(profile, "id", "") or "").strip()
self._singbox_editor_protocol = protocol
self._singbox_editor_source_raw = json.loads(json.dumps(raw))
self.ent_singbox_proto_name.setText(
str(getattr(profile, "name", "") or "").strip() or fallback_name or self._singbox_editor_profile_id
)
self.chk_singbox_proto_enabled.setChecked(bool(getattr(profile, "enabled", True)))
pidx = self.cmb_singbox_proto_protocol.findData(protocol)
self.cmb_singbox_proto_protocol.setCurrentIndex(pidx if pidx >= 0 else 0)
self.ent_singbox_vless_server.setText(str(proxy.get("server") or "").strip())
try:
self.spn_singbox_vless_port.setValue(int(proxy.get("server_port") or 443))
except Exception:
self.spn_singbox_vless_port.setValue(443)
self.ent_singbox_vless_uuid.setText(str(proxy.get("uuid") or "").strip())
self.ent_singbox_proto_password.setText(str(proxy.get("password") or "").strip())
flow_value = str(proxy.get("flow") or "").strip()
idx = self.cmb_singbox_vless_flow.findData(flow_value)
if idx >= 0:
self.cmb_singbox_vless_flow.setCurrentIndex(idx)
else:
self.cmb_singbox_vless_flow.setEditText(flow_value)
pe = str(proxy.get("packet_encoding") or "").strip().lower()
if pe in ("none", "off", "false"):
pe = ""
idx = self.cmb_singbox_vless_packet_encoding.findData(pe)
self.cmb_singbox_vless_packet_encoding.setCurrentIndex(idx if idx >= 0 else 0)
ss_method = str(proxy.get("method") or "").strip().lower()
idx = self.cmb_singbox_ss_method.findData(ss_method)
if idx >= 0:
self.cmb_singbox_ss_method.setCurrentIndex(idx)
else:
self.cmb_singbox_ss_method.setEditText(ss_method)
self.ent_singbox_ss_plugin.setText(str(proxy.get("plugin") or "").strip())
try:
self.spn_singbox_hy2_up_mbps.setValue(int(proxy.get("up_mbps") or 0))
except Exception:
self.spn_singbox_hy2_up_mbps.setValue(0)
try:
self.spn_singbox_hy2_down_mbps.setValue(int(proxy.get("down_mbps") or 0))
except Exception:
self.spn_singbox_hy2_down_mbps.setValue(0)
obfs = proxy.get("obfs") if isinstance(proxy.get("obfs"), dict) else {}
self.ent_singbox_hy2_obfs.setText(str(obfs.get("type") or "").strip())
self.ent_singbox_hy2_obfs_password.setText(str(obfs.get("password") or "").strip())
cc = str(proxy.get("congestion_control") or "").strip()
idx = self.cmb_singbox_tuic_congestion.findData(cc)
if idx >= 0:
self.cmb_singbox_tuic_congestion.setCurrentIndex(idx)
else:
self.cmb_singbox_tuic_congestion.setCurrentIndex(0)
udp_mode = str(proxy.get("udp_relay_mode") or "").strip()
idx = self.cmb_singbox_tuic_udp_mode.findData(udp_mode)
self.cmb_singbox_tuic_udp_mode.setCurrentIndex(idx if idx >= 0 else 0)
self.chk_singbox_tuic_zero_rtt.setChecked(bool(proxy.get("zero_rtt_handshake", False)))
self.ent_singbox_wg_private_key.setText(str(proxy.get("private_key") or "").strip())
self.ent_singbox_wg_peer_public_key.setText(str(proxy.get("peer_public_key") or "").strip())
self.ent_singbox_wg_psk.setText(str(proxy.get("pre_shared_key") or "").strip())
local_addr = proxy.get("local_address") or []
if not isinstance(local_addr, list):
if str(local_addr or "").strip():
local_addr = [str(local_addr).strip()]
else:
local_addr = []
self.ent_singbox_wg_local_address.setText(
",".join([str(x).strip() for x in local_addr if str(x).strip()])
)
reserved = proxy.get("reserved") or []
if not isinstance(reserved, list):
if str(reserved or "").strip():
reserved = [str(reserved).strip()]
else:
reserved = []
self.ent_singbox_wg_reserved.setText(
",".join([str(x).strip() for x in reserved if str(x).strip()])
)
try:
self.spn_singbox_wg_mtu.setValue(int(proxy.get("mtu") or 0))
except Exception:
self.spn_singbox_wg_mtu.setValue(0)
idx = self.cmb_singbox_vless_transport.findData(transport_type)
self.cmb_singbox_vless_transport.setCurrentIndex(idx if idx >= 0 else 0)
self.ent_singbox_vless_path.setText(path)
self.ent_singbox_vless_grpc_service.setText(grpc_service)
idx = self.cmb_singbox_vless_security.findData(security)
self.cmb_singbox_vless_security.setCurrentIndex(idx if idx >= 0 else 0)
self.ent_singbox_vless_sni.setText(str(tls.get("server_name") or "").strip())
self.ent_singbox_tls_alpn.setText(alpn_text)
idx = self.cmb_singbox_vless_utls_fp.findData(str(utls.get("fingerprint") or "").strip())
self.cmb_singbox_vless_utls_fp.setCurrentIndex(idx if idx >= 0 else 0)
self.ent_singbox_vless_reality_pk.setText(str(reality.get("public_key") or "").strip())
self.ent_singbox_vless_reality_sid.setText(str(reality.get("short_id") or "").strip())
self.chk_singbox_vless_insecure.setChecked(bool(tls.get("insecure", False)))
self.chk_singbox_vless_sniff.setChecked(bool(inbound.get("sniff", True)))
finally:
self._singbox_editor_loading = False
self.on_singbox_vless_editor_changed()
def _validate_singbox_editor_form(self) -> None:
protocol = self._current_editor_protocol()
addr = self.ent_singbox_vless_server.text().strip()
if not addr:
raise RuntimeError("Address is required")
security = str(self.cmb_singbox_vless_security.currentData() or "none").strip().lower()
transport = str(self.cmb_singbox_vless_transport.currentData() or "tcp").strip().lower()
if protocol == "vless":
if not self.ent_singbox_vless_uuid.text().strip():
raise RuntimeError("UUID is required for VLESS")
if security == "reality" and not self.ent_singbox_vless_reality_pk.text().strip():
raise RuntimeError("Reality public key is required for Reality security mode")
elif protocol == "trojan":
if not self.ent_singbox_proto_password.text().strip():
raise RuntimeError("Password is required for Trojan")
if security == "reality":
raise RuntimeError("Reality security is not supported for Trojan in this editor")
elif protocol == "shadowsocks":
method = str(self.cmb_singbox_ss_method.currentData() or "").strip()
if not method:
method = self.cmb_singbox_ss_method.currentText().strip()
if not method:
raise RuntimeError("SS method is required for Shadowsocks")
if not self.ent_singbox_proto_password.text().strip():
raise RuntimeError("Password is required for Shadowsocks")
elif protocol == "hysteria2":
if not self.ent_singbox_proto_password.text().strip():
raise RuntimeError("Password is required for Hysteria2")
elif protocol == "tuic":
if not self.ent_singbox_vless_uuid.text().strip():
raise RuntimeError("UUID is required for TUIC")
if not self.ent_singbox_proto_password.text().strip():
raise RuntimeError("Password is required for TUIC")
elif protocol == "wireguard":
if not self.ent_singbox_wg_private_key.text().strip():
raise RuntimeError("WireGuard private key is required")
if not self.ent_singbox_wg_peer_public_key.text().strip():
raise RuntimeError("WireGuard peer public key is required")
local_addr = [str(x).strip() for x in self.ent_singbox_wg_local_address.text().split(",") if str(x).strip()]
if not local_addr:
raise RuntimeError("WireGuard local address is required (CIDR list)")
self._parse_wg_reserved_values(
[str(x).strip() for x in self.ent_singbox_wg_reserved.text().split(",") if str(x).strip()],
strict=True,
)
if protocol in ("vless", "trojan"):
if transport == "grpc" and not self.ent_singbox_vless_grpc_service.text().strip():
raise RuntimeError("gRPC service is required for gRPC transport")
if transport in ("ws", "http", "httpupgrade") and not self.ent_singbox_vless_path.text().strip():
raise RuntimeError("Transport path is required for selected transport")
def _build_singbox_editor_raw_config(self) -> dict[str, Any]:
base = self._singbox_editor_source_raw
if not isinstance(base, dict):
base = {}
raw: dict[str, Any] = json.loads(json.dumps(base))
protocol = self._current_editor_protocol()
outbounds = raw.get("outbounds") or []
if not isinstance(outbounds, list):
outbounds = []
proxy_idx = -1
for i, row in enumerate(outbounds):
if not isinstance(row, dict):
continue
t = str(row.get("type") or "").strip().lower()
tag = str(row.get("tag") or "").strip().lower()
if self._is_supported_editor_protocol(t) or tag == "proxy":
proxy_idx = i
break
proxy: dict[str, Any]
if proxy_idx >= 0:
proxy = dict(outbounds[proxy_idx]) if isinstance(outbounds[proxy_idx], dict) else {}
else:
proxy = {}
proxy["type"] = protocol
proxy["tag"] = str(proxy.get("tag") or "proxy")
proxy["server"] = self.ent_singbox_vless_server.text().strip()
proxy["server_port"] = int(self.spn_singbox_vless_port.value())
# clear protocol-specific keys before repopulating
for key in (
"uuid",
"password",
"method",
"plugin",
"flow",
"packet_encoding",
"up_mbps",
"down_mbps",
"obfs",
"congestion_control",
"udp_relay_mode",
"zero_rtt_handshake",
"private_key",
"peer_public_key",
"pre_shared_key",
"local_address",
"reserved",
"mtu",
):
proxy.pop(key, None)
if protocol == "vless":
proxy["uuid"] = self.ent_singbox_vless_uuid.text().strip()
flow = str(self.cmb_singbox_vless_flow.currentData() or "").strip()
if not flow:
flow = self.cmb_singbox_vless_flow.currentText().strip()
if flow:
proxy["flow"] = flow
packet_encoding = str(self.cmb_singbox_vless_packet_encoding.currentData() or "").strip().lower()
if packet_encoding and packet_encoding != "none":
proxy["packet_encoding"] = packet_encoding
elif protocol == "trojan":
proxy["password"] = self.ent_singbox_proto_password.text().strip()
elif protocol == "shadowsocks":
method = str(self.cmb_singbox_ss_method.currentData() or "").strip()
if not method:
method = self.cmb_singbox_ss_method.currentText().strip()
proxy["method"] = method
proxy["password"] = self.ent_singbox_proto_password.text().strip()
plugin = self.ent_singbox_ss_plugin.text().strip()
if plugin:
proxy["plugin"] = plugin
elif protocol == "hysteria2":
proxy["password"] = self.ent_singbox_proto_password.text().strip()
up = int(self.spn_singbox_hy2_up_mbps.value())
down = int(self.spn_singbox_hy2_down_mbps.value())
if up > 0:
proxy["up_mbps"] = up
if down > 0:
proxy["down_mbps"] = down
obfs_type = self.ent_singbox_hy2_obfs.text().strip()
if obfs_type:
obfs: dict[str, Any] = {"type": obfs_type}
obfs_password = self.ent_singbox_hy2_obfs_password.text().strip()
if obfs_password:
obfs["password"] = obfs_password
proxy["obfs"] = obfs
elif protocol == "tuic":
proxy["uuid"] = self.ent_singbox_vless_uuid.text().strip()
proxy["password"] = self.ent_singbox_proto_password.text().strip()
cc = str(self.cmb_singbox_tuic_congestion.currentData() or "").strip()
if not cc:
cc = self.cmb_singbox_tuic_congestion.currentText().strip()
if cc:
proxy["congestion_control"] = cc
udp_mode = str(self.cmb_singbox_tuic_udp_mode.currentData() or "").strip()
if udp_mode:
proxy["udp_relay_mode"] = udp_mode
if self.chk_singbox_tuic_zero_rtt.isChecked():
proxy["zero_rtt_handshake"] = True
elif protocol == "wireguard":
proxy["private_key"] = self.ent_singbox_wg_private_key.text().strip()
proxy["peer_public_key"] = self.ent_singbox_wg_peer_public_key.text().strip()
psk = self.ent_singbox_wg_psk.text().strip()
if psk:
proxy["pre_shared_key"] = psk
local_addr = [str(x).strip() for x in self.ent_singbox_wg_local_address.text().split(",") if str(x).strip()]
if local_addr:
proxy["local_address"] = local_addr
reserved_vals = self._parse_wg_reserved_values(
[str(x).strip() for x in self.ent_singbox_wg_reserved.text().split(",") if str(x).strip()],
strict=True,
)
if reserved_vals:
proxy["reserved"] = reserved_vals
mtu = int(self.spn_singbox_wg_mtu.value())
if mtu > 0:
proxy["mtu"] = mtu
transport_type = str(self.cmb_singbox_vless_transport.currentData() or "tcp").strip().lower()
if protocol in ("vless", "trojan"):
self._apply_proxy_transport(
proxy,
transport=transport_type,
path=self.ent_singbox_vless_path.text().strip(),
grpc_service=self.ent_singbox_vless_grpc_service.text().strip(),
)
else:
proxy.pop("transport", None)
security = str(self.cmb_singbox_vless_security.currentData() or "none").strip().lower()
if protocol == "vless":
pass
elif protocol == "trojan":
if security == "reality":
security = "tls"
elif protocol in ("hysteria2", "tuic"):
security = "tls"
else:
security = "none"
alpn = []
for p in self.ent_singbox_tls_alpn.text().split(","):
v = str(p or "").strip()
if v:
alpn.append(v)
self._apply_proxy_tls(
proxy,
security=security,
sni=self.ent_singbox_vless_sni.text().strip(),
utls_fp=str(self.cmb_singbox_vless_utls_fp.currentData() or "").strip(),
tls_insecure=bool(self.chk_singbox_vless_insecure.isChecked()),
reality_public_key=self.ent_singbox_vless_reality_pk.text().strip(),
reality_short_id=self.ent_singbox_vless_reality_sid.text().strip(),
alpn=alpn,
)
if proxy_idx >= 0:
outbounds[proxy_idx] = proxy
else:
outbounds.insert(0, proxy)
has_direct = any(
isinstance(row, dict)
and str(row.get("type") or "").strip().lower() == "direct"
and str(row.get("tag") or "").strip().lower() == "direct"
for row in outbounds
)
if not has_direct:
outbounds.append({"type": "direct", "tag": "direct"})
raw["outbounds"] = outbounds
inbounds = raw.get("inbounds") or []
if not isinstance(inbounds, list):
inbounds = []
inbound_idx = -1
for i, row in enumerate(inbounds):
if not isinstance(row, dict):
continue
tag = str(row.get("tag") or "").strip().lower()
t = str(row.get("type") or "").strip().lower()
if tag == "socks-in" or t == "socks":
inbound_idx = i
break
inbound = (
dict(inbounds[inbound_idx]) if inbound_idx >= 0 and isinstance(inbounds[inbound_idx], dict) else {}
)
inbound["type"] = str(inbound.get("type") or "socks")
inbound["tag"] = str(inbound.get("tag") or "socks-in")
inbound["listen"] = str(inbound.get("listen") or "127.0.0.1")
inbound["listen_port"] = int(inbound.get("listen_port") or 10808)
sniff = bool(self.chk_singbox_vless_sniff.isChecked())
inbound["sniff"] = sniff
inbound["sniff_override_destination"] = sniff
if inbound_idx >= 0:
inbounds[inbound_idx] = inbound
else:
inbounds.insert(0, inbound)
raw["inbounds"] = inbounds
route = raw.get("route") if isinstance(raw.get("route"), dict) else {}
route["final"] = str(route.get("final") or "direct")
rules = route.get("rules") or []
if not isinstance(rules, list):
rules = []
has_proxy_rule = False
for row in rules:
if not isinstance(row, dict):
continue
outbound = str(row.get("outbound") or "").strip().lower()
inbound_list = row.get("inbound") or []
if not isinstance(inbound_list, list):
inbound_list = []
inbound_norm = [str(x).strip().lower() for x in inbound_list if str(x).strip()]
if outbound == "proxy" and "socks-in" in inbound_norm:
has_proxy_rule = True
break
if not has_proxy_rule:
rules.insert(0, {"inbound": ["socks-in"], "outbound": "proxy"})
route["rules"] = rules
raw["route"] = route
return raw
def _save_singbox_editor_draft(self, client, *, profile_id: str = ""):
protocol = self._current_editor_protocol()
self._validate_singbox_editor_form()
raw_cfg = self._build_singbox_editor_raw_config()
name = self.ent_singbox_proto_name.text().strip()
enabled = bool(self.chk_singbox_proto_enabled.isChecked())
res = self.ctrl.singbox_profile_save_raw_for_client(
client,
profile_id=profile_id,
name=name,
enabled=enabled,
protocol=protocol,
raw_config=raw_cfg,
)
profile = self.ctrl.singbox_profile_get_for_client(client, profile_id=profile_id)
self._singbox_editor_profile_id = str(getattr(profile, "id", "") or "").strip()
self._singbox_editor_profile_client_id = str(getattr(client, "id", "") or "").strip()
self._singbox_editor_protocol = str(getattr(profile, "protocol", "") or protocol).strip().lower() or protocol
self._singbox_editor_source_raw = json.loads(json.dumps(getattr(profile, "raw_config", {}) or {}))
return res
def _sync_selected_singbox_profile_link(self, *, silent: bool = True) -> None:
client = self._selected_transport_client()
if client is None:
return
try:
preferred_pid = str(getattr(client, "id", "") or "").strip()
res = self.ctrl.singbox_profile_ensure_linked(
client,
preferred_profile_id=preferred_pid,
)
except Exception as e:
if not silent:
raise
self._append_transport_log(f"[profile] auto-link skipped: {e}")
return
line = (res.pretty_text or "").strip()
if not line:
return
# Keep noisy "already linked" messages out of normal flow.
if "already linked" in line.lower() and silent:
return
self._append_transport_log(f"[profile] {line}")
self.ctrl.log_gui(f"[singbox-profile] {line}")

View File

@@ -0,0 +1,271 @@
from __future__ import annotations
import json
from PySide6.QtWidgets import QApplication, QInputDialog, QMenu
from main_window.constants import SINGBOX_EDITOR_PROTOCOL_OPTIONS
class SingBoxLinksActionsMixin:
def _apply_singbox_editor_values(self, values: dict[str, Any]) -> None:
incoming = dict(values or {})
target_protocol = str(incoming.get("protocol") or self._current_editor_protocol() or "vless").strip().lower() or "vless"
payload = self._seed_editor_values_for_protocol(
target_protocol,
profile_name=str(incoming.get("profile_name") or "").strip(),
)
payload.update(incoming)
self._singbox_editor_loading = True
try:
name = str(payload.get("profile_name") or "").strip()
self.ent_singbox_proto_name.setText(name)
self.chk_singbox_proto_enabled.setChecked(bool(payload.get("enabled", True)))
protocol = str(payload.get("protocol") or "").strip().lower()
if protocol:
pidx = self.cmb_singbox_proto_protocol.findData(protocol)
self.cmb_singbox_proto_protocol.setCurrentIndex(pidx if pidx >= 0 else self.cmb_singbox_proto_protocol.currentIndex())
self.ent_singbox_vless_server.setText(str(payload.get("server") or "").strip())
try:
self.spn_singbox_vless_port.setValue(int(payload.get("port") or 443))
except Exception:
self.spn_singbox_vless_port.setValue(443)
self.ent_singbox_vless_uuid.setText(str(payload.get("uuid") or "").strip())
self.ent_singbox_proto_password.setText(str(payload.get("password") or "").strip())
flow_v = str(payload.get("flow") or "").strip()
flow_idx = self.cmb_singbox_vless_flow.findData(flow_v)
if flow_idx >= 0:
self.cmb_singbox_vless_flow.setCurrentIndex(flow_idx)
else:
self.cmb_singbox_vless_flow.setEditText(flow_v)
packet_v = str(payload.get("packet_encoding") or "").strip().lower()
if packet_v in ("none", "off", "false"):
packet_v = ""
packet_idx = self.cmb_singbox_vless_packet_encoding.findData(packet_v)
self.cmb_singbox_vless_packet_encoding.setCurrentIndex(packet_idx if packet_idx >= 0 else 0)
transport_v = str(payload.get("transport") or "tcp").strip().lower()
transport_idx = self.cmb_singbox_vless_transport.findData(transport_v)
self.cmb_singbox_vless_transport.setCurrentIndex(transport_idx if transport_idx >= 0 else 0)
self.ent_singbox_vless_path.setText(str(payload.get("path") or "").strip())
self.ent_singbox_vless_grpc_service.setText(str(payload.get("grpc_service") or "").strip())
sec_v = str(payload.get("security") or "none").strip().lower()
sec_idx = self.cmb_singbox_vless_security.findData(sec_v)
self.cmb_singbox_vless_security.setCurrentIndex(sec_idx if sec_idx >= 0 else 0)
self.ent_singbox_vless_sni.setText(str(payload.get("sni") or "").strip())
fp_v = str(payload.get("utls_fp") or "").strip().lower()
fp_idx = self.cmb_singbox_vless_utls_fp.findData(fp_v)
self.cmb_singbox_vless_utls_fp.setCurrentIndex(fp_idx if fp_idx >= 0 else 0)
self.ent_singbox_vless_reality_pk.setText(str(payload.get("reality_public_key") or "").strip())
self.ent_singbox_vless_reality_sid.setText(str(payload.get("reality_short_id") or "").strip())
self.chk_singbox_vless_insecure.setChecked(bool(payload.get("tls_insecure", False)))
self.chk_singbox_vless_sniff.setChecked(bool(payload.get("sniff", True)))
ss_method = str(payload.get("ss_method") or "").strip().lower()
if ss_method:
idx = self.cmb_singbox_ss_method.findData(ss_method)
if idx >= 0:
self.cmb_singbox_ss_method.setCurrentIndex(idx)
else:
self.cmb_singbox_ss_method.setEditText(ss_method)
else:
self.cmb_singbox_ss_method.setCurrentIndex(0)
self.ent_singbox_ss_plugin.setText(str(payload.get("ss_plugin") or "").strip())
try:
self.spn_singbox_hy2_up_mbps.setValue(int(payload.get("hy2_up_mbps") or 0))
except Exception:
self.spn_singbox_hy2_up_mbps.setValue(0)
try:
self.spn_singbox_hy2_down_mbps.setValue(int(payload.get("hy2_down_mbps") or 0))
except Exception:
self.spn_singbox_hy2_down_mbps.setValue(0)
self.ent_singbox_hy2_obfs.setText(str(payload.get("hy2_obfs") or "").strip())
self.ent_singbox_hy2_obfs_password.setText(str(payload.get("hy2_obfs_password") or "").strip())
tuic_cc = str(payload.get("tuic_congestion") or "").strip()
idx = self.cmb_singbox_tuic_congestion.findData(tuic_cc)
self.cmb_singbox_tuic_congestion.setCurrentIndex(idx if idx >= 0 else 0)
tuic_udp = str(payload.get("tuic_udp_mode") or "").strip()
idx = self.cmb_singbox_tuic_udp_mode.findData(tuic_udp)
self.cmb_singbox_tuic_udp_mode.setCurrentIndex(idx if idx >= 0 else 0)
self.chk_singbox_tuic_zero_rtt.setChecked(bool(payload.get("tuic_zero_rtt", False)))
self.ent_singbox_wg_private_key.setText(str(payload.get("wg_private_key") or "").strip())
self.ent_singbox_wg_peer_public_key.setText(str(payload.get("wg_peer_public_key") or "").strip())
self.ent_singbox_wg_psk.setText(str(payload.get("wg_psk") or "").strip())
wg_local = payload.get("wg_local_address") or []
if isinstance(wg_local, list):
self.ent_singbox_wg_local_address.setText(
",".join([str(x).strip() for x in wg_local if str(x).strip()])
)
else:
self.ent_singbox_wg_local_address.setText(str(wg_local or "").strip())
wg_reserved = payload.get("wg_reserved") or []
if isinstance(wg_reserved, list):
self.ent_singbox_wg_reserved.setText(
",".join([str(x).strip() for x in wg_reserved if str(x).strip()])
)
else:
self.ent_singbox_wg_reserved.setText(str(wg_reserved or "").strip())
try:
self.spn_singbox_wg_mtu.setValue(int(payload.get("wg_mtu") or 0))
except Exception:
self.spn_singbox_wg_mtu.setValue(0)
finally:
self._singbox_editor_loading = False
self.on_singbox_vless_editor_changed()
def _create_singbox_connection(
self,
*,
profile_name: str,
protocol: str = "vless",
raw_config: dict[str, Any] | None = None,
editor_values: dict[str, Any] | None = None,
auto_save: bool = False,
) -> str:
name = str(profile_name or "").strip() or "SingBox connection"
client_id = self._next_free_transport_client_id(name)
proto = self._normalized_seed_protocol(protocol)
config = self._default_new_singbox_client_config(client_id, protocol=proto)
created = self.ctrl.transport_client_create_action(
client_id=client_id,
kind="singbox",
name=name,
enabled=True,
config=config,
)
line = (created.pretty_text or "").strip() or f"create {client_id}"
self._append_transport_log(f"[engine] {line}")
self.ctrl.log_gui(f"[transport-engine] {line}")
if not created.ok:
raise RuntimeError(line)
self.refresh_transport_engines(silent=True)
if not self._select_transport_engine_by_id(client_id):
raise RuntimeError(f"created client '{client_id}' was not found after refresh")
self._sync_selected_singbox_profile_link(silent=False)
client, _eid, pid = self._selected_singbox_profile_context()
seed_raw = raw_config if isinstance(raw_config, dict) else self._seed_raw_config_for_protocol(proto)
saved_seed = self.ctrl.singbox_profile_save_raw_for_client(
client,
profile_id=pid,
name=name,
enabled=True,
protocol=proto,
raw_config=seed_raw,
)
seed_line = (saved_seed.pretty_text or "").strip() or f"save profile {pid}"
self._append_transport_log(f"[profile] {seed_line}")
self.ctrl.log_gui(f"[singbox-profile] {seed_line}")
self._load_singbox_editor_for_selected(silent=True)
if editor_values:
payload = dict(editor_values)
seeded = self._seed_editor_values_for_protocol(proto, profile_name=name)
seeded.update(payload)
payload = seeded
if not str(payload.get("profile_name") or "").strip():
payload["profile_name"] = name
self._apply_singbox_editor_values(payload)
if auto_save:
saved = self._save_singbox_editor_draft(client, profile_id=pid)
save_line = (saved.pretty_text or "").strip() or f"save profile {pid}"
self._append_transport_log(f"[profile] {save_line}")
self.ctrl.log_gui(f"[singbox-profile] {save_line}")
return client_id
def on_singbox_create_connection_click(self) -> None:
menu = QMenu(self)
act_clip = menu.addAction("Create from clipboard")
act_link = menu.addAction("Create from link...")
act_manual = menu.addAction("Create manual")
pos = self.btn_singbox_profile_create.mapToGlobal(
self.btn_singbox_profile_create.rect().bottomLeft()
)
chosen = menu.exec(pos)
if chosen is None:
return
if chosen == act_clip:
self._safe(self.on_singbox_create_connection_from_clipboard, title="Create connection error")
return
if chosen == act_link:
self._safe(self.on_singbox_create_connection_from_link, title="Create connection error")
return
if chosen == act_manual:
self._safe(self.on_singbox_create_connection_manual, title="Create connection error")
def on_singbox_create_connection_from_clipboard(self) -> None:
raw = str(QApplication.clipboard().text() or "").strip()
if not raw:
raise RuntimeError("Clipboard is empty")
payload = self._parse_connection_link_payload(raw)
profile_name = str(payload.get("profile_name") or "").strip() or "SingBox Clipboard"
cid = self._create_singbox_connection(
profile_name=profile_name,
protocol=str(payload.get("protocol") or "vless"),
raw_config=payload.get("raw_config") if isinstance(payload.get("raw_config"), dict) else None,
editor_values=payload.get("editor_values") if isinstance(payload.get("editor_values"), dict) else None,
auto_save=True,
)
self.on_singbox_profile_edit_dialog(cid)
def on_singbox_create_connection_from_link(self) -> None:
raw, ok = QInputDialog.getText(
self,
"Create connection from link",
"Paste connection link (vless:// trojan:// ss:// hysteria2:// hy2:// tuic:// wireguard://):",
)
if not ok:
return
payload = self._parse_connection_link_payload(raw)
profile_name = str(payload.get("profile_name") or "").strip() or "SingBox Link"
cid = self._create_singbox_connection(
profile_name=profile_name,
protocol=str(payload.get("protocol") or "vless"),
raw_config=payload.get("raw_config") if isinstance(payload.get("raw_config"), dict) else None,
editor_values=payload.get("editor_values") if isinstance(payload.get("editor_values"), dict) else None,
auto_save=True,
)
self.on_singbox_profile_edit_dialog(cid)
def on_singbox_create_connection_manual(self) -> None:
name, ok = QInputDialog.getText(
self,
"Create manual connection",
"Connection name:",
)
if not ok:
return
profile_name = str(name or "").strip() or "SingBox Manual"
proto_title, ok = QInputDialog.getItem(
self,
"Create manual connection",
"Protocol:",
[label for label, _pid in SINGBOX_EDITOR_PROTOCOL_OPTIONS],
0,
False,
)
if not ok:
return
proto_map = {label.lower(): pid for label, pid in SINGBOX_EDITOR_PROTOCOL_OPTIONS}
proto = self._normalized_seed_protocol(proto_map.get(str(proto_title or "").strip().lower(), "vless"))
cid = self._create_singbox_connection(
profile_name=profile_name,
protocol=proto,
editor_values=self._seed_editor_values_for_protocol(proto, profile_name=profile_name),
auto_save=False,
)
self.on_singbox_profile_edit_dialog(cid)

View File

@@ -0,0 +1,337 @@
from __future__ import annotations
import base64
import binascii
import json
import re
from urllib.parse import unquote
from typing import Any
from main_window.constants import SINGBOX_EDITOR_PROTOCOL_IDS, SINGBOX_PROTOCOL_SEED_SPEC
class SingBoxLinksHelpersMixin:
def _slugify_connection_id(self, text: str) -> str:
raw = str(text or "").strip().lower()
raw = re.sub(r"[^a-z0-9]+", "-", raw)
raw = re.sub(r"-{2,}", "-", raw).strip("-")
if not raw:
raw = "connection"
if not raw.startswith("sg-"):
raw = f"sg-{raw}"
return raw
def _next_free_transport_client_id(self, base_hint: str) -> str:
base = self._slugify_connection_id(base_hint)
existing = {str(getattr(c, "id", "") or "").strip() for c in (self._transport_clients or [])}
if base not in existing:
return base
i = 2
while True:
cid = f"{base}-{i}"
if cid not in existing:
return cid
i += 1
def _template_singbox_client(self):
selected = self._selected_transport_client()
if selected is not None and str(getattr(selected, "kind", "") or "").strip().lower() == "singbox":
return selected
for c in self._transport_clients or []:
if str(getattr(c, "kind", "") or "").strip().lower() == "singbox":
return c
return None
def _default_new_singbox_client_config(self, client_id: str, *, protocol: str = "vless") -> dict[str, Any]:
cfg: dict[str, Any] = {}
tpl = self._template_singbox_client()
if tpl is not None:
src_cfg = getattr(tpl, "config", {}) or {}
if isinstance(src_cfg, dict):
for key in (
"runner",
"runtime_mode",
"require_binary",
"exec_start",
"singbox_bin",
"packaging_profile",
"packaging_system_fallback",
"bin_root",
"hardening_enabled",
"hardening_profile",
"restart",
"restart_sec",
"watchdog_sec",
"start_limit_interval_sec",
"start_limit_burst",
"timeout_start_sec",
"timeout_stop_sec",
"bootstrap_bypass_strict",
"netns_enabled",
"netns_name",
"netns_auto_cleanup",
"netns_setup_strict",
"singbox_dns_migrate_legacy",
"singbox_dns_migrate_strict",
):
if key in src_cfg:
cfg[key] = json.loads(json.dumps(src_cfg.get(key)))
cid = str(client_id or "").strip()
if not cid:
return cfg
for key in ("profile", "profile_id", "singbox_profile_id"):
cfg.pop(key, None)
config_path = f"/etc/selective-vpn/transports/{cid}/singbox.json"
cfg["config_path"] = config_path
cfg["singbox_config_path"] = config_path
runner = str(cfg.get("runner") or "").strip().lower()
if not runner:
cfg["runner"] = "systemd"
runner = "systemd"
if runner == "systemd":
cfg["unit"] = "singbox@.service"
if "runtime_mode" not in cfg:
cfg["runtime_mode"] = "exec"
if "require_binary" not in cfg:
cfg["require_binary"] = True
cfg["profile_id"] = cid
cfg["protocol"] = self._normalized_seed_protocol(protocol)
return cfg
def _normalized_seed_protocol(self, protocol: str) -> str:
proto = str(protocol or "vless").strip().lower() or "vless"
if proto not in SINGBOX_EDITOR_PROTOCOL_IDS:
proto = "vless"
return proto
def _protocol_seed_spec(self, protocol: str) -> dict[str, Any]:
proto = self._normalized_seed_protocol(protocol)
spec = SINGBOX_PROTOCOL_SEED_SPEC.get(proto) or SINGBOX_PROTOCOL_SEED_SPEC.get("vless") or {}
if not isinstance(spec, dict):
spec = {}
return dict(spec)
def _seed_editor_values_for_protocol(self, protocol: str, *, profile_name: str = "") -> dict[str, Any]:
proto = self._normalized_seed_protocol(protocol)
spec = self._protocol_seed_spec(proto)
security = str(spec.get("security") or "none").strip().lower() or "none"
port = int(spec.get("port") or (51820 if proto == "wireguard" else 443))
return {
"profile_name": str(profile_name or "").strip(),
"enabled": True,
"protocol": proto,
"server": "",
"port": port,
"uuid": "",
"password": "",
"flow": "",
"packet_encoding": "",
"transport": "tcp",
"path": "",
"grpc_service": "",
"security": security,
"sni": "",
"utls_fp": "",
"reality_public_key": "",
"reality_short_id": "",
"tls_insecure": False,
"sniff": True,
"ss_method": "aes-128-gcm",
"ss_plugin": "",
"hy2_up_mbps": 0,
"hy2_down_mbps": 0,
"hy2_obfs": "",
"hy2_obfs_password": "",
"tuic_congestion": "",
"tuic_udp_mode": "",
"tuic_zero_rtt": False,
"wg_private_key": "",
"wg_peer_public_key": "",
"wg_psk": "",
"wg_local_address": "",
"wg_reserved": "",
"wg_mtu": 0,
}
def _seed_raw_config_for_protocol(self, protocol: str) -> dict[str, Any]:
proto = self._normalized_seed_protocol(protocol)
spec = self._protocol_seed_spec(proto)
port = int(spec.get("port") or (51820 if proto == "wireguard" else 443))
proxy: dict[str, Any] = {
"type": proto,
"tag": "proxy",
"server": "",
"server_port": port,
}
proxy_defaults = spec.get("proxy_defaults") or {}
if isinstance(proxy_defaults, dict):
for key, value in proxy_defaults.items():
proxy[key] = json.loads(json.dumps(value))
tls_security = str(spec.get("tls_security") or "").strip().lower()
if tls_security in ("tls", "reality"):
self._apply_proxy_tls(proxy, security=tls_security)
return self._build_singbox_raw_config_from_proxy(proxy, sniff=True)
def _parse_wg_reserved_values(self, raw_values: list[str], *, strict: bool) -> list[int]:
vals = [str(x).strip() for x in list(raw_values or []) if str(x).strip()]
if len(vals) > 3:
if strict:
raise RuntimeError("WG reserved accepts up to 3 values (0..255)")
vals = vals[:3]
out: list[int] = []
for token in vals:
try:
num = int(token)
except Exception:
if strict:
raise RuntimeError(f"WG reserved value '{token}' is not an integer")
continue
if num < 0 or num > 255:
if strict:
raise RuntimeError(f"WG reserved value '{token}' must be in range 0..255")
continue
out.append(num)
return out
def _query_value(self, query: dict[str, list[str]], *keys: str) -> str:
for k in keys:
vals = query.get(str(k or "").strip())
if not vals:
continue
v = str(vals[0] or "").strip()
if v:
return unquote(v)
return ""
def _query_bool(self, query: dict[str, list[str]], *keys: str) -> bool:
v = self._query_value(query, *keys).strip().lower()
return v in ("1", "true", "yes", "on")
def _query_csv(self, query: dict[str, list[str]], *keys: str) -> list[str]:
raw = self._query_value(query, *keys)
if not raw:
return []
out: list[str] = []
for p in raw.split(","):
val = str(p or "").strip()
if val:
out.append(val)
return out
def _normalize_link_transport(self, value: str) -> str:
v = str(value or "").strip().lower() or "tcp"
if v == "raw":
v = "tcp"
if v in ("h2", "http2"):
v = "http"
if v not in ("tcp", "ws", "grpc", "http", "httpupgrade", "quic"):
v = "tcp"
return v
def _b64_urlsafe_decode(self, value: str) -> str:
raw = str(value or "").strip()
if not raw:
return ""
pad = "=" * ((4 - (len(raw) % 4)) % 4)
try:
data = base64.urlsafe_b64decode((raw + pad).encode("utf-8"))
return data.decode("utf-8", errors="replace")
except (binascii.Error, ValueError):
return ""
def _apply_proxy_transport(
self,
proxy: dict[str, Any],
*,
transport: str,
path: str = "",
grpc_service: str = "",
) -> None:
t = self._normalize_link_transport(transport)
if t in ("", "tcp"):
proxy.pop("transport", None)
return
tx: dict[str, Any] = {"type": t}
if t in ("ws", "http", "httpupgrade"):
tx["path"] = str(path or "/").strip() or "/"
if t == "grpc":
tx["service_name"] = str(grpc_service or "").strip()
proxy["transport"] = tx
def _apply_proxy_tls(
self,
proxy: dict[str, Any],
*,
security: str,
sni: str = "",
utls_fp: str = "",
tls_insecure: bool = False,
reality_public_key: str = "",
reality_short_id: str = "",
alpn: list[str] | None = None,
) -> None:
sec = str(security or "").strip().lower()
if sec not in ("none", "tls", "reality"):
sec = "none"
if sec == "none":
proxy.pop("tls", None)
return
tls: dict[str, Any] = {
"enabled": True,
"insecure": bool(tls_insecure),
}
if str(sni or "").strip():
tls["server_name"] = str(sni).strip()
if str(utls_fp or "").strip():
tls["utls"] = {"enabled": True, "fingerprint": str(utls_fp).strip().lower()}
alpn_vals = [str(x).strip() for x in list(alpn or []) if str(x).strip()]
if alpn_vals:
tls["alpn"] = alpn_vals
if sec == "reality":
reality: dict[str, Any] = {
"enabled": True,
"public_key": str(reality_public_key or "").strip(),
}
sid = str(reality_short_id or "").strip()
if sid:
reality["short_id"] = sid
tls["reality"] = reality
proxy["tls"] = tls
def _build_singbox_raw_config_from_proxy(
self,
proxy: dict[str, Any],
*,
sniff: bool = True,
) -> dict[str, Any]:
return {
"inbounds": [
{
"type": "socks",
"tag": "socks-in",
"listen": "127.0.0.1",
"listen_port": 10808,
"sniff": bool(sniff),
"sniff_override_destination": bool(sniff),
}
],
"outbounds": [
proxy,
{"type": "direct", "tag": "direct"},
],
"route": {
"final": "direct",
"rules": [
{"inbound": ["socks-in"], "outbound": "proxy"},
],
},
}

View File

@@ -0,0 +1,16 @@
from __future__ import annotations
from main_window.singbox.links_actions_mixin import SingBoxLinksActionsMixin
from main_window.singbox.links_helpers_mixin import SingBoxLinksHelpersMixin
from main_window.singbox.links_parsers_mixin import SingBoxLinksParsersMixin
class SingBoxLinksMixin(
SingBoxLinksActionsMixin,
SingBoxLinksParsersMixin,
SingBoxLinksHelpersMixin,
):
"""Facade mixin for SingBox link import/create workflow."""
__all__ = ["SingBoxLinksMixin"]

View File

@@ -0,0 +1,391 @@
from __future__ import annotations
import re
from urllib.parse import parse_qs, unquote, urlsplit
from typing import Any
class SingBoxLinksParsersMixin:
def _parse_vless_link_payload(self, link: str) -> dict[str, Any]:
u = urlsplit(link)
query = parse_qs(u.query or "", keep_blank_values=True)
uuid = unquote(str(u.username or "").strip())
host = str(u.hostname or "").strip()
if not uuid:
raise RuntimeError("VLESS link has no UUID")
if not host:
raise RuntimeError("VLESS link has no host")
try:
port = int(u.port or 443)
except Exception:
port = 443
transport = self._normalize_link_transport(self._query_value(query, "type", "transport"))
security = self._query_value(query, "security").strip().lower() or "none"
if security == "xtls":
security = "tls"
if security not in ("none", "tls", "reality"):
security = "none"
path = self._query_value(query, "path", "spx")
if not path and str(u.path or "").strip() not in ("", "/"):
path = unquote(str(u.path or "").strip())
grpc_service = self._query_value(query, "serviceName", "service_name")
if transport == "grpc" and not grpc_service:
grpc_service = self._query_value(query, "path")
flow = self._query_value(query, "flow")
packet_encoding = self._query_value(query, "packetEncoding", "packet_encoding").strip().lower()
if packet_encoding in ("none", "off", "false"):
packet_encoding = ""
sni = self._query_value(query, "sni", "host")
utls_fp = self._query_value(query, "fp", "fingerprint")
reality_pk = self._query_value(query, "pbk", "public_key")
reality_sid = self._query_value(query, "sid", "short_id")
tls_insecure = self._query_bool(query, "allowInsecure", "insecure")
profile_name = unquote(str(u.fragment or "").strip()) or host
proxy: dict[str, Any] = {
"type": "vless",
"tag": "proxy",
"server": host,
"server_port": port,
"uuid": uuid,
}
if packet_encoding:
proxy["packet_encoding"] = packet_encoding
if flow:
proxy["flow"] = flow
self._apply_proxy_transport(proxy, transport=transport, path=path, grpc_service=grpc_service)
self._apply_proxy_tls(
proxy,
security=security,
sni=sni,
utls_fp=utls_fp,
tls_insecure=tls_insecure,
reality_public_key=reality_pk,
reality_short_id=reality_sid,
)
editor_values = {
"profile_name": profile_name,
"enabled": True,
"server": host,
"port": port,
"uuid": uuid,
"flow": flow,
"packet_encoding": packet_encoding,
"transport": transport,
"path": path,
"grpc_service": grpc_service,
"security": security,
"sni": sni,
"utls_fp": utls_fp,
"reality_public_key": reality_pk,
"reality_short_id": reality_sid,
"tls_insecure": tls_insecure,
"sniff": True,
}
return {
"protocol": "vless",
"profile_name": profile_name,
"raw_config": self._build_singbox_raw_config_from_proxy(proxy, sniff=True),
"editor_values": editor_values,
}
def _parse_trojan_link_payload(self, link: str) -> dict[str, Any]:
u = urlsplit(link)
query = parse_qs(u.query or "", keep_blank_values=True)
password = unquote(str(u.username or "").strip()) or self._query_value(query, "password")
host = str(u.hostname or "").strip()
if not password:
raise RuntimeError("Trojan link has no password")
if not host:
raise RuntimeError("Trojan link has no host")
try:
port = int(u.port or 443)
except Exception:
port = 443
transport = self._normalize_link_transport(self._query_value(query, "type", "transport"))
path = self._query_value(query, "path")
grpc_service = self._query_value(query, "serviceName", "service_name")
security = self._query_value(query, "security").strip().lower() or "tls"
if security not in ("none", "tls"):
security = "tls"
sni = self._query_value(query, "sni", "host")
utls_fp = self._query_value(query, "fp", "fingerprint")
tls_insecure = self._query_bool(query, "allowInsecure", "insecure")
alpn = self._query_csv(query, "alpn")
profile_name = unquote(str(u.fragment or "").strip()) or host
proxy: dict[str, Any] = {
"type": "trojan",
"tag": "proxy",
"server": host,
"server_port": port,
"password": password,
}
self._apply_proxy_transport(proxy, transport=transport, path=path, grpc_service=grpc_service)
self._apply_proxy_tls(
proxy,
security=security,
sni=sni,
utls_fp=utls_fp,
tls_insecure=tls_insecure,
alpn=alpn,
)
return {
"protocol": "trojan",
"profile_name": profile_name,
"raw_config": self._build_singbox_raw_config_from_proxy(proxy, sniff=True),
}
def _parse_ss_link_payload(self, link: str) -> dict[str, Any]:
raw = str(link or "").strip()
u = urlsplit(raw)
query = parse_qs(u.query or "", keep_blank_values=True)
profile_name = unquote(str(u.fragment or "").strip()) or "Shadowsocks"
body = raw[len("ss://"):]
body = body.split("#", 1)[0]
body = body.split("?", 1)[0]
method = ""
password = ""
host_port = ""
if "@" in body:
left, host_port = body.rsplit("@", 1)
creds = left
if ":" not in creds:
creds = self._b64_urlsafe_decode(creds)
if ":" not in creds:
raise RuntimeError("Shadowsocks link has invalid credentials")
method, password = creds.split(":", 1)
else:
decoded = self._b64_urlsafe_decode(body)
if "@" not in decoded:
raise RuntimeError("Shadowsocks link has invalid payload")
creds, host_port = decoded.rsplit("@", 1)
if ":" not in creds:
raise RuntimeError("Shadowsocks link has invalid credentials")
method, password = creds.split(":", 1)
hp = urlsplit("//" + host_port)
host = str(hp.hostname or "").strip()
if not host:
raise RuntimeError("Shadowsocks link has no host")
try:
port = int(hp.port or 8388)
except Exception:
port = 8388
proxy: dict[str, Any] = {
"type": "shadowsocks",
"tag": "proxy",
"server": host,
"server_port": port,
"method": str(method or "").strip(),
"password": str(password or "").strip(),
}
plugin = self._query_value(query, "plugin")
if plugin:
proxy["plugin"] = plugin
return {
"protocol": "shadowsocks",
"profile_name": profile_name,
"raw_config": self._build_singbox_raw_config_from_proxy(proxy, sniff=True),
}
def _parse_hysteria2_link_payload(self, link: str) -> dict[str, Any]:
u = urlsplit(link)
query = parse_qs(u.query or "", keep_blank_values=True)
password = unquote(str(u.username or "").strip()) or self._query_value(query, "password")
host = str(u.hostname or "").strip()
if not password:
raise RuntimeError("Hysteria2 link has no password")
if not host:
raise RuntimeError("Hysteria2 link has no host")
try:
port = int(u.port or 443)
except Exception:
port = 443
profile_name = unquote(str(u.fragment or "").strip()) or host
proxy: dict[str, Any] = {
"type": "hysteria2",
"tag": "proxy",
"server": host,
"server_port": port,
"password": password,
}
up_mbps = self._query_value(query, "up_mbps", "upmbps", "up")
down_mbps = self._query_value(query, "down_mbps", "downmbps", "down")
try:
if up_mbps:
proxy["up_mbps"] = int(float(up_mbps))
except Exception:
pass
try:
if down_mbps:
proxy["down_mbps"] = int(float(down_mbps))
except Exception:
pass
obfs_type = self._query_value(query, "obfs")
if obfs_type:
obfs: dict[str, Any] = {"type": obfs_type}
obfs_pw = self._query_value(query, "obfs-password", "obfs_password")
if obfs_pw:
obfs["password"] = obfs_pw
proxy["obfs"] = obfs
self._apply_proxy_tls(
proxy,
security="tls",
sni=self._query_value(query, "sni"),
tls_insecure=self._query_bool(query, "allowInsecure", "insecure"),
alpn=self._query_csv(query, "alpn"),
)
return {
"protocol": "hysteria2",
"profile_name": profile_name,
"raw_config": self._build_singbox_raw_config_from_proxy(proxy, sniff=True),
}
def _parse_tuic_link_payload(self, link: str) -> dict[str, Any]:
u = urlsplit(link)
query = parse_qs(u.query or "", keep_blank_values=True)
uuid = unquote(str(u.username or "").strip())
password = unquote(str(u.password or "").strip())
host = str(u.hostname or "").strip()
if not uuid:
raise RuntimeError("TUIC link has no UUID")
if not password:
raise RuntimeError("TUIC link has no password")
if not host:
raise RuntimeError("TUIC link has no host")
try:
port = int(u.port or 443)
except Exception:
port = 443
profile_name = unquote(str(u.fragment or "").strip()) or host
proxy: dict[str, Any] = {
"type": "tuic",
"tag": "proxy",
"server": host,
"server_port": port,
"uuid": uuid,
"password": password,
}
cc = self._query_value(query, "congestion_control", "congestion")
if cc:
proxy["congestion_control"] = cc
udp_mode = self._query_value(query, "udp_relay_mode")
if udp_mode:
proxy["udp_relay_mode"] = udp_mode
if self._query_bool(query, "zero_rtt_handshake", "zero_rtt"):
proxy["zero_rtt_handshake"] = True
self._apply_proxy_tls(
proxy,
security="tls",
sni=self._query_value(query, "sni", "host"),
utls_fp=self._query_value(query, "fp", "fingerprint"),
tls_insecure=self._query_bool(query, "allowInsecure", "insecure"),
alpn=self._query_csv(query, "alpn"),
)
return {
"protocol": "tuic",
"profile_name": profile_name,
"raw_config": self._build_singbox_raw_config_from_proxy(proxy, sniff=True),
}
def _parse_wireguard_link_payload(self, link: str) -> dict[str, Any]:
u = urlsplit(link)
query = parse_qs(u.query or "", keep_blank_values=True)
private_key = unquote(str(u.username or "").strip()) or self._query_value(query, "private_key", "privateKey")
host = str(u.hostname or "").strip()
if not host:
raise RuntimeError("WireGuard link has no host")
if not private_key:
raise RuntimeError("WireGuard link has no private key")
try:
port = int(u.port or 443)
except Exception:
port = 443
peer_public_key = self._query_value(query, "peer_public_key", "public_key", "peerPublicKey")
if not peer_public_key:
raise RuntimeError("WireGuard link has no peer public key")
local_address = self._query_csv(query, "local_address", "address", "localAddress")
if not local_address:
raise RuntimeError("WireGuard link has no local address")
profile_name = unquote(str(u.fragment or "").strip()) or host
proxy: dict[str, Any] = {
"type": "wireguard",
"tag": "proxy",
"server": host,
"server_port": port,
"private_key": private_key,
"peer_public_key": peer_public_key,
"local_address": local_address,
}
psk = self._query_value(query, "pre_shared_key", "psk", "preSharedKey")
if psk:
proxy["pre_shared_key"] = psk
reserved_vals = self._parse_wg_reserved_values(self._query_csv(query, "reserved"), strict=True)
if reserved_vals:
proxy["reserved"] = reserved_vals
mtu_val = self._query_value(query, "mtu")
try:
mtu = int(mtu_val) if mtu_val else 0
except Exception:
mtu = 0
if mtu > 0:
proxy["mtu"] = mtu
return {
"protocol": "wireguard",
"profile_name": profile_name,
"raw_config": self._build_singbox_raw_config_from_proxy(proxy, sniff=True),
}
def _extract_first_connection_link(self, text: str) -> str:
raw = str(text or "").strip()
if not raw:
return ""
m = re.search(r"(?i)(vless|trojan|ss|hysteria2|hy2|tuic|wireguard|wg)://\S+", raw)
if m:
return str(m.group(0) or "").strip()
if "://" in raw:
return raw.splitlines()[0].strip()
return ""
def _parse_connection_link_payload(self, text: str) -> dict[str, Any]:
raw = self._extract_first_connection_link(text)
if not raw:
raise RuntimeError(
"No supported link found. Supported schemes: "
"vless:// trojan:// ss:// hysteria2:// hy2:// tuic:// wireguard:// wg://"
)
u = urlsplit(raw)
scheme = str(u.scheme or "").strip().lower()
if scheme == "vless":
return self._parse_vless_link_payload(raw)
if scheme == "trojan":
return self._parse_trojan_link_payload(raw)
if scheme == "ss":
return self._parse_ss_link_payload(raw)
if scheme in ("hysteria2", "hy2"):
return self._parse_hysteria2_link_payload(raw)
if scheme == "tuic":
return self._parse_tuic_link_payload(raw)
if scheme in ("wireguard", "wg"):
return self._parse_wireguard_link_payload(raw)
raise RuntimeError(f"Unsupported link scheme: {scheme}")

View File

@@ -0,0 +1,136 @@
from __future__ import annotations
from PySide6.QtCore import Qt
from PySide6.QtWidgets import QMenu, QMessageBox
class SingBoxRuntimeCardsMixin:
def on_singbox_profile_card_context_menu(self, pos) -> None:
item = self.lst_singbox_profile_cards.itemAt(pos)
if item is None:
return
cid = str(item.data(Qt.UserRole) or "").strip()
if not cid:
return
menu = QMenu(self)
act_run = menu.addAction("Run")
act_edit = menu.addAction("Edit")
act_delete = menu.addAction("Delete")
chosen = menu.exec(self.lst_singbox_profile_cards.viewport().mapToGlobal(pos))
if chosen is None:
return
if not self._select_transport_engine_by_id(cid):
QMessageBox.warning(self, "SingBox profile", f"Profile '{cid}' is no longer available.")
return
if chosen == act_run:
self.on_transport_engine_action("start")
return
if chosen == act_edit:
self.on_singbox_profile_edit_dialog(cid)
return
if chosen == act_delete:
self.on_transport_engine_delete(cid)
return
def on_singbox_profile_card_selected(self) -> None:
if self._syncing_singbox_selection:
return
items = self.lst_singbox_profile_cards.selectedItems()
if not items:
return
cid = str(items[0].data(Qt.UserRole) or "").strip()
if not cid:
return
idx = self.cmb_transport_engine.findData(cid)
if idx < 0:
return
if idx != self.cmb_transport_engine.currentIndex():
self._syncing_singbox_selection = True
try:
self.cmb_transport_engine.setCurrentIndex(idx)
finally:
self._syncing_singbox_selection = False
return
self._refresh_singbox_profile_card_styles()
self._sync_selected_singbox_profile_link(silent=True)
self._load_singbox_editor_for_selected(silent=True)
self._update_transport_engine_view()
def _singbox_value_label(self, key: str, value: str) -> str:
v = str(value or "").strip().lower()
if key == "routing":
if v == "full":
return "Full tunnel"
return "Selective"
if key == "dns":
if v == "singbox_dns":
return "SingBox DNS"
return "System resolver"
if key == "killswitch":
if v == "off":
return "Disabled"
return "Enabled"
return v or ""
def _effective_singbox_policy(self) -> tuple[str, str, str]:
route = str(self.cmb_singbox_global_routing.currentData() or "selective").strip().lower()
dns = str(self.cmb_singbox_global_dns.currentData() or "system_resolver").strip().lower()
killswitch = str(self.cmb_singbox_global_killswitch.currentData() or "on").strip().lower()
if not self.chk_singbox_profile_use_global_routing.isChecked():
route = str(self.cmb_singbox_profile_routing.currentData() or route).strip().lower()
if route == "global":
route = str(self.cmb_singbox_global_routing.currentData() or "selective").strip().lower()
if not self.chk_singbox_profile_use_global_dns.isChecked():
dns = str(self.cmb_singbox_profile_dns.currentData() or dns).strip().lower()
if dns == "global":
dns = str(self.cmb_singbox_global_dns.currentData() or "system_resolver").strip().lower()
if not self.chk_singbox_profile_use_global_killswitch.isChecked():
killswitch = str(self.cmb_singbox_profile_killswitch.currentData() or killswitch).strip().lower()
if killswitch == "global":
killswitch = str(self.cmb_singbox_global_killswitch.currentData() or "on").strip().lower()
return route, dns, killswitch
def _refresh_singbox_profile_effective(self) -> None:
route, dns, killswitch = self._effective_singbox_policy()
route_txt = self._singbox_value_label("routing", route)
dns_txt = self._singbox_value_label("dns", dns)
kill_txt = self._singbox_value_label("killswitch", killswitch)
self.lbl_singbox_profile_effective.setText(
f"Effective: routing={route_txt} | dns={dns_txt} | kill-switch={kill_txt}"
)
self.lbl_singbox_profile_effective.setStyleSheet("color: gray;")
def _apply_singbox_profile_controls(self) -> None:
self.cmb_singbox_profile_routing.setEnabled(
not self.chk_singbox_profile_use_global_routing.isChecked()
)
self.cmb_singbox_profile_dns.setEnabled(
not self.chk_singbox_profile_use_global_dns.isChecked()
)
self.cmb_singbox_profile_killswitch.setEnabled(
not self.chk_singbox_profile_use_global_killswitch.isChecked()
)
self._refresh_singbox_profile_effective()
def _apply_singbox_compact_visibility(self) -> None:
show_profile = bool(self.btn_singbox_toggle_profile_settings.isChecked())
self.grp_singbox_profile_settings.setVisible(show_profile)
self.btn_singbox_toggle_profile_settings.setText(
"Hide profile settings" if show_profile else "Profile settings"
)
show_global = bool(self.btn_singbox_toggle_global_defaults.isChecked())
self.grp_singbox_global_defaults.setVisible(show_global)
self.btn_singbox_toggle_global_defaults.setText(
"Hide global defaults" if show_global else "Global defaults"
)
show_log = bool(self.btn_singbox_toggle_activity.isChecked())
self.grp_singbox_activity.setVisible(show_log)
self.btn_singbox_toggle_activity.setText(
"Hide activity log" if show_log else "Activity log"
)

View File

@@ -0,0 +1,16 @@
from __future__ import annotations
from main_window.singbox.runtime_cards_mixin import SingBoxRuntimeCardsMixin
from main_window.singbox.runtime_profiles_mixin import SingBoxRuntimeProfilesMixin
from main_window.singbox.runtime_transport_mixin import SingBoxRuntimeTransportMixin
class SingBoxRuntimeMixin(
SingBoxRuntimeProfilesMixin,
SingBoxRuntimeTransportMixin,
SingBoxRuntimeCardsMixin,
):
"""Facade mixin for SingBox runtime/profile actions."""
__all__ = ["SingBoxRuntimeMixin"]

View File

@@ -0,0 +1,428 @@
from __future__ import annotations
from typing import Literal
from PySide6.QtCore import Qt
from PySide6.QtWidgets import (
QApplication,
QDialog,
QHBoxLayout,
QLabel,
QMessageBox,
QPushButton,
QVBoxLayout,
)
class SingBoxRuntimeProfilesMixin:
def on_singbox_profile_edit_dialog(self, cid: str = "") -> None:
def work():
target = str(cid or "").strip() or self._selected_transport_engine_id()
if not target:
raise RuntimeError("Select a transport engine first")
if not self._select_transport_engine_by_id(target):
raise RuntimeError(f"Transport engine '{target}' not found")
self._sync_selected_singbox_profile_link(silent=True)
self._load_singbox_editor_for_selected(silent=False)
client = self._selected_transport_client()
pid = self._selected_singbox_profile_id()
if client is None or not pid:
raise RuntimeError("Select a SingBox profile first")
profile_name = self.ent_singbox_proto_name.text().strip() or str(getattr(client, "name", "") or pid).strip()
host_layout = self.grp_singbox_profile_settings.layout()
if host_layout is None:
raise RuntimeError("internal layout is unavailable")
editor = self.grp_singbox_proto_editor
insert_at = host_layout.indexOf(editor)
if insert_at >= 0:
host_layout.removeWidget(editor)
moved = False
dlg = QDialog(self)
dlg.setModal(True)
dlg.setWindowTitle(f"Edit SingBox profile: {profile_name}")
dlg.resize(860, 680)
dlg_layout = QVBoxLayout(dlg)
try:
hint = QLabel("Edit protocol fields and save draft. Use profile card menu for Run/Delete.")
hint.setStyleSheet("color: gray;")
dlg_layout.addWidget(hint)
editor.setTitle(f"{self._singbox_editor_default_title} · {profile_name}")
editor.setParent(dlg)
editor.setVisible(True)
moved = True
dlg_layout.addWidget(editor, stretch=1)
actions = QHBoxLayout()
btn_save = QPushButton("Save draft")
btn_close = QPushButton("Close")
actions.addWidget(btn_save)
actions.addStretch(1)
actions.addWidget(btn_close)
dlg_layout.addLayout(actions)
def save_draft_clicked() -> None:
try:
selected_client, _eid, selected_pid = self._selected_singbox_profile_context()
saved = self._save_singbox_editor_draft(selected_client, profile_id=selected_pid)
line = (saved.pretty_text or "").strip() or f"save profile {selected_pid}"
self._append_transport_log(f"[profile] {line}")
self.ctrl.log_gui(f"[singbox-profile] {line}")
self.lbl_transport_engine_meta.setText(f"Engine: profile {selected_pid} draft saved")
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
self._render_singbox_profile_cards()
self._sync_singbox_profile_card_selection(self._selected_transport_engine_id())
QMessageBox.information(dlg, "SingBox profile", line)
except Exception as e:
QMessageBox.critical(dlg, "SingBox profile save error", str(e))
btn_save.clicked.connect(save_draft_clicked)
btn_close.clicked.connect(dlg.accept)
dlg.exec()
finally:
if moved:
dlg_layout.removeWidget(editor)
editor.setParent(self.grp_singbox_profile_settings)
editor.setTitle(self._singbox_editor_default_title)
if insert_at >= 0:
host_layout.insertWidget(insert_at, editor)
else:
host_layout.addWidget(editor)
editor.setVisible(False)
self._safe(work, title="SingBox profile edit error")
def on_transport_engine_action(
self,
action: Literal["provision", "start", "stop", "restart"],
) -> None:
def work():
cid = self._selected_transport_engine_id()
if not cid:
raise RuntimeError("Select a transport engine first")
self.lbl_transport_engine_meta.setText(f"Engine: {action} {cid}...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
if action == "start":
selected_client = self._selected_transport_client()
if selected_client is not None and str(getattr(selected_client, "kind", "") or "").strip().lower() == "singbox":
_client, _eid, pid = self._selected_singbox_profile_context()
self.lbl_transport_engine_meta.setText(f"Engine: preparing profile {pid} for start...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
pre = self.ctrl.singbox_profile_apply_action(
pid,
client_id=cid,
restart=False,
skip_runtime=True,
check_binary=True,
client=selected_client,
)
pre_line = (pre.pretty_text or "").strip() or f"apply profile {pid}"
self._append_transport_log(f"[profile] {pre_line}")
self.ctrl.log_gui(f"[singbox-profile] {pre_line}")
if not pre.ok:
raise RuntimeError(f"profile preflight failed: {pre_line}")
ok, msg = self._apply_transport_switch_policy(cid)
self._append_transport_log(f"[switch] {msg}")
self.ctrl.log_gui(f"[transport-switch] {msg}")
if not ok:
if "canceled by user" in msg.lower():
self.refresh_transport_engines(silent=True)
return
raise RuntimeError(msg)
res = self.ctrl.transport_client_action(cid, action if action != "start" else "start")
line = (res.pretty_text or "").strip() or f"{action} {cid}"
self._append_transport_log(f"[engine] {line}")
self.ctrl.log_gui(f"[transport-engine] {line}")
if not res.ok:
raise RuntimeError(line)
self.refresh_transport_engines(silent=True)
self.refresh_status_tab()
self._safe(work, title="Transport engine error")
def on_transport_engine_delete(self, cid: str = "") -> None:
def work():
target = str(cid or "").strip() or self._selected_transport_engine_id()
if not target:
raise RuntimeError("Select a transport engine first")
if not self._select_transport_engine_by_id(target):
raise RuntimeError(f"Transport engine '{target}' not found")
ans = QMessageBox.question(
self,
"Delete transport profile",
(
f"Delete profile '{target}'?\n\n"
"The client configuration and related runtime artifacts will be removed."
),
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No,
)
if ans != QMessageBox.Yes:
return
self.lbl_transport_engine_meta.setText(f"Engine: deleting {target}...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
res = self.ctrl.transport_client_delete_action(target, force=False, cleanup=True)
if not res.ok and "force=true" in (res.pretty_text or "").lower():
force_ans = QMessageBox.question(
self,
"Profile is referenced",
(
"This profile is referenced by current transport policy.\n"
"Force delete anyway?"
),
QMessageBox.Yes | QMessageBox.No,
QMessageBox.No,
)
if force_ans == QMessageBox.Yes:
res = self.ctrl.transport_client_delete_action(target, force=True, cleanup=True)
else:
self._append_transport_log(f"[engine] delete {target}: canceled by user")
self.ctrl.log_gui(f"[transport-engine] delete {target}: canceled by user")
return
line = (res.pretty_text or "").strip() or f"delete {target}"
self._append_transport_log(f"[engine] {line}")
self.ctrl.log_gui(f"[transport-engine] {line}")
if not res.ok:
raise RuntimeError(line)
self.refresh_transport_engines(silent=True)
self.refresh_status_tab()
self._safe(work, title="Transport engine delete error")
def on_transport_policy_rollback(self) -> None:
def work():
self.lbl_transport_engine_meta.setText("Engine: rollback policy...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
res = self.ctrl.transport_policy_rollback_action()
line = (res.pretty_text or "").strip() or "policy rollback"
self._append_transport_log(f"[switch] {line}")
self.ctrl.log_gui(f"[transport-switch] {line}")
if not res.ok:
raise RuntimeError(line)
self.refresh_transport_engines(silent=True)
self.refresh_status_tab()
self._safe(work, title="Transport rollback error")
def on_toggle_singbox_profile_settings(self, checked: bool = False) -> None:
if checked and self.btn_singbox_toggle_global_defaults.isChecked():
self.btn_singbox_toggle_global_defaults.setChecked(False)
self._apply_singbox_compact_visibility()
self._save_ui_preferences()
def on_toggle_singbox_global_defaults(self, checked: bool = False) -> None:
if checked and self.btn_singbox_toggle_profile_settings.isChecked():
self.btn_singbox_toggle_profile_settings.setChecked(False)
self._apply_singbox_compact_visibility()
self._save_ui_preferences()
def on_toggle_singbox_activity(self, _checked: bool = False) -> None:
self._apply_singbox_compact_visibility()
self._save_ui_preferences()
def on_singbox_profile_scope_changed(self, _state: int = 0) -> None:
self._apply_singbox_profile_controls()
self._save_ui_preferences()
self._update_transport_engine_view()
def on_singbox_global_defaults_changed(self, _index: int = 0) -> None:
self._refresh_singbox_profile_effective()
self._save_ui_preferences()
self._update_transport_engine_view()
def on_singbox_global_save(self) -> None:
def work():
self._save_ui_preferences()
route, dns, killswitch = self._effective_singbox_policy()
msg = (
"Global defaults saved: "
f"routing={self._singbox_value_label('routing', route)}, "
f"dns={self._singbox_value_label('dns', dns)}, "
f"kill-switch={self._singbox_value_label('killswitch', killswitch)}"
)
self._append_transport_log(f"[profile] {msg}")
self.ctrl.log_gui(f"[singbox-settings] {msg}")
self._safe(work, title="SingBox settings error")
def on_singbox_profile_save(self) -> None:
def work():
client, eid, pid = self._selected_singbox_profile_context()
self._save_ui_preferences()
self.lbl_transport_engine_meta.setText(f"Engine: saving draft for {pid}...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
saved = self._save_singbox_editor_draft(client, profile_id=pid)
save_line = (saved.pretty_text or "").strip() or f"save profile {pid}"
self._append_transport_log(f"[profile] {save_line}")
self.ctrl.log_gui(f"[singbox-profile] {save_line}")
route, dns, killswitch = self._effective_singbox_policy()
msg = (
f"profile settings saved for {eid}: "
f"routing={self._singbox_value_label('routing', route)}, "
f"dns={self._singbox_value_label('dns', dns)}, "
f"kill-switch={self._singbox_value_label('killswitch', killswitch)}"
)
self._append_transport_log(f"[profile] {msg}")
self.ctrl.log_gui(f"[singbox-profile] {msg}")
self.lbl_transport_engine_meta.setText(f"Engine: profile {pid} draft saved")
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
self.refresh_transport_engines(silent=True)
self._safe(work, title="SingBox profile save error")
def _selected_singbox_profile_context(self):
client = self._selected_transport_client()
eid = self._selected_transport_engine_id()
pid = self._selected_singbox_profile_id()
if not eid or client is None:
raise RuntimeError("Select a transport engine first")
if not pid:
raise RuntimeError("Select a SingBox profile first")
return client, eid, pid
def _run_singbox_profile_action(
self,
*,
verb: str,
runner,
refresh_status: bool = False,
sync_draft: bool = False,
) -> None:
client, eid, pid = self._selected_singbox_profile_context()
if sync_draft:
self.lbl_transport_engine_meta.setText(f"Engine: syncing draft for {pid}...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
saved = self._save_singbox_editor_draft(client, profile_id=pid)
save_line = (saved.pretty_text or "").strip() or f"save profile {pid}"
self._append_transport_log(f"[profile] {save_line}")
self.ctrl.log_gui(f"[singbox-profile] {save_line}")
self.lbl_transport_engine_meta.setText(f"Engine: {verb} profile {pid}...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
res = runner(client, eid, pid)
line = (res.pretty_text or "").strip() or f"{verb} profile {pid}"
self._append_transport_log(f"[profile] {line}")
self.ctrl.log_gui(f"[singbox-profile] {line}")
if res.ok:
self.lbl_transport_engine_meta.setText(f"Engine: profile {pid} {verb} done")
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
else:
self.lbl_transport_engine_meta.setText(f"Engine: profile {pid} {verb} failed")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
self.refresh_transport_engines(silent=True)
if refresh_status:
self.refresh_status_tab()
def on_singbox_profile_preview(self) -> None:
self._safe(
lambda: self._run_singbox_profile_action(
verb="previewing",
runner=lambda client, _eid, pid: self.ctrl.singbox_profile_render_preview_action(
pid,
check_binary=None,
persist=False,
client=client,
),
refresh_status=False,
sync_draft=True,
),
title="SingBox profile preview error",
)
def on_singbox_profile_validate(self) -> None:
self._safe(
lambda: self._run_singbox_profile_action(
verb="validating",
runner=lambda client, _eid, pid: self.ctrl.singbox_profile_validate_action(
pid,
client=client,
),
refresh_status=False,
sync_draft=True,
),
title="SingBox profile validate error",
)
def on_singbox_profile_apply(self) -> None:
self._safe(
lambda: self._run_singbox_profile_action(
verb="applying",
runner=lambda client, eid, pid: self.ctrl.singbox_profile_apply_action(
pid,
client_id=eid,
restart=True,
skip_runtime=False,
client=client,
),
refresh_status=True,
sync_draft=True,
),
title="SingBox profile apply error",
)
def on_singbox_profile_rollback(self) -> None:
self._safe(
lambda: self._run_singbox_profile_action(
verb="rolling back",
runner=lambda client, eid, pid: self.ctrl.singbox_profile_rollback_action(
pid,
client_id=eid,
restart=True,
skip_runtime=False,
client=client,
),
refresh_status=True,
),
title="SingBox profile rollback error",
)
def on_singbox_profile_history(self) -> None:
def work():
client, _eid, pid = self._selected_singbox_profile_context()
self.lbl_transport_engine_meta.setText(f"Engine: loading history for {pid}...")
self.lbl_transport_engine_meta.setStyleSheet("color: orange;")
QApplication.processEvents()
lines = self.ctrl.singbox_profile_history_lines(pid, limit=20, client=client)
if not lines:
line = f"history profile {pid}: no entries"
self._append_transport_log(f"[history] {line}")
self.ctrl.log_gui(f"[singbox-profile] {line}")
self.lbl_transport_engine_meta.setText(f"Engine: history {pid} is empty")
self.lbl_transport_engine_meta.setStyleSheet("color: gray;")
return
header = f"history profile {pid}: {len(lines)} entries"
self._append_transport_log(f"[history] {header}")
self.ctrl.log_gui(f"[singbox-profile] {header}")
for ln in lines:
self._append_transport_log(f"[history] {ln}")
self.lbl_transport_engine_meta.setText(f"Engine: history loaded for {pid}")
self.lbl_transport_engine_meta.setStyleSheet("color: green;")
self._safe(work, title="SingBox profile history error")

File diff suppressed because it is too large Load Diff