Add location_country_code/city_code to operator proxy flow + SingboxProcessMonitor ABC + observer_helpers

core/models/OperatorProxySession.py

@@ -18,3 +18,5 @@ class OperatorProxySession:
     operator_hysteria2_host: Optional[str] = None
     operator_vless_host: Optional[str] = None
     server_ip: Optional[str] = None        # pre-resolved IP — avoids DNS leak at connect time
+    location_country_code: Optional[str] = None
+    location_city_code: Optional[str] = None

core/models/system/SystemProfile.py

@@ -79,6 +79,23 @@ class SystemProfile(BaseProfile):
         operator_proxy_session_file_path = self.get_operator_proxy_session_path()
         with open(operator_proxy_session_file_path, 'w') as operator_proxy_session_file:
             operator_proxy_session_file.write(operator_proxy_session_file_contents)
+        if operator_proxy_session.location_country_code and operator_proxy_session.location_city_code:
+            try:
+                from core.models.orm_models.Location import Location
+                from core.models.manage.session_management import get_session
+                from sqlalchemy import select
+                session = get_session()
+                loc = session.execute(
+                    select(Location).where(
+                        (Location.country_code == operator_proxy_session.location_country_code) &
+                        (Location.code == operator_proxy_session.location_city_code)
+                    )
+                ).scalar_one_or_none()
+                if loc:
+                    self.location = loc
+                    self.save()
+            except Exception:
+                pass
 
     def get_operator_proxy_session_path(self):
         return f'{self.get_config_path()}/operator_proxy_session.json'

core/services/WebServiceApiService.py

@@ -181,6 +181,8 @@ class WebServiceApiService:
                 data['operator'].get('hysteria2_host'),
                 data['operator'].get('vless_host'),
                 data.get('server_ip'),
+                data.get('location_country_code'),
+                data.get('location_city_code'),
             )
 
         return None

core/services/encrypted_proxy/observer_helpers.py

