From 1a60511492dbc8221988437b970f495b6caed1a2 Mon Sep 17 00:00:00 2001 From: zenaku Date: Wed, 8 Jul 2026 11:14:41 -0500 Subject: [PATCH] refactor: align cli with core architecture --- .gitignore | 3 +- .../encrypted_proxy/observer_helpers.py | 25 ++- .../encrypted_proxy/killswitch_monitor.py | 143 ++++++++++++++++++ core/utils/encrypted_proxy/singbox.py | 13 ++ 4 files changed, 176 insertions(+), 8 deletions(-) create mode 100644 core/utils/encrypted_proxy/killswitch_monitor.py diff --git a/.gitignore b/.gitignore index cecfc1b..bf7465e 100644 --- a/.gitignore +++ b/.gitignore @@ -4,4 +4,5 @@ prototype_client.py .venv __pycache__ dist -.env \ No newline at end of file +.env +docs.md \ No newline at end of file diff --git a/core/services/encrypted_proxy/observer_helpers.py b/core/services/encrypted_proxy/observer_helpers.py index 9b4d458..228f732 100644 --- a/core/services/encrypted_proxy/observer_helpers.py +++ b/core/services/encrypted_proxy/observer_helpers.py @@ -81,6 +81,7 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: """Factory: returns event handler closures bound to state.""" def _update_connection_data(event) -> ConnectionStateData: + """Extract connection fields from event subject into a ConnectionStateData.""" d = event.subject or {} return ConnectionStateData( tunnel_interface=d.get('tunnel_if', "?"), @@ -91,6 +92,7 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: ) def _update_ui_labels(conn: ConnectionStateData) -> None: + """Refresh all CLI status labels from current connection data.""" from cli.ui import ( label_connected, label_killswitch, label_dns, label_ipv6, label_ipv4 ) @@ -101,7 +103,13 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: if conn.server_ip: label_ipv4(server_ip=conn.server_ip, reachable=True) + def _update_monitor_port(event_data: Dict) -> None: + """Update monitor with new SOCKS5 port.""" + if state['monitor'] is not None: + state['monitor'].set_connection_data(socks5_port=event_data.get('socks5_port')) + def on_connecting(event) -> None: + """Handle connecting event: show attempt spinner.""" try: cleanup_timer(state) if (state['timer_state'] and @@ -122,6 +130,7 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: logger.error(f"Error in on_connecting: {e}", exc_info=True) def on_connected(event) -> None: + """Handle connected event: update labels, monitor port, and start session timer.""" try: cleanup_spinner(state) cleanup_reconnecting_spinner(state) @@ -132,9 +141,8 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: 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')) + event_data = event.subject or {} + _update_monitor_port(event_data) if state['timer_state'] is not None: from cli.ui import session_timer @@ -146,6 +154,7 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: logger.error(f"Error in on_connected: {e}", exc_info=True) def on_connected_token(event) -> None: + """Handle session token event: pass token to monitor.""" try: if state['monitor'] is not None: token = (event.subject or {}).get('session_token') @@ -156,6 +165,7 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: logger.error(f"Error in on_connected_token: {e}", exc_info=True) def on_disconnected(event) -> None: + """Handle disconnected event: cleanup UI and show disconnect label.""" try: cleanup_all(state) from cli.ui import label_disconnect @@ -165,6 +175,7 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: logger.error(f"Error in on_disconnected: {e}", exc_info=True) def on_error(event) -> None: + """Handle error event: cleanup UI and show error label.""" try: cleanup_all(state) from cli.ui import label_error @@ -177,9 +188,9 @@ def create_event_handlers(state: Dict[str, Any], drop_state: Dict, tunnel_state: logger.error(f"Error in on_error: {e}", exc_info=True) return { - 'on_connecting': on_connecting, - 'on_connected': on_connected, + 'on_connecting': on_connecting, + 'on_connected': on_connected, 'on_connected_token': on_connected_token, - 'on_disconnected': on_disconnected, - 'on_error': on_error, + 'on_disconnected': on_disconnected, + 'on_error': on_error, } \ No newline at end of file diff --git a/core/utils/encrypted_proxy/killswitch_monitor.py b/core/utils/encrypted_proxy/killswitch_monitor.py new file mode 100644 index 0000000..2d71c70 --- /dev/null +++ b/core/utils/encrypted_proxy/killswitch_monitor.py @@ -0,0 +1,143 @@ +import subprocess +import threading +import time +from core.utils.encrypted_proxy import killswitch +from core.utils.encrypted_proxy.singbox import SingboxRunner, SingboxProcessMonitor +from core.utils.encrypted_proxy.singbox import CAUSE_DISCONNECT, CAUSE_LEAK, CAUSE_DISPLACED + +drop_state: dict = {"count": 0} +tunnel_state: dict = {"alive": True} + + +class KillswitchMonitor(SingboxProcessMonitor): + """CLI-specific monitor: adds killswitch + drops detection.""" + + LOG_PREFIX = "hydraveil-drop" + + def __init__(self): + super().__init__(SingboxRunner()) + self._on_disconnect = None + self._on_leak = None + self._on_displaced = None + self._socks5_port = None + self._session_token = None + self._cause_event = threading.Event() + self._cause: str | None = None + + # ── Setters ─────────────────────────────────────────────────────────────── + + def set_on_disconnect(self, fn): + self._on_disconnect = fn + + def set_on_leak(self, fn): + self._on_leak = fn + + def set_on_displaced(self, fn): + self._on_displaced = fn + + def set_connection_data(self, socks5_port: int): + self._socks5_port = socks5_port + + def set_session_token(self, token: str): + self._session_token = token + + # ── Abstract impl ───────────────────────────────────────────────────────── + + def on_process_alive(self) -> None: + tunnel_state["alive"] = True + + def on_process_dead(self) -> None: + tunnel_state["alive"] = False + + def on_monitor_stopped(self, cause: str) -> None: + if cause == CAUSE_DISCONNECT and self._on_disconnect: + self._on_disconnect() + elif cause == CAUSE_LEAK and self._on_leak: + self._on_leak() + elif cause == CAUSE_DISPLACED and self._on_displaced: + self._on_displaced() + + # ── Override monitor loop to add CLI-specific checks ────────────────────── + + def _monitor_loop(self) -> None: + while True: + with self._lock: + if not self._running: + break + + interrupted = self._stop_event.wait(timeout=self.POLL_INTERVAL) + if interrupted: + break + + # Killswitch check + if not killswitch.status(): + tunnel_state["alive"] = False + self._trigger(CAUSE_DISCONNECT) + return + + # Sing-box check + alive = self._runner.is_running() + tunnel_state["alive"] = alive + if not alive: + self._trigger(CAUSE_LEAK) + return + + # Session token check + if self._session_token is not None: + from core.models.system.SystemState import SystemState + current = SystemState.get() + if current is None or current.session_token != self._session_token: + tunnel_state["alive"] = False + self._trigger(CAUSE_DISPLACED) + return + + self._poll_drops() + + def _trigger(self, cause: str): + self._cause = cause + with self._lock: + self._running = False + self._cause_event.set() + self._stop_event.set() + + def _poll_drops(self) -> None: + since = self._last_poll_ts or "10 seconds ago" + try: + result = subprocess.run( + [ + "journalctl", "-k", "--no-pager", + f"--grep={self.LOG_PREFIX}", + "--since", since, + "-o", "short-iso", + ], + capture_output=True, + text=True, + timeout=5, + ) + new_drops = result.stdout.count(self.LOG_PREFIX) + if new_drops: + drop_state["count"] += new_drops + except (subprocess.TimeoutExpired, FileNotFoundError): + pass + finally: + self._last_poll_ts = time.strftime("%Y-%m-%dT%H:%M:%S") + + # ── Override run_blocking to handle cause_event ─────────────────────────── + + def run_blocking(self) -> None: + with self._lock: + self._running = True + + self._stop_event.clear() + self._cause_event.clear() + self._cause = None + drop_state["count"] = 0 + tunnel_state["alive"] = True + self._last_poll_ts = time.strftime("%Y-%m-%dT%H:%M:%S") + + t = threading.Thread(target=self._monitor_loop, daemon=True) + t.start() + self._cause_event.wait() + t.join(timeout=2) + + self.on_monitor_stopped(self._cause or "user_requested") \ No newline at end of file diff --git a/core/utils/encrypted_proxy/singbox.py b/core/utils/encrypted_proxy/singbox.py index 35549b6..60de1dd 100644 --- a/core/utils/encrypted_proxy/singbox.py +++ b/core/utils/encrypted_proxy/singbox.py @@ -148,18 +148,31 @@ class SingboxRunner: # ── Kill helpers ────────────────────────────────────────────────────────── def _kill_orphans(self) -> None: + """Kill any lingering sing-box processes using escalating force.""" + + # Phase 1: Check if anything is actually running if not self._has_running_processes(): return + logger.debug("Found running sing-box processes, attempting graceful stop") + + # Phase 2: Graceful stop attempt via wrapper self._attempt_graceful_stop() + + # Phase 3: Wait for graceful shutdown to complete if self._wait_for_process_exit(self._ORPHAN_GRACEFUL_TIMEOUT_S): logger.info("sing-box stopped gracefully") return + + # Phase 4: Graceful stop failed — force kill remaining processes logger.warning("Graceful stop timeout, force killing remaining sing-box processes") self._force_kill_all_processes() + + # Phase 5: Verify all processes are actually dead 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: