diff --git a/core/models/OperatorProxySession.py b/core/models/OperatorProxySession.py index 065d18d..da7e628 100644 --- a/core/models/OperatorProxySession.py +++ b/core/models/OperatorProxySession.py @@ -17,4 +17,6 @@ class OperatorProxySession: operator_domain: Optional[str] = None 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 \ No newline at end of file + 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 \ No newline at end of file diff --git a/core/models/system/SystemProfile.py b/core/models/system/SystemProfile.py index db86ea3..0a1de60 100644 --- a/core/models/system/SystemProfile.py +++ b/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' diff --git a/core/services/WebServiceApiService.py b/core/services/WebServiceApiService.py index 70a0b2d..484f8e3 100644 --- a/core/services/WebServiceApiService.py +++ b/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 diff --git a/core/services/encrypted_proxy/observer_helpers.py b/core/services/encrypted_proxy/observer_helpers.py new file mode 100644 index 0000000..9b4d458 --- /dev/null +++ b/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, + } \ No newline at end of file diff --git a/core/utils/encrypted_proxy/singbox.py b/core/utils/encrypted_proxy/singbox.py index 3589ade..35549b6 100644 --- a/core/utils/encrypted_proxy/singbox.py +++ b/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 \ No newline at end of file + t.join(timeout=5) + self._runner.stop() + self.on_monitor_stopped("user_requested") \ No newline at end of file