@@ -0,0 +1,185 @@
+from core.errors.logger import logger
+from core.utils.encrypted_proxy import killswitch
+
+from typing import Optional, Dict, Any
+from dataclasses import dataclass
+
+
+@dataclass
+class ConnectionStateData:
+    tunnel_interface: str = "?"
+    server_ip: Optional[str] = None
+    socks5_port: str = "?"
+    dns_enabled: bool = True
+    dns_active: bool = False
+    session_token: Optional[str] = None
+
+
+def create_ui_state() -> Dict[str, Any]:
+    return {
+        'spinner':         None,
+        'monitor':         None,
+        'timer_state':     None,
+        'connection_data': ConnectionStateData(),
+    }
+
+
+# ── Cleanup helpers ───────────────────────────────────────────────────────────
+
+def cleanup_spinner(state: Dict[str, Any]) -> None:
+    if state['spinner']:
+        try:
+            state['spinner'].stop()
+        except Exception as e:
+            logger.error(f"Failed to stop spinner: {e}")
+        finally:
+            state['spinner'] = None
+
+
+def cleanup_monitor(state: Dict[str, Any]) -> None:
+    if state['monitor'] is not None:
+        try:
+            state['monitor'].stop()
+        except Exception as e:
+            logger.error(f"Failed to stop monitor: {e}")
+        finally:
+            state['monitor'] = None
+
+
+def cleanup_timer(state: Dict[str, Any]) -> None:
+    if state['timer_state'] and state['timer_state'].get('stop'):
+        try:
+            state['timer_state']['stop'].set()
+        except Exception as e:
+            logger.error(f"Failed to stop timer: {e}")
+        finally:
+            if state['timer_state']:
+                state['timer_state']['stop'] = None
+
+
+def cleanup_reconnecting_spinner(state: Dict[str, Any]) -> None:
+    if state['timer_state'] and state['timer_state'].get('reconnecting'):
+        try:
+            state['timer_state']['reconnecting'].set()
+        except Exception as e:
+            logger.error(f"Failed to stop reconnecting spinner: {e}")
+        finally:
+            if state['timer_state']:
+                state['timer_state']['reconnecting'] = None
+
+
+def cleanup_all(state: Dict[str, Any]) -> None:
+    cleanup_spinner(state)
+    cleanup_timer(state)
+    cleanup_reconnecting_spinner(state)
+    cleanup_monitor(state)
+
+
+# ── Handler factory ───────────────────────────────────────────────────────────
+
+def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: Dict):
+    """Factory: returns event handler closures bound to state."""
+
+    def _update_connection_data(event) -> ConnectionStateData:
+        d = event.subject or {}
+        return ConnectionStateData(
+            tunnel_interface=d.get('tunnel_if', "?"),
+            server_ip=d.get('server_ip'),
+            socks5_port=d.get('socks5_port', "?"),
+            dns_enabled=d.get('dns_enabled', True),
+            dns_active=d.get('dns_active', False),
+        )
+
+    def _update_ui_labels(conn: ConnectionStateData) -> None:
+        from cli.ui import (
+            label_connected, label_killswitch, label_dns, label_ipv6, label_ipv4
+        )
+        label_connected(f"tunnel={conn.tunnel_interface}  port={conn.socks5_port}")
+        label_killswitch(killswitch.status())
+        label_dns(enabled=conn.dns_enabled, active=conn.dns_active)
+        label_ipv6(blocked=True)
+        if conn.server_ip:
+            label_ipv4(server_ip=conn.server_ip, reachable=True)
+
+    def on_connecting(event) -> None:
+        try:
+            cleanup_timer(state)
+            if (state['timer_state'] and
+                    state['timer_state'].get('reconnecting') is None and
+                    state['timer_state'].get('_retrying')):
+                from cli.ui import connecting_spinner
+                state['timer_state']['reconnecting'] = connecting_spinner()
+
+            d = event.subject or {}
+            attempt = d.get("attempt_count", "?")
+            total   = d.get("maximum_number_of_attempts", "?")
+
+            cleanup_spinner(state)
+            from cli.ui import Spinner
+            state['spinner'] = Spinner(f"Connecting... attempt {attempt}/{total}")
+            state['spinner'].start()
+        except Exception as e:
+            logger.error(f"Error in on_connecting: {e}", exc_info=True)
+
+    def on_connected(event) -> None:
+        try:
+            cleanup_spinner(state)
+            cleanup_reconnecting_spinner(state)
+
+            if state['timer_state']:
+                state['timer_state']['_retrying'] = False
+
+            state['connection_data'] = _update_connection_data(event)
+            _update_ui_labels(state['connection_data'])
+
+            d = event.subject or {}
+            if state['monitor'] is not None:
+                state['monitor'].set_connection_data(socks5_port=d.get('socks5_port'))
+
+            if state['timer_state'] is not None:
+                from cli.ui import session_timer
+                state['timer_state']['stop'] = session_timer(
+                    drop_state=drop_state,
+                    tunnel_state=tunnel_state,
+                )
+        except Exception as e:
+            logger.error(f"Error in on_connected: {e}", exc_info=True)
+
+    def on_connected_token(event) -> None:
+        try:
+            if state['monitor'] is not None:
+                token = (event.subject or {}).get('session_token')
+                if token:
+                    state['monitor'].set_session_token(token)
+                    state['connection_data'].session_token = token
+        except Exception as e:
+            logger.error(f"Error in on_connected_token: {e}", exc_info=True)
+
+    def on_disconnected(event) -> None:
+        try:
+            cleanup_all(state)
+            from cli.ui import label_disconnect
+            tunnel_if = (event.subject or {}).get('tunnel_if', '?')
+            label_disconnect(f"tunnel={tunnel_if}")
+        except Exception as e:
+            logger.error(f"Error in on_disconnected: {e}", exc_info=True)
+
+    def on_error(event) -> None:
+        try:
+            cleanup_all(state)
+            from cli.ui import label_error
+            msg = event.subject
+            if isinstance(msg, dict):
+                msg = msg.get('message', str(msg))
+            label_error(str(msg or "Unknown error"))
+            logger.error(f"Connection error: {msg}")
+        except Exception as e:
+            logger.error(f"Error in on_error: {e}", exc_info=True)
+
+    return {
+        'on_connecting':     on_connecting,
+        'on_connected':      on_connected,
+        'on_connected_token': on_connected_token,
+        'on_disconnected':   on_disconnected,
+        'on_error':          on_error,
+    }

