269 lines
11 KiB
Python
269 lines
11 KiB
Python
from __future__ import annotations
|
|
|
|
from typing import Literal
|
|
|
|
from PySide6.QtWidgets import QApplication, QMessageBox
|
|
|
|
from dns_benchmark_dialog import DNSBenchmarkDialog
|
|
from main_window.constants import LOCATION_TARGET_ROLE
|
|
from traffic_mode_dialog import TrafficModeDialog
|
|
|
|
|
|
class RuntimeOpsMixin:
|
|
def on_toggle_autoconnect(self) -> None:
|
|
def work():
|
|
current = self.ctrl.vpn_autoconnect_enabled()
|
|
enable = not current
|
|
self.ctrl.vpn_set_autoconnect(enable)
|
|
self.ctrl.log_gui(f"VPN autoconnect set to {enable}")
|
|
self.refresh_vpn_tab()
|
|
self._safe(work, title="Autoconnect error")
|
|
|
|
def on_location_activated(self, _index: int) -> None:
|
|
self._safe(self._apply_selected_location, title="Location error")
|
|
|
|
def on_set_location(self) -> None:
|
|
self._safe(self._apply_selected_location, title="Location error")
|
|
|
|
def _apply_selected_location(self) -> None:
|
|
idx = self.cmb_locations.currentIndex()
|
|
if idx < 0:
|
|
return
|
|
|
|
iso = str(self.cmb_locations.currentData() or "").strip().upper()
|
|
target = str(self.cmb_locations.currentData(LOCATION_TARGET_ROLE) or "").strip()
|
|
label = str(self.cmb_locations.currentText() or "").strip()
|
|
if not target:
|
|
target = iso
|
|
if not iso or not target:
|
|
return
|
|
|
|
desired = (self._vpn_desired_location or "").strip().lower()
|
|
if desired and desired in (iso.lower(), target.lower()):
|
|
return
|
|
|
|
self.lbl_locations_meta.setText(f"Applying location {target}...")
|
|
self.lbl_locations_meta.setStyleSheet("color: orange;")
|
|
|
|
self._start_vpn_location_switching(target)
|
|
self.refresh_login_banner()
|
|
QApplication.processEvents()
|
|
|
|
try:
|
|
self.ctrl.vpn_set_location(target=target, iso=iso, label=label)
|
|
except Exception:
|
|
self._stop_vpn_location_switching()
|
|
self.refresh_login_banner()
|
|
raise
|
|
|
|
self.ctrl.log_gui(f"VPN location set to {target} ({iso})")
|
|
self._vpn_desired_location = target
|
|
self.refresh_vpn_tab()
|
|
self._trigger_vpn_egress_refresh(reason=f"location switch to {target}")
|
|
|
|
# ---- Routes actions ------------------------------------------------
|
|
|
|
def on_routes_action(
|
|
self, action: Literal["start", "stop", "restart"]
|
|
) -> None:
|
|
def work():
|
|
res = self.ctrl.routes_service_action(action)
|
|
self._set_text(self.txt_routes, res.pretty_text or str(res))
|
|
self.refresh_status_tab()
|
|
self._safe(work, title="Routes error")
|
|
|
|
def _append_routes_log(self, msg: str) -> None:
|
|
line = (msg or "").strip()
|
|
if not line:
|
|
return
|
|
self._append_text(self.txt_routes, line + "\n")
|
|
self.ctrl.log_gui(line)
|
|
|
|
def on_open_traffic_settings(self) -> None:
|
|
def work():
|
|
def refresh_all_traffic() -> None:
|
|
self.refresh_routes_tab()
|
|
self.refresh_status_tab()
|
|
|
|
dlg = TrafficModeDialog(
|
|
self.ctrl,
|
|
log_cb=self._append_routes_log,
|
|
refresh_cb=refresh_all_traffic,
|
|
parent=self,
|
|
)
|
|
dlg.exec()
|
|
refresh_all_traffic()
|
|
self._safe(work, title="Traffic mode dialog error")
|
|
|
|
def on_test_traffic_mode(self) -> None:
|
|
def work():
|
|
view = self.ctrl.traffic_mode_test()
|
|
msg = (
|
|
f"Traffic mode test: desired={view.desired_mode}, applied={view.applied_mode}, "
|
|
f"iface={view.active_iface or '-'}, probe_ok={view.probe_ok}, "
|
|
f"healthy={view.healthy}, auto_local_bypass={view.auto_local_bypass}, "
|
|
f"bypass_routes={view.bypass_candidates}, overrides={view.overrides_applied}, "
|
|
f"cgroup_uids={view.cgroup_resolved_uids}, cgroup_warning={view.cgroup_warning or '-'}, "
|
|
f"message={view.message}, probe={view.probe_message}"
|
|
)
|
|
self._append_routes_log(msg)
|
|
self.refresh_routes_tab()
|
|
self.refresh_status_tab()
|
|
self._safe(work, title="Traffic mode test error")
|
|
|
|
def on_routes_precheck_debug(self) -> None:
|
|
def work():
|
|
res = self.ctrl.routes_precheck_debug(run_now=True)
|
|
txt = (res.pretty_text or "").strip()
|
|
if res.ok:
|
|
QMessageBox.information(self, "Resolve precheck debug", txt or "OK")
|
|
else:
|
|
QMessageBox.critical(self, "Resolve precheck debug", txt or "ERROR")
|
|
self.refresh_routes_tab()
|
|
self.refresh_status_tab()
|
|
self.refresh_trace_tab()
|
|
self._safe(work, title="Resolve precheck debug error")
|
|
|
|
def on_toggle_timer(self) -> None:
|
|
def work():
|
|
enabled = self.chk_timer.isChecked()
|
|
res = self.ctrl.routes_timer_set(enabled)
|
|
self.ctrl.log_gui(f"Routes timer set to {enabled}")
|
|
self._set_text(self.txt_routes, res.pretty_text or str(res))
|
|
self.refresh_routes_tab()
|
|
self._safe(work, title="Timer error")
|
|
|
|
def on_fix_policy_route(self) -> None:
|
|
def work():
|
|
res = self.ctrl.routes_fix_policy_route()
|
|
self._set_text(self.txt_routes, res.pretty_text or str(res))
|
|
self.refresh_status_tab()
|
|
self._safe(work, title="Policy route error")
|
|
|
|
# ---- DNS actions ---------------------------------------------------
|
|
|
|
def _schedule_dns_autosave(self, _text: str = "") -> None:
|
|
if self._dns_ui_refresh:
|
|
return
|
|
self.dns_save_timer.start()
|
|
|
|
def _apply_dns_autosave(self) -> None:
|
|
def work():
|
|
if self._dns_ui_refresh:
|
|
return
|
|
self.ctrl.dns_mode_set(
|
|
self.chk_dns_via_smartdns.isChecked(),
|
|
self.ent_smartdns_addr.text().strip(),
|
|
)
|
|
self.ctrl.log_gui("DNS settings autosaved")
|
|
self._safe(work, title="DNS save error")
|
|
|
|
def on_open_dns_benchmark(self) -> None:
|
|
def work():
|
|
dlg = DNSBenchmarkDialog(
|
|
self.ctrl,
|
|
settings=self._ui_settings,
|
|
refresh_cb=self.refresh_dns_tab,
|
|
parent=self,
|
|
)
|
|
dlg.exec()
|
|
self.refresh_dns_tab()
|
|
self._safe(work, title="DNS benchmark error")
|
|
|
|
def on_dns_mode_toggle(self) -> None:
|
|
def work():
|
|
via = self.chk_dns_via_smartdns.isChecked()
|
|
self.ctrl.dns_mode_set(via, self.ent_smartdns_addr.text().strip())
|
|
mode = "hybrid_wildcard" if via else "direct"
|
|
self.ctrl.log_gui(f"DNS mode changed: mode={mode}")
|
|
self.refresh_dns_tab()
|
|
self._safe(work, title="DNS mode error")
|
|
|
|
def on_smartdns_unit_toggle(self) -> None:
|
|
def work():
|
|
enable = self.chk_dns_unit_relay.isChecked()
|
|
action = "start" if enable else "stop"
|
|
self.ctrl.smartdns_service_action(action)
|
|
self.ctrl.log_smartdns(f"SmartDNS unit action from GUI: {action}")
|
|
self.refresh_dns_tab()
|
|
self.refresh_status_tab()
|
|
self._safe(work, title="SmartDNS error")
|
|
|
|
def on_smartdns_runtime_toggle(self) -> None:
|
|
def work():
|
|
if self._dns_ui_refresh:
|
|
return
|
|
enable = self.chk_dns_runtime_nftset.isChecked()
|
|
st = self.ctrl.smartdns_runtime_set(enabled=enable, restart=True)
|
|
self.ctrl.log_smartdns(
|
|
f"SmartDNS runtime accelerator set from GUI: enabled={enable} changed={st.changed} restarted={st.restarted} source={st.wildcard_source}"
|
|
)
|
|
self.refresh_dns_tab()
|
|
self.refresh_trace_tab()
|
|
self._safe(work, title="SmartDNS runtime error")
|
|
|
|
def on_smartdns_prewarm(self) -> None:
|
|
def work():
|
|
aggressive = bool(self.chk_routes_prewarm_aggressive.isChecked())
|
|
result = self.ctrl.smartdns_prewarm(aggressive_subs=aggressive)
|
|
mode_txt = "aggressive_subs=on" if aggressive else "aggressive_subs=off"
|
|
self.ctrl.log_smartdns(f"SmartDNS prewarm requested from GUI: {mode_txt}")
|
|
txt = (result.pretty_text or "").strip()
|
|
if result.ok:
|
|
QMessageBox.information(self, "SmartDNS prewarm", txt or "OK")
|
|
else:
|
|
QMessageBox.critical(self, "SmartDNS prewarm", txt or "ERROR")
|
|
self.refresh_trace_tab()
|
|
self._safe(work, title="SmartDNS prewarm error")
|
|
|
|
def _update_prewarm_mode_label(self, _state: int = 0) -> None:
|
|
aggressive = bool(self.chk_routes_prewarm_aggressive.isChecked())
|
|
if aggressive:
|
|
self.lbl_routes_prewarm_mode.setText("Prewarm mode: aggressive (subs enabled)")
|
|
self.lbl_routes_prewarm_mode.setStyleSheet("color: orange;")
|
|
else:
|
|
self.lbl_routes_prewarm_mode.setText("Prewarm mode: wildcard-only")
|
|
self.lbl_routes_prewarm_mode.setStyleSheet("color: gray;")
|
|
|
|
def _on_prewarm_aggressive_changed(self, _state: int = 0) -> None:
|
|
self._update_prewarm_mode_label(_state)
|
|
self._save_ui_preferences()
|
|
|
|
# ---- Domains actions -----------------------------------------------
|
|
|
|
def on_domains_load(self) -> None:
|
|
def work():
|
|
name = self._get_selected_domains_file()
|
|
content, source, path = self._load_file_content(name)
|
|
is_readonly = name in ("last-ips-map-direct", "last-ips-map-wildcard", "wildcard-observed-hosts")
|
|
self.txt_domains.setReadOnly(is_readonly)
|
|
self.btn_domains_save.setEnabled(not is_readonly)
|
|
self._set_text(self.txt_domains, content)
|
|
ro = "read-only" if is_readonly else "editable"
|
|
self.lbl_domains_info.setText(f"{name} ({source}, {ro}) [{path}]")
|
|
self._safe(work, title="Domains load error")
|
|
|
|
def on_domains_save(self) -> None:
|
|
def work():
|
|
name = self._get_selected_domains_file()
|
|
content = self.txt_domains.toPlainText()
|
|
self._save_file_content(name, content)
|
|
self.ctrl.log_gui(f"Domains file saved: {name}")
|
|
self._safe(work, title="Domains save error")
|
|
|
|
# ---- close event ---------------------------------------------------
|
|
|
|
def closeEvent(self, event) -> None: # pragma: no cover - GUI
|
|
try:
|
|
self._save_ui_preferences()
|
|
self._login_flow_autopoll_stop()
|
|
self.loc_typeahead_timer.stop()
|
|
if self.locations_thread:
|
|
self.locations_thread.quit()
|
|
self.locations_thread.wait(1500)
|
|
if self.events_thread:
|
|
self.events_thread.stop()
|
|
self.events_thread.wait(1500)
|
|
finally:
|
|
super().closeEvent(event)
|