IP address checking for singbox enforced in singbox runner

core/models/Result.py

@@ -19,6 +19,7 @@ class ResultError(Enum):
     SUBSCRIPTION = "subscription"
     PROCESS_GOT_KILLED = "process_got_killed"
     PROCESS_WONT_START = "process_wont_start"
+    PROCESS_MISMATCH = "process_mismatch"
     NMCLI = "nmcli_issues"
     FIREWALL = "firewall"
     CLIENT_DNS = "client_dns"
@@ -26,6 +27,7 @@ class ResultError(Enum):
     INTERFACE = "interface"
     TIMEOUT = "timeout"
     INVALID_API_REPLY = "invalid_api_reply"
+    LEAK_ISSUE = "leak_issue"
     UNKNOWN = "unknown"
 
 @dataclass

core/services/networking/general_connection_tools/ip.py

@@ -0,0 +1,96 @@
+from core.services.networking.httpx.classify_response import classify_response
+
+# models
+from core.models.Result import Result, ResultError
+from core.services.networking.api_requests.ApiResponseModel import ApiResponse, ErrorType
+
+# errors/loggers/observers
+from core.Constants import Constants
+from core.errors.logger import logger
+from essentials.observers.ConnectionObserver import ConnectionObserver
+
+# generic
+import time
+import random
+import httpx
+from typing import Optional
+
+API_ENDPOINT_FORMATS = {
+    "https://api.ipify.org?format=json": "ip",
+    "https://api.myip.com": "ip",
+    "https://wtfismyip.com/json": "YourFuckingIPAddress",
+    "https://httpbin.org/ip": "origin",
+    "https://api.myip.com": "ip",
+    "https://ipinfo.io/json": "ip"
+}
+
+MAX_ATTEMPTS = 3
+
+def test(connection_observer: Optional[ConnectionObserver]) -> Result:
+    """
+    Cycles through 3 possible endpoints,
+    to attempt to get an IP address for the systemwide connection.
+    """
+    attempt = 0
+    endpoint_index = 0
+    endpoint_list = list(API_ENDPOINT_FORMATS.keys())
+
+    _ip_test_client = httpx.Client(http2=True, timeout=6)
+
+    while attempt < MAX_ATTEMPTS:
+        endpoint_url = endpoint_list[attempt]
+        attempt += 1
+        logger.info(f"Attempt {attempt}: Trying {endpoint_url}")
+        if connection_observer:
+            connection_observer.notify("connecting", f"Testing IP. Attempt {attempt}")
+        try:
+            raw_response = _ip_test_client.get(endpoint_url)
+            
+            filtered_response = classify_response(raw_response)
+
+            if filtered_response.valid:
+                ip_address = extract_ip_from_response(
+                    response=filtered_response.data,
+                    api_url=endpoint_url
+                )
+                if ip_address is not None:
+                    return Result(valid=True, data=ip_address)
+                else:
+                    logger.error(f"Skipping {endpoint_url}")
+
+        except Exception as e:
+            logger.error(f"We are NOT error handling this.  Skipping {endpoint_url}. Invalid API reply or failed connection {str(e)}")
+
+    logger.error(f"We went through {MAX_ATTEMPTS} attempts and could not get a reply.")
+    return Result(valid=False, error_type=ResultError.CONNECTION, message="Could not connect.")
+
+
+def extract_ip_from_response(response: dict, api_url: str) -> str | bool:
+    """
+    Purpose:
+        Extract the IP address from the API's response,
+        based on which keys that API uses.
+
+    Rank:
+        Helper
+
+    On Success:
+        Returns String of IP address
+
+    On Failure,
+        Returns None
+    """
+
+    if api_url not in API_ENDPOINT_FORMATS:
+        logger.critical(f"Systemic Bug! Our local data in extract_ip_from_response doesn't match what we just read for the URL {api_url}.")
+        return None
+
+    desired_key = API_ENDPOINT_FORMATS[api_url]
+
+    if desired_key not in response:
+        logger.error(f"Our local data does NOT match the new format for {api_url}.")
+        return None
+
+    final_value = response[desired_key]
+    return final_value
+

core/services/networking/systemwide/encrypted_proxy/close_singbox.py

@@ -8,7 +8,7 @@ from core.services.networking.systemwide.encrypted_proxy import singbox
 from core.utils.basic_operations import process_tools
 from core.utils.run_commands import run_generic_command
 
-from typing import Callable, cast
+from typing import Callable, cast, Optional
 import time
 
 KILL_WAIT_TIME = 1.3
@@ -45,7 +45,7 @@ def _escalate_kill(pid: int) -> Result:
     return Result(valid=force.valid, message=f"Force kill {'succeeded' if force.valid else 'failed'}")
 
 