core/utils/encrypted_proxy/singbox.py

@@ -2,21 +2,34 @@ import json
 import os
 import subprocess
 import time
+from abc import ABC, abstractmethod
 from pathlib import Path
+from typing import Optional
+import threading
 
 from core.Constants import Constants
 
+import logging
+logger = logging.getLogger(__name__)
+
 
 class SingboxRunner:
 
-    _WRAPPER = Constants.SINGBOX_WRAPPER
-    _PID_FILE = Path(Constants.SINGBOX_PID_FILE)
-    _LOG_FILE = Path(Constants.SINGBOX_LOG_FILE)
+    _WRAPPER    = Constants.SINGBOX_WRAPPER
+    _PID_FILE   = Path(Constants.SINGBOX_PID_FILE)
+    _LOG_FILE   = Path(Constants.SINGBOX_LOG_FILE)
     _CONFIG_DIR = Path(Constants.SINGBOX_CONFIG_DIR)
 
-    _START_TIMEOUT_S = 15.0
-    _START_POLL_S = 0.5
-    _START_INITIAL_S = 2.0
+    _START_TIMEOUT_S  = 15.0
+    _START_POLL_S     = 0.5
+    _START_INITIAL_S  = 2.0
+
+    _ORPHAN_GRACEFUL_TIMEOUT_S      = 4.0
+    _ORPHAN_FINAL_VERIFY_TIMEOUT_S  = 2.0
+    _ORPHAN_POLL_S                  = 0.3
+
+    def __init__(self):
+        self._wrapper_process: Optional[subprocess.Popen] = None
 
     def start(self, config_path: Path) -> bool:
         self.stop()
@@ -24,13 +37,14 @@ class SingboxRunner:
         self._ensure_log_file()
 
         try:
-            subprocess.Popen(
+            self._wrapper_process = subprocess.Popen(
                 ['sudo', self._WRAPPER, str(config_path), 'start'],
                 stdout=subprocess.DEVNULL,
                 stderr=subprocess.DEVNULL,
                 stdin=subprocess.DEVNULL,
                 env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
             )
+            logger.info(f"Wrapper spawned (PID {self._wrapper_process.pid})")
         except FileNotFoundError:
             raise RuntimeError(
                 f'Wrapper not found: {self._WRAPPER}\n'
@@ -51,6 +65,15 @@ class SingboxRunner:
         )
 
     def stop(self) -> None:
+        if self._wrapper_process is not None:
+            try:
+                self._wrapper_process.terminate()
+                self._wrapper_process.wait(timeout=2)
+            except subprocess.TimeoutExpired:
+                self._wrapper_process.kill()
+            finally:
+                self._wrapper_process = None
+
         self._kill_orphans()
 
         try:
@@ -63,6 +86,7 @@ class SingboxRunner:
                 timeout=15,
             )
         except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
+            logger.warning("Graceful stop failed, attempting emergency kill")
             self._emergency_kill()
 
     def is_running(self) -> bool:
@@ -100,7 +124,7 @@ class SingboxRunner:
                 env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
             )
             return result.stdout.strip()
-        except Exception:
+        except (subprocess.TimeoutExpired, FileNotFoundError, OSError):
             try:
                 lines = self._LOG_FILE.read_text(errors='replace').splitlines()
                 return '\n'.join(lines[-80:])
@@ -121,6 +145,88 @@ class SingboxRunner:
             tmp.unlink(missing_ok=True)
             raise
 
