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

@@ -12,7 +12,10 @@ class SystemStateController:
         return SystemState.exists()
 
     @staticmethod
-    def create(profile_id: int, firewalled: bool, dns_set: bool) -> SystemState:
+    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

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"

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):
 

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.")
+

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)
+

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)
 

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: <flags>"
+        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.")