From bf22f764d06ce7b6bd5f40d9c644a5a4c7f1e06a Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Wed, 29 Jul 2026 18:58:59 -0400 Subject: [PATCH] Singbox takedown modules, with a graceful shutdown, then kill, then searching for processes with the phrase. And doing tunnel checks inbetween --- core/controllers/SystemStateController.py | 7 +- core/models/Result.py | 1 + core/models/system/SystemState.py | 2 + .../encrypted_proxy/process_closure_tools.py | 205 ++++++++++++++++++ .../systemwide/encrypted_proxy/singbox.py | 7 + .../encrypted_proxy/singbox_runner.py | 29 ++- .../general_tools/interface_tools.py | 31 +++ 7 files changed, 277 insertions(+), 5 deletions(-) create mode 100644 core/services/networking/systemwide/encrypted_proxy/process_closure_tools.py create mode 100644 core/services/networking/systemwide/general_tools/interface_tools.py diff --git a/core/controllers/SystemStateController.py b/core/controllers/SystemStateController.py index de25700..9da0c19 100644 --- a/core/controllers/SystemStateController.py +++ b/core/controllers/SystemStateController.py @@ -12,8 +12,11 @@ class SystemStateController: return SystemState.exists() @staticmethod - def create(profile_id: int, firewalled: bool, dns_set: bool) -> SystemState: - current_state = SystemState(profile_id, firewalled, dns_set) + def create(profile_id: int, firewalled: bool, dns_set: bool, process_id: int = None) -> SystemState: + if process_id: + current_state = SystemState(profile_id, firewalled, dns_set, process_id) + else: + current_state = SystemState(profile_id, firewalled, dns_set) current_state.save() return current_state diff --git a/core/models/Result.py b/core/models/Result.py index 1ea265f..5c3ca13 100644 --- a/core/models/Result.py +++ b/core/models/Result.py @@ -11,6 +11,7 @@ class ResultError(Enum): INVALID_INPUT = "invalid_input" MISSING_FILE = "missing_file" MISSING_DEPENDENCY = "missing_dependency" + MISSING_DATA = "missing_data" CONNECTION = "connection" DATABASE = "database" PERMISSION = "permission" diff --git a/core/models/system/SystemState.py b/core/models/system/SystemState.py index 5467699..07b49fb 100644 --- a/core/models/system/SystemState.py +++ b/core/models/system/SystemState.py @@ -6,6 +6,7 @@ from typing import Self import json import os import pathlib +from typing import Optional @dataclass_json @dataclass @@ -13,6 +14,7 @@ class SystemState: profile_id: int firewalled: bool dns_set: bool + process_id: Optional[int] = None def save(self: Self): diff --git a/core/services/networking/systemwide/encrypted_proxy/process_closure_tools.py b/core/services/networking/systemwide/encrypted_proxy/process_closure_tools.py new file mode 100644 index 0000000..a194c86 --- /dev/null +++ b/core/services/networking/systemwide/encrypted_proxy/process_closure_tools.py @@ -0,0 +1,205 @@ +from core.services.networking.systemwide.general_tools.interface_tools import check_interface_exists, check_interface_up +from core.errors.exceptions import SudoScript +from core.models.system.SystemState import SystemState +from core.models.Result import Result, ResultError +from core.errors.logger import logger +from core.services.networking.systemwide.encrypted_proxy import singbox +from core.utils.basic_operations import process_tools +import subprocess +from typing import Callable, cast +import time + +KILL_WAIT_TIME = 1.3 +APP_NAME = 'sing-box' + + +def is_tunnel_active(interface: str) -> bool: + """Returns True if active, False if inactive. Raises on check failure.""" + result = check_interface_exists(interface) + if result.valid: + return True + elif result.error_type == ResultError.INTERFACE: + return False + else: + raise RuntimeError(f"Could not verify tunnel status: {result.message}") + +def _try_with_permission_fallback(operation: Callable, interface: str) -> Result: + """Execute operation; if denied, check if tunnel is still active.""" + try: + return operation() + except SudoScript as e: + logger.error("The Sudo Scripts giving power to kill this are being denied permission, before we flag this, let's see if the proxy is active, which does NOT need permission to check,") + existance = check_interface_exists(interface) + if not existance.valid and existance.error_type == ResultError.INTERFACE: + not_existing = "Permission denied, but the proxy interface is down, so this is acceptable" + logger.info(not_existing) + return Result(valid=True, message=not_existing) + error_msg = "Critical Issue, the tunnel is up, but we lack permission to take it down." + logger.error(error_msg) + return Result(valid=False, error_type=ResultError.PERMISSION, message=error_msg) + + +def _escalate_kill(pid: int) -> Result: + """Graceful → wait → force kill escalation.""" + if not process_tools.is_running(pid): + return Result(valid=True, message=f"PID {pid} already down") + + graceful = singbox.turn_off(pid) + time.sleep(KILL_WAIT_TIME) + + if not process_tools.is_running(pid): + return Result(valid=True, message=f"PID {pid} gracefully closed") + + force = singbox.force_kill(pid) + return Result(valid=force.valid, message=f"Force kill {'succeeded' if force.valid else 'failed'}") + + +def _get_process_id(current_state: SystemState) -> Result: + process_id = current_state.process_id + + if not process_id: + return Result(valid=False, error_type=ResultError.MISSING_DATA, message="Missing the critical process id.") + + try: + process_id_as_int = int(process_id) + return Result(valid=True, data=process_id_as_int) + except ValueError as e: + return Result(valid=False, error_type=ResultError.INVALID_INPUT, data=e, message="The process id is not a valid integer.") + + +# Function is private because it leverages singbox only functionality. +def _shut_down_by_known_process_id(process_id: int) -> Result: + function_name = "_shut_down_by_known_process_id" + + active = process_tools.is_running(process_id) + + if not active: + logger.error(f"[{function_name}] The process ID we have for Singbox is not actually active still.") + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"The process ID {process_id} Already was down.") + + graceful_close = singbox.turn_off(process_id) + if not graceful_close.valid: + logger.error(f"[{function_name}] Could not gracefully close it. So we'll {KILL_WAIT_TIME} seconds and kill it.") + + still_active = process_tools.is_running(process_id) + + if not still_active: + return Result(valid=True, message=f"The process ID {process_id} gracefully closed.") + + time.sleep(KILL_WAIT_TIME) + force_kill = singbox.force_kill(process_id) + + if force_kill.valid: + return Result(valid=True, message=f"The process ID {process_id} was forcefully closed.") + else: + return Result(valid=False, message=f"Even a force kill couldn't stop the process ID {process_id}.") + + + +def get_any_pid_with_the_phrase(which_application: str) -> Result: + result = subprocess.run(['pgrep', '-a', which_application], capture_output=True, text=True) + if result.returncode == 0: + lines = result.stdout.strip().split('\n') + pid_data = [{'pid': int(line.split()[0]), 'command': line} for line in lines if line] + if pid_data: + return Result(valid=True, data=pid_data) # List of dicts with pid and full command + else: + return Result(valid=False, error_type=ResultError.MISSING_DATA, message="No processes found") + else: + return Result(valid=False, error_type=ResultError.NOT_SUPPORTED, message="pgrep failed") + + +# Function is public because it works for any app +def get_pid_by_exact_match(which_application: str) -> Result: + result = subprocess.run(['pgrep', '-x', which_application], capture_output=True, text=True) + if result.returncode == 0: + try: + process_id = int(result.stdout.strip()) # Strip newline, convert to int + return Result(valid=True, data=process_id) + except ValueError: + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="Could not parse PID") + else: + return Result(valid=False, error_type=ResultError.NOT_SUPPORTED, message="pgrep failed") + + +def _hunt_and_kill(interface_name: str) -> Result: + """ + Purpose: + Kill all pids with the phrase + + Flow: + Gets a list of all pids with the phrase in it, + Goes through each and tries to kill it with escalating force. + Finally evaluates for each pid if it killed the tunnel. + """ + function_name = "_hunt_and_kill" + + id_results = get_any_pid_with_the_phrase(APP_NAME) + if id_results.valid: + pid_data_list = id_results.data + else: + return id_results + + for idx, pid_info in enumerate(pid_data_list): + pid = pid_info['pid'] + command = pid_info['command'] + + logger.info(f"[{function_name}] Attempting to kill PID {pid}: {command}") + graceful_or_force = _escalate_kill(pid) + logger.info(f"Escalation of killing of pid {pid} resulted in {graceful_or_force.valid} and {graceful_or_force.message}") + if not graceful_or_force.valid: + logger.info("Skipping to next pid..") + continue + + if not process_tools.is_running(pid): + logger.info(f"[{function_name}] PID {pid} force killed successfully") + + # EVALUATION: Did killing this PID stop the app? is the interface still running? + existance = check_interface_exists(interface=interface_name) + if not existance.valid: + logger.info(f"[{function_name}] App/Tunnel is fully stopped after killing PID {pid}") + return Result(valid=True, message=f"App stopped after killing PID {pid}") + else: + logger.warning(f"[{function_name}] App still running, continuing to next PID") + continue + + else: + logger.error(f"[{function_name}] Could not kill PID {pid}, moving to next") + continue + + return Result(valid=False, error_type=ResultError.UNKNOWN, message="Could not kill any PIDs") + + +def _try_shutdown_by_exact_match() -> Result: + exact_match = get_pid_by_exact_match(APP_NAME) + if exact_match.valid: + logger.info(f"Attempting exact match PID {exact_match.data}") + return _shut_down_by_known_process_id(cast(int, exact_match.data)) + else: + return exact_match + + +def _try_known_pid_from_state(current_state: SystemState) -> Result: + pid_query = _get_process_id(current_state) + if not pid_query.valid: + return pid_query + return _shut_down_by_known_process_id(cast(int, pid_query.data)) + + +def orchestrate_closing(current_state: SystemState, interface_name: str) -> Result: + strategies = [ + lambda: _try_known_pid_from_state(current_state), + lambda: _try_shutdown_by_exact_match(), + lambda: _hunt_and_kill(interface_name), + ] + quantity_of_strategies = len(strategies) + + for strategy in strategies: + result = _try_with_permission_fallback(strategy, interface_name) + if result.valid: + return result + if not is_tunnel_active(interface_name): + return Result(valid=True, message="Tunnel inactive despite strategy failure") + + return Result(valid=False, message=f"All {quantity_of_strategies} strategies tried, we simply can not bring down the proxy, which is still up.") + diff --git a/core/services/networking/systemwide/encrypted_proxy/singbox.py b/core/services/networking/systemwide/encrypted_proxy/singbox.py index 88af9e9..0d88f7b 100644 --- a/core/services/networking/systemwide/encrypted_proxy/singbox.py +++ b/core/services/networking/systemwide/encrypted_proxy/singbox.py @@ -32,3 +32,10 @@ def hello_world_test(name: str) -> str: command = ['bash', '-c', f'. {SINGBOX_WRAPPER} && hello_world_test "{name}"'] human_readable_goal = "Hello world test" return run_generic_command(command, human_readable_goal, timeout=5) + + +def get_pid_by_exact_match(name: str) -> str: + command = ['pgrep', '-x', 'sing-box'] + human_readable_goal = "Get the exact match of the PID" + return run_generic_command(command, human_readable_goal, timeout=5) + diff --git a/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py b/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py index 6a3cbca..6f11975 100644 --- a/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py +++ b/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py @@ -1,4 +1,5 @@ from core.services.networking.systemwide.encrypted_proxy import singbox +from core.services.networking.systemwide.encrypted_proxy.process_closure_tools import orchestrate_closing from core.utils.basic_operations import process_tools from core.models.Result import Result, ResultError from core.errors.logger import logger @@ -11,6 +12,7 @@ from core.services.networking.systemwide.general_connection_tools.general_firewa from core.errors.exceptions import FirewallError, DNSError # generic +import subprocess import time QUANTITY_OF_ATTEMPTS = 2 @@ -53,6 +55,26 @@ def _attempt_start_with_retry(config_path: str, quantity_of_attempts: int = 2) - logger.error(f"[SINGBOX] Attempt {current_attempt} for Singbox Failed. Because: {activation_result.message}. Trying again..") +def end_singbox( + connection_observer: Optional[ConnectionObserver] = None +) -> Result: + + function_name = "END_SINGBOX" + + current_state = SystemState.get() + if not current_state: + return Result(valid=False, error_type=ResultError.MISSING_FILE, message="Already disabled or missing State JSON.") + + dead_singbox = orchestrate_closing( + current_state=current_state, + interface_name=SINGBOX_TUN_IF + ) + + if dead_singbox.valid: + logger.info("Successfully took down Singbox tunnel") + SystemState.dissolve() + + def start_singbox( profile_id: int, @@ -104,9 +126,10 @@ def start_singbox( logger.info(f"Setting State JSON with INTENDED firewall & Dns settings") current_state = SystemStateController.create( - profile_id=profile.id, + profile_id=profile_id, firewalled=True, # intended setting, not result yet - dns_set=True # intended setting, not result yet + dns_set=True, # intended setting, not result yet + process_id=process_id ) # ============= FIREWALL ============= @@ -132,5 +155,5 @@ def start_singbox( # raises error if not okay. # ============= CONCLUSION ============= - return Result(valid=True, data=current_state) + return Result(valid=True, data=process_id) diff --git a/core/services/networking/systemwide/general_tools/interface_tools.py b/core/services/networking/systemwide/general_tools/interface_tools.py new file mode 100644 index 0000000..b4528e1 --- /dev/null +++ b/core/services/networking/systemwide/general_tools/interface_tools.py @@ -0,0 +1,31 @@ +from core.models.Result import Result, ResultError + +import subprocess +import re + +def check_interface_exists(interface: str) -> Result: + result = subprocess.run(['ip', 'link', 'show', interface], capture_output=True) + if result.returncode != 0: + return Result(valid=False, error_type=ResultError.NOT_SUPPORTED, message=f"We could not run the command to even check the interface. {result.stdout}") + elif 'does not exist' in result.stderr.decode(): + return Result(valid=False, error_type=ResultError.INTERFACE, message="Interface does not exist.") + else: + # Check if interface name exists in the expected format + # Pattern: "digits: interface_name: " + output = result.stdout.decode() + pattern = rf'^\d+:\s+{re.escape(interface)}:\s+<[^>]+>' + if re.search(pattern, output, re.MULTILINE): + return Result(valid=True, data=output) + else: + return Result(valid=False, error_type=ResultError.INTERFACE, message="Interface format not recognized.") + + + +def check_interface_up(output: str) -> Result: + """ + Check if interface has LOWER_UP flag (meaning it's operationally up) + """ + if re.search(r'LOWER_UP', output): + return Result(valid=True) + else: + return Result(valid=False, error_type=ResultError.INTERFACE, message="Interface is not up. LOWER_UP flag not found.")