+    # ── Kill helpers ──────────────────────────────────────────────────────────
+
+    def _kill_orphans(self) -> None:
+        if not self._has_running_processes():
+            return
+        logger.debug("Found running sing-box processes, attempting graceful stop")
+        self._attempt_graceful_stop()
+        if self._wait_for_process_exit(self._ORPHAN_GRACEFUL_TIMEOUT_S):
+            logger.info("sing-box stopped gracefully")
+            return
+        logger.warning("Graceful stop timeout, force killing remaining sing-box processes")
+        self._force_kill_all_processes()
+        if self._wait_for_process_exit(self._ORPHAN_FINAL_VERIFY_TIMEOUT_S):
+            logger.info("All sing-box processes terminated")
+            return
+        logger.error("Failed to terminate all sing-box processes after force kill")
+
+    def _has_running_processes(self) -> bool:
+        try:
+            result = subprocess.run(['pgrep', '-x', 'sing-box'], capture_output=True, text=True)
+            return result.returncode == 0
+        except OSError:
+            return False
+
+    def _attempt_graceful_stop(self) -> None:
+        try:
+            subprocess.run(
+                ['sudo', self._WRAPPER, '', 'stop'],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                stdin=subprocess.DEVNULL,
+                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
+                timeout=15,
+            )
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+
+    def _wait_for_process_exit(self, timeout_s: float) -> bool:
+        deadline = time.monotonic() + timeout_s
+        while time.monotonic() < deadline:
+            if not self._has_running_processes():
+                return True
+            time.sleep(self._ORPHAN_POLL_S)
+        return False
+
+    def _force_kill_all_processes(self) -> None:
+        try:
+            result = subprocess.run(['pgrep', '-x', 'sing-box'], capture_output=True, text=True)
+            if result.returncode != 0:
+                return
+            pids = [int(p) for p in result.stdout.strip().splitlines() if p.strip().isdigit()]
+            for pid in pids:
+                self._force_kill_process(pid)
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+
+    def _force_kill_process(self, pid: int) -> None:
+        try:
+            subprocess.run(
+                ['sudo', 'kill', '-9', str(pid)],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+                stdin=subprocess.DEVNULL,
+                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
+                timeout=5,
+            )
+            logger.debug(f"Force killed sing-box process {pid}")
+        except (OSError, subprocess.TimeoutExpired):
+            pass
+
+    def _emergency_kill(self) -> None:
+        if not self._PID_FILE.exists():
+            return
+        try:
+            pid = int(self._PID_FILE.read_text().strip())
+            if pid > 0:
+                self._force_kill_process(pid)
+        except (ValueError, OSError):
+            pass
+        finally:
+            self._PID_FILE.unlink(missing_ok=True)
+
     def _ensure_log_file(self) -> None:
         self._LOG_FILE.parent.mkdir(parents=True, exist_ok=True)
         if not self._LOG_FILE.exists():
@@ -136,123 +242,77 @@ class SingboxRunner:
                 f'  Allowed:   {self._CONFIG_DIR}'
             )
 
-    def _emergency_kill(self) -> None:
-        if not self._PID_FILE.exists():
-            return
 
-        try:
-            pid = int(self._PID_FILE.read_text().strip())
-            if pid > 0:
-                subprocess.run(
-                    ['sudo', 'kill', '-9', str(pid)],
-                    stdout=subprocess.DEVNULL,
-                    stderr=subprocess.DEVNULL,
-                    stdin=subprocess.DEVNULL,
-                    env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
-                    timeout=5,
-                )
-        except (ValueError, OSError, subprocess.TimeoutExpired):
-            pass
-        finally:
-            self._PID_FILE.unlink(missing_ok=True)
+# ── Abstract base monitor ─────────────────────────────────────────────────────
 
-    def _kill_orphans(self) -> None:
-        try:
-            result = subprocess.run(
-                ['pgrep', '-x', 'sing-box'],
-                capture_output=True,
-                text=True,
-            )
+CAUSE_DISCONNECT = 'disconnect'
+CAUSE_LEAK       = 'leak'
+CAUSE_DISPLACED  = 'displaced'
 
