ui: snap app picker + runtime scopes list/cleanup

selective-vpn-gui/traffic_mode_dialog.py

@@ -50,6 +50,14 @@ class DesktopAppEntry:
     source: str  # system|flatpak|user
 
 
+@dataclass(frozen=True)
+class RuntimeScopeInfo:
+    unit: str
+    target: str  # vpn|direct|?
+    cgroup_path: str
+    cgroup_id: int
+
+
 class TrafficModeDialog(QDialog):
     def __init__(
         self,
@@ -219,8 +227,8 @@ RU: Восстанавливает маршруты/nft из последнег
         row_cmd.addWidget(self.ed_app_cmd, stretch=1)
         self.btn_app_pick = QPushButton("Pick app...")
         self.btn_app_pick.setToolTip(
-            "EN: Pick an installed app from .desktop entries (system + flatpak) and fill the command.\n"
-            "RU: Выбрать установленное приложение из .desktop (system + flatpak) и заполнить команду."
+            "EN: Pick an installed app from .desktop entries (system + flatpak + snap) and fill the command.\n"
+            "RU: Выбрать установленное приложение из .desktop (system + flatpak + snap) и заполнить команду."
         )
         self.btn_app_pick.clicked.connect(self.on_app_pick)
         row_cmd.addWidget(self.btn_app_pick)
@@ -306,6 +314,53 @@ RU: Восстанавливает маршруты/nft из последнег
 
         tab_apps_layout.addWidget(run_group)
 
+        scopes_group = QGroupBox("Active svpn scopes (systemd --user)")
+        scopes_layout = QVBoxLayout(scopes_group)
+
+        scopes_row = QHBoxLayout()
+        self.btn_scopes_refresh = QPushButton("Refresh scopes")
+        self.btn_scopes_refresh.setToolTip(
+            "EN: Refresh list of running svpn-* scopes.\n"
+            "RU: Обновить список запущенных svpn-* scope."
+        )
+        self.btn_scopes_refresh.clicked.connect(self.refresh_running_scopes)
+        scopes_row.addWidget(self.btn_scopes_refresh)
+
+        self.btn_scopes_stop_selected = QPushButton("Stop selected")
+        self.btn_scopes_stop_selected.setToolTip(
+            "EN: Unmarks + stops selected scopes.\n"
+            "RU: Удаляет метки + останавливает выбранные scope."
+        )
+        self.btn_scopes_stop_selected.clicked.connect(self.on_scopes_stop_selected)
+        scopes_row.addWidget(self.btn_scopes_stop_selected)
+
+        self.btn_scopes_cleanup = QPushButton("Cleanup all svpn scopes")
+        self.btn_scopes_cleanup.setToolTip(
+            "EN: Unmarks + stops ALL running svpn-* scopes.\n"
+            "RU: Удаляет метки + останавливает ВСЕ запущенные svpn-* scope."
+        )
+        self.btn_scopes_cleanup.clicked.connect(self.on_scopes_cleanup_all)
+        scopes_row.addWidget(self.btn_scopes_cleanup)
+
+        scopes_row.addStretch(1)
+        scopes_layout.addLayout(scopes_row)
+
+        self.lst_scopes = QListWidget()
+        self.lst_scopes.setSelectionMode(QAbstractItemView.ExtendedSelection)
+        self.lst_scopes.setToolTip(
+            "EN: Running svpn scopes. Double click to copy unit name.\n"
+            "RU: Запущенные svpn scope. Двойной клик копирует имя unit."
+        )
+        self.lst_scopes.itemDoubleClicked.connect(lambda it: self._copy_scope_unit(it))
+        self.lst_scopes.setFixedHeight(140)
+        scopes_layout.addWidget(self.lst_scopes)
+
+        self.lbl_scopes = QLabel("Running scopes: —")
+        self.lbl_scopes.setStyleSheet("color: gray;")
+        scopes_layout.addWidget(self.lbl_scopes)
+
+        tab_apps_layout.addWidget(scopes_group)
+
         self.txt_app = QPlainTextEdit()
         self.txt_app.setReadOnly(True)
         tab_apps_layout.addWidget(self.txt_app, stretch=1)
@@ -411,6 +466,7 @@ RU: Применяет policy-rules и проверяет health. При оши
 
         QtCore.QTimer.singleShot(0, self.refresh_state)
         QtCore.QTimer.singleShot(0, self.refresh_appmarks_counts)
+        QtCore.QTimer.singleShot(0, self.refresh_running_scopes)
 
     def _is_operation_error(self, message: str) -> bool:
         low = (message or "").strip().lower()
@@ -1047,6 +1103,257 @@ RU: Применяет policy-rules и проверяет health. При оши
 
         self._safe(work, title="Clear app marks error")
 
+    # -----------------------------------------------------------------
+    # Scopes list + cleanup (runtime)
+    # -----------------------------------------------------------------
+
+    def _scope_target_from_unit(self, unit: str) -> str:
+        u = (unit or "").strip()
+        if not u.startswith("svpn-"):
+            return ""
+        rest = u[len("svpn-") :]
+        if rest.startswith("vpn-"):
+            return "vpn"
+        if rest.startswith("direct-"):
+            return "direct"
+        return ""
+
+    def _systemctl_user(self, args: list[str]) -> tuple[int, str]:
+        p = subprocess.run(
+            ["systemctl", "--user"] + list(args or []),
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+        out = ((p.stdout or "") + (p.stderr or "")).strip()
+        return int(p.returncode or 0), out
+
+    def _list_running_svpn_scopes(self) -> list[str]:
+        code, out = self._systemctl_user(
+            [
+                "list-units",
+                "--type=scope",
+                "--state=running",
+                "--no-legend",
+                "--no-pager",
+                "--plain",
+            ]
+        )
+        if code != 0:
+            raise RuntimeError(out or f"systemctl list-units failed: rc={code}")
+
+        units: list[str] = []
+        for raw in (out or "").splitlines():
+            line = raw.strip()
+            if not line:
+                continue
+            fields = line.split()
+            if not fields:
+                continue
+            unit = fields[0].strip()
+            if unit.startswith("svpn-") and unit.endswith(".scope"):
+                units.append(unit)
+        units.sort()
+        return units
+
+    def _control_group_for_unit(self, unit: str) -> str:
+        code, out = self._systemctl_user(["show", "-p", "ControlGroup", "--value", unit])
+        if code != 0:
+            raise RuntimeError(out or f"systemctl show failed: rc={code}")
+        return (out or "").strip()
+
+    def _cgroup_inode_id(self, cgroup_path: str) -> int:
+        cg = (cgroup_path or "").strip()
+        if not cg:
+            return 0
+        rel = cg.lstrip("/")
+        if not rel:
+            return 0
+        full = os.path.join("/sys/fs/cgroup", rel)
+        try:
+            st = os.stat(full)
+            return int(getattr(st, "st_ino", 0) or 0)
+        except Exception:
+            return 0
+
+    def refresh_running_scopes(self) -> None:
+        def work() -> None:
+            units = self._list_running_svpn_scopes()
+            self.lst_scopes.clear()
+
+            for unit in units:
+                target = self._scope_target_from_unit(unit)
+                cg = ""
+                cg_id = 0
+                try:
+                    cg = self._control_group_for_unit(unit)
+                    cg_id = self._cgroup_inode_id(cg)
+                except Exception:
+                    pass
+
+                label = unit
+                if target in ("vpn", "direct"):
+                    label += f"  [{target.upper()}]"
+
+                it = QListWidgetItem(label)
+                info = RuntimeScopeInfo(
+                    unit=unit,
+                    target=target or "?",
+                    cgroup_path=cg,
+                    cgroup_id=int(cg_id or 0),
+                )
+                it.setData(QtCore.Qt.UserRole, info)
+                it.setToolTip(
+                    f"Unit: {unit}\n"
+                    f"Target: {target or '-'}\n"
+                    f"ControlGroup: {cg or '-'}\n"
+                    f"cgroup_id: {cg_id or '-'}\n\n"
+                    "EN: Stop selected will unmark by cgroup_id (if known) and stop the scope.\n"
+                    "RU: Stop selected удалит метку по cgroup_id (если известен) и остановит scope."
+                )
+                self.lst_scopes.addItem(it)
+
+            self.lbl_scopes.setText(f"Running scopes: {len(units)}")
+            has_any = self.lst_scopes.count() > 0
+            self.btn_scopes_stop_selected.setEnabled(has_any)
+            self.btn_scopes_cleanup.setEnabled(has_any)
+
+        self._safe(work, title="Scopes refresh error")
+
+    def _copy_scope_unit(self, it: QListWidgetItem) -> None:
+        unit = ""
+        info = it.data(QtCore.Qt.UserRole) if it else None
+        if isinstance(info, RuntimeScopeInfo):
+            unit = info.unit
+        else:
+            unit = (it.text() or "").split()[0].strip() if it else ""
+        if not unit:
+            return
+        try:
+            QtGui.QGuiApplication.clipboard().setText(unit)
+        except Exception:
+            pass
+        self._append_app_log(f"[scope] copied unit: {unit}")
+        self._set_action_status(f"Copied unit: {unit}", ok=True)
+
+    def _stop_scope_unit(self, unit: str) -> None:
+        u = (unit or "").strip()
+        if not u:
+            return
+        code, out = self._systemctl_user(["stop", u])
+        if code == 0:
+            if out:
+                self._append_app_log(out)
+            return
+        code2, out2 = self._systemctl_user(["kill", u])
+        if code2 != 0:
+            raise RuntimeError(out2 or out or f"stop/kill failed: {u}")
+
+    def _unmark_scope(self, info: RuntimeScopeInfo) -> None:
+        target = (info.target or "").strip().lower()
+        cg_id = int(info.cgroup_id or 0)
+        if target not in ("vpn", "direct") or cg_id <= 0:
+            return
+        res = self.ctrl.traffic_appmarks_apply(
+            op="del",
+            target=target,
+            cgroup=str(cg_id),
+        )
+        if not res.ok:
+            raise RuntimeError(res.message or "unmark failed")
+
+    def _selected_scope_infos(self) -> list[RuntimeScopeInfo]:
+        infos: list[RuntimeScopeInfo] = []
+        for it in self.lst_scopes.selectedItems() or []:
+            info = it.data(QtCore.Qt.UserRole) if it else None
+            if isinstance(info, RuntimeScopeInfo):
+                infos.append(info)
+        return infos
+
+    def on_scopes_stop_selected(self) -> None:
+        def work() -> None:
+            infos = self._selected_scope_infos()
+            if not infos:
+                QMessageBox.information(self, "No selection", "Select one or more scopes first.")
+                return
+
+            if QMessageBox.question(
+                self,
+                "Stop selected",
+                f"Unmark + stop {len(infos)} selected scope(s)?",
+            ) != QMessageBox.StandardButton.Yes:
+                return
+
+            for info in infos:
+                self._append_app_log(
+                    f"[scope] stop: unit={info.unit} target={info.target} cgroup_id={info.cgroup_id}"
+                )
+                try:
+                    self._unmark_scope(info)
+                    self._append_app_log("[scope] unmark OK")
+                except Exception as e:
+                    self._append_app_log(f"[scope] unmark WARN: {e}")
+
+                self._stop_scope_unit(info.unit)
+                self._append_app_log("[scope] stop OK")
+
+            self.refresh_running_scopes()
+            self.refresh_appmarks_counts()
+            self._set_action_status(f"Stopped scopes: {len(infos)}", ok=True)
+
+        self._safe(work, title="Stop selected scopes error")
+
+    def on_scopes_cleanup_all(self) -> None:
+        def work() -> None:
+            units = self._list_running_svpn_scopes()
+            if not units:
+                self._set_action_status("No running svpn scopes", ok=True)
+                self.refresh_running_scopes()
+                return
+
+            if QMessageBox.question(
+                self,
+                "Cleanup all",
+                f"Unmark + stop ALL running svpn scopes ({len(units)})?",
+            ) != QMessageBox.StandardButton.Yes:
+                return
+
+            stopped = 0
+            for unit in units:
+                target = self._scope_target_from_unit(unit)
+                cg = ""
+                cg_id = 0
+                try:
+                    cg = self._control_group_for_unit(unit)
+                    cg_id = self._cgroup_inode_id(cg)
+                except Exception:
+                    pass
+                info = RuntimeScopeInfo(
+                    unit=unit,
+                    target=target or "?",
+                    cgroup_path=cg,
+                    cgroup_id=int(cg_id or 0),
+                )
+                self._append_app_log(
+                    f"[scope] cleanup: unit={info.unit} target={info.target} cgroup_id={info.cgroup_id}"
+                )
+                try:
+                    self._unmark_scope(info)
+                    self._append_app_log("[scope] unmark OK")
+                except Exception as e:
+                    self._append_app_log(f"[scope] unmark WARN: {e}")
+                try:
+                    self._stop_scope_unit(info.unit)
+                    stopped += 1
+                except Exception as e:
+                    self._append_app_log(f"[scope] stop ERROR: {e}")
+
+            self.refresh_running_scopes()
+            self.refresh_appmarks_counts()
+            self._set_action_status(f"Cleanup done: stopped={stopped}/{len(units)}", ok=True)
+
+        self._safe(work, title="Cleanup scopes error")
+
     def on_rollback(self) -> None:
         def work() -> None:
             res = self.ctrl.routes_clear()
@@ -1754,6 +2061,7 @@ def _scan_desktop_entries() -> list[DesktopAppEntry]:
         ("/usr/share/applications", "system"),
         (os.path.join(home, ".local/share/flatpak/exports/share/applications"), "flatpak"),
         ("/var/lib/flatpak/exports/share/applications", "flatpak"),
+        ("/var/lib/snapd/desktop/applications", "snap"),
     ]
 
     seen: set[tuple[str, str]] = set()
@@ -1791,9 +2099,9 @@ class AppPickerDialog(QDialog):
         root = QVBoxLayout(self)
 
         note = QLabel(
-            "EN: Pick an installed GUI app from .desktop entries (system + flatpak). "
+            "EN: Pick an installed GUI app from .desktop entries (system + flatpak + snap). "
             "The command will be filled without %u/%U/%f placeholders.\n"
-            "RU: Выбери приложение из .desktop (system + flatpak). "
+            "RU: Выбери приложение из .desktop (system + flatpak + snap). "
             "Команда будет заполнена без плейсхолдеров %u/%U/%f."
         )
         note.setWordWrap(True)