-def get_process_id_from_state(current_state: SystemState) -> Result:
+def get_process_id_from_state(current_state: Optional[SystemState]) -> Result:
     if current_state is None:
         return Result(valid=False, error_type=ResultError.MISSING_DATA, message="Missing the State itself.")
 
@@ -143,7 +143,7 @@ def _try_shutdown_by_exact_match() -> Result:
         return exact_match
 
 
-def _try_known_pid_from_state(current_state: SystemState) -> Result:
+def _try_known_pid_from_state(current_state: Optional[SystemState]) -> Result:
     pid_query = get_process_id_from_state(current_state)
     if not pid_query.valid:
         return pid_query

core/services/networking/systemwide/encrypted_proxy/singbox_runner.py

@@ -4,6 +4,7 @@ from core.services.networking.systemwide.general_tools import interface_tools
 from core.services.networking.systemwide import killswitch
 from core.services.networking.systemwide.wireguard.wg_firewall_dns import revert_dns
 from core.utils.basic_operations import pid_tools
+from core.services.networking.general_connection_tools import ip
 
 from core.models.Result import Result, ResultError
 from core.errors.logger import logger
@@ -48,6 +49,12 @@ def set_dns_for_singbox(current_state: SystemState):
 
 
 def launch_singbox_binary(profile_id: int) -> Result:
+    """
+    1) Launch the singbox binary,
+    2) Test if the process id is working
+    3) If not, check if the app is running by name. (on any pid)
+    """
+
     activation_result = singbox.start(profile_id)
     if not activation_result.valid:
         return activation_result
@@ -57,12 +64,19 @@ def launch_singbox_binary(profile_id: int) -> Result:
     logger.info(f"Waiting 2 seconds to see if the process id {process_id} is still alive..")
     time.sleep(2)
 
-    # Evaluate if running.
+    # Evaluate if running by that exact pid:
     active = pid_tools.is_running(pid=process_id, process_name="sing-box")
     logger.info(f"Process {process_id} is {active}")
 
     if active:
         return Result(valid=True, data=process_id)
+    else:
+        logger.error(f"Initial pid test showed it is NOT running for pid {process_id}.  Now we're doing a more exhaustive check by app name that might be a different ID")
+        double_check = pid_tools.get_pid_by_app_name(exact_app_name="sing-box")
+
+        if double_check.valid:
+            logger.info(f"We had the wrong pid!! We had {process_id}, when it's really {double_check.data}")
+            return Result(valid=True, error_type=ResultError.PROCESS_MISMATCH, data=double_check.data)
         else:
             error_msg = f"While Singbox might have literally allowed the binary to begin, it's killing the process on id {process_id}"
             logger.error(f"[{function_name}] {error_msg}")
@@ -214,7 +228,7 @@ def start_singbox(
         return Result(valid=False, error_type=ResultError.INTERFACE, data=process_id)
 
     # ============= SETUP STATE =============
-    # Even if firewall is off, we want to save the fact we turned Singbox on, before we raise errors.
+    # Even if the connection is dead, firewall is off, we want to save the fact we turned Singbox on, before we raise errors.
 
     logger.info("Setting State JSON with INTENDED firewall & Dns settings")
     current_state = SystemStateController.create(
@@ -224,6 +238,23 @@ def start_singbox(
         process_id=process_id
     )
 
+    # ============= CONFIRM IP ADDRESS =============
+    ip_result = ip.test(connection_observer)
+    if not ip_result.valid:
+        logger.error(f"IP address check is unable to connect.  Killing Singbox pid {process_id}")
+        closed = orchestrate_closing(current_state=current_state, interface=Constants.SINGBOX_TUN_IF)
+        raise ConnectionError("Could not connect with IP adddress check.")
+        # return Result(valid=False, error_type=ResultError.CONNECTION, message="Could not connect with IP adddress check.")
+
+    observed_ip_address = ip_result.data
+    if observed_ip_address == server_ip:
+        logger.info("Observed IP matches the intended server IP.  Proceeding..")
+    else:
+        error_msg = f"Critical Leak! IP address check of {observed_ip_address} does NOT match the intended {server_ip}."
+        logger.error(f"{error_msg} Killing Singbox pid {process_id}")
+        closed = orchestrate_closing(current_state=current_state, interface=Constants.SINGBOX_TUN_IF)
+        return Result(valid=False, error_type=ResultError.LEAK_ISSUE, message=error_msg)
+
     # ============= FIREWALL =============
     logger.info(f"[{function_name}] Attempting to enable the Firewall for {Constants.SINGBOX_TUN_IF} and {Constants.SINGBOX_INTERNAL_SUBNET}...")
     firewall_result = enable_firewall_w_retry(  # this function is "generic" as it's protocol neutral