-            if result.returncode != 0:
-                return
 
-            pids = [
-                int(p)
-                for p in result.stdout.strip().splitlines()
-                if p.strip().isdigit()
-            ]
+class SingboxProcessMonitor(ABC):
+    """Abstract monitor for sing-box process health.
 
-            if not pids:
-                return
+    Clients inherit this and implement event callbacks based on their
+    threading model (threading.Event for CLI, Qt signals for GUI).
+    """
 
-            subprocess.run(
-                ['sudo', self._WRAPPER, '', 'stop'],
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-                stdin=subprocess.DEVNULL,
-                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
-                timeout=15,
-            )
+    POLL_INTERVAL = 5
 
-            deadline = time.monotonic() + 4.0
+    def __init__(self, runner: SingboxRunner):
+        self._runner  = runner
+        self._lock    = threading.Lock()
+        self._running = False
+        self._stop_event = threading.Event()
+        self._last_poll_ts = None
 
-            while time.monotonic() < deadline:
-                check = subprocess.run(
-                    ['pgrep', '-x', 'sing-box'],
-                    capture_output=True,
-                )
+    def stop(self):
+        with self._lock:
+            self._running = False
+        self._stop_event.set()
 
-                if check.returncode != 0:
-                    return
+    @abstractmethod
+    def on_process_alive(self) -> None:
+        pass
 
-                time.sleep(0.3)
+    @abstractmethod
+    def on_process_dead(self) -> None:
+        pass
 
-            for pid in pids:
-                try:
-                    subprocess.run(
-                        ['sudo', 'kill', '-9', str(pid)],
-                        stdout=subprocess.DEVNULL,
-                        stderr=subprocess.DEVNULL,
-                        stdin=subprocess.DEVNULL,
-                        env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
-                        timeout=5,
-                    )
-                except (OSError, subprocess.TimeoutExpired):
-                    pass
+    @abstractmethod
+    def on_monitor_stopped(self, cause: str) -> None:
+        pass
 
-        except (OSError, ValueError, subprocess.TimeoutExpired):
-            pass
-        try:
-            result = subprocess.run(
-                ['pgrep', '-x', 'sing-box'],
-                capture_output=True,
-                text=True,
-            )
+    def _monitor_loop(self) -> None:
+        while True:
+            with self._lock:
+                if not self._running:
+                    break
 
-            if result.returncode != 0:
-                return
+            interrupted = self._stop_event.wait(timeout=self.POLL_INTERVAL)
+            if interrupted:
+                break
 
-            pids = [
-                int(p)
-                for p in result.stdout.strip().splitlines()
-                if p.strip().isdigit()
-            ]
+            alive = self._runner.is_running()
+            if alive:
+                self.on_process_alive()
+            else:
+                self.on_process_dead()
+                break
 
-            if not pids:
-                return
+    def run_blocking(self) -> None:
+        with self._lock:
+            self._running = True
 
-            subprocess.run(
-                ['sudo', self._WRAPPER, '', 'stop'],
-                stdout=subprocess.DEVNULL,
-                stderr=subprocess.DEVNULL,
-                stdin=subprocess.DEVNULL,
-                env={**os.environ, 'SUDO_ASKPASS': '/bin/false'},
-                timeout=15,
-            )
+        self._stop_event.clear()
+        t = threading.Thread(target=self._monitor_loop, daemon=False)
+        t.start()
 
-            deadline = time.monotonic() + 4.0
+        self._stop_event.wait()
 
-            while time.monotonic() < deadline:
-                check = subprocess.run(
-                    ['pgrep', '-x', 'sing-box'],
-                    capture_output=True,
-                )
+        with self._lock:
+            self._running = False
 
-                if check.returncode != 0:
-                    return
-
-                time.sleep(0.3)
-
-        except (OSError, ValueError, subprocess.TimeoutExpired):
-            pass
+        t.join(timeout=5)
+        self._runner.stop()
+        self.on_monitor_stopped("user_requested")