From 840c2ccb58191f8f21a86e07289d4a86bdcd67eb Mon Sep 17 00:00:00 2001 From: SimplifiedPrivacy Date: Mon, 19 Oct 2026 16:32:43 -0400 Subject: [PATCH] Improved State Management integrated with the UI upgrades, to evaluate failed zombie systemwide connections. And then prompt the user if they'd like to evaluate them for re-use, or close it. --- core/errors/translate_error.py | 20 ++- core/infrastructure/global_state.py | 4 + .../global_state_coordinator.py | 153 +++++++++++++++++- core/infrastructure/plumbing.py | 36 ++++- core/models/Result.py | 2 + core/models/system/SystemState.py | 13 ++ .../dns_tools/process_dns_result.py | 18 ++- .../general_tools/test_any_protocol.py | 64 ++++++++ .../systemwide/systemwide_wireguard.py | 2 +- core/utils/type_checker.py | 98 ++++++++--- 10 files changed, 376 insertions(+), 34 deletions(-) create mode 100644 core/services/networking/systemwide/general_tools/test_any_protocol.py diff --git a/core/errors/translate_error.py b/core/errors/translate_error.py index 4e5cdc5..7e2ae05 100644 --- a/core/errors/translate_error.py +++ b/core/errors/translate_error.py @@ -10,6 +10,7 @@ from core.utils.type_checker import type_checker from typing import Type, Optional +import traceback ERROR_TO_UI_MESSAGE = { ProfileActivationError: "The profile could not be enabled", @@ -18,7 +19,7 @@ ERROR_TO_UI_MESSAGE = { ProfileModificationError: "WireGuard configuration could not be attached.", } -@type_checker + def translate_error(error: Type[Exception], profile_id: int) -> Result: """Translate legacy core errors into actions and/or messages via Result Objects.""" logger.error(str(error)) @@ -64,7 +65,22 @@ def translate_error(error: Type[Exception], profile_id: int) -> Result: return Result(valid=False, error_type=ResultError.MISSING_SOFTWARE, message=error_msg) else: - return Result(valid=False, error_type=ResultError.UNKNOWN, message="An unknown error occured.") + # Extract the traceback from the exception object + tb = error.__traceback__ + + # extract_tb returns a list of frames; the LAST one [-1] is where the error happened + summary = traceback.extract_tb(tb) + if summary: + last_frame = summary[-1] + func_name = last_frame.name # The function name + line_no = last_frame.lineno # The line number + file_name = last_frame.filename # The file path + else: + func_name, line_no, file_name = "Unknown", "Unknown", "Unknown" + + error_msg = f"Error in {func_name} at {file_name}:{line_no} -> {str(error)}" + logger.error(error_msg) + return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_msg) @type_checker diff --git a/core/infrastructure/global_state.py b/core/infrastructure/global_state.py index a8e9fa1..bf26d5c 100644 --- a/core/infrastructure/global_state.py +++ b/core/infrastructure/global_state.py @@ -1,5 +1,6 @@ from core.models.system.SystemState import SystemState from core.models.session.SessionState import SessionState +from core.models.enum_models import ProtocolChoice from dataclasses import dataclass, field from typing import Optional @@ -7,6 +8,7 @@ from typing import Optional @dataclass class GlobalState: systemwide_active: bool = False + protocol: ProtocolChoice = None firewalled: bool = False dns_set: bool = False systemwide_profile_id: Optional[int] = None @@ -37,6 +39,7 @@ class ProfileStateRepository: def turn_off_system_connection(self): """Clear the system-wide memory cache.""" self.global_state.systemwide_profile_id = None + self.global_state.protocol = None self.global_state.firewalled = False self.global_state.dns_set = False self.global_state.systemwide_active = False @@ -45,6 +48,7 @@ class ProfileStateRepository: def sync_system_state(self, state: SystemState): """Called by SystemStateController to update memory cache.""" self.global_state.systemwide_profile_id = state.profile_id + self.global_state.protocol = state.protocol self.global_state.firewalled = state.firewalled self.global_state.dns_set = state.dns_set if state.profile_id: diff --git a/core/infrastructure/global_state_coordinator.py b/core/infrastructure/global_state_coordinator.py index 64d4a61..1c2060a 100644 --- a/core/infrastructure/global_state_coordinator.py +++ b/core/infrastructure/global_state_coordinator.py @@ -1,11 +1,36 @@ +from core.services.networking.general_connection_tools.connection_enable import establish_connection +from core.controllers.ConfigurationController import ConfigurationController from core.infrastructure.global_state import ProfileStateRepository +from core.controllers.ProfileController import ProfileController from core.controllers.SystemStateController import SystemStateController from core.controllers.SessionStateController import SessionStateController +from core.models.system.SystemState import SystemState +from core.models.enum_models import ProtocolChoice from core.models.Result import Result, ResultError +from core.services.networking.systemwide import killswitch +from core.services.networking.systemwide.general_tools import interface_tools +from core.services.networking.systemwide.general_tools.test_any_protocol import test_ip_systemwide_for_any_protocol +from core.services.networking.general_connection_tools.cross_protocol import kill_any_protocol +from core.services.networking.systemwide.wireguard.wg_firewall_dns import revert_dns +from core.services.networking.systemwide.dns_tools.process_dns_result import quick_domain_dns_check +from core.observers.Notifications import Notifications, EventType +from essentials.observers.ConnectionObserver import ConnectionObserver +from core.Constants import Constants from core.errors.logger import logger +# generic +from functools import partial +SINGBOX_PROTOCOLS = (ProtocolChoice.HYSTERIA2, ProtocolChoice.VLESS) + +def verify_systemwide_state(current_state: SystemState) -> bool: + """Confirm a valid connection, to update GlobalState (for UI)""" + state_repo = ProfileStateRepository() + state_repo.sync_system_state(current_state) + state_repo.verified = True + Notifications.emit(EventType.SYSTEM_ENABLE, current_state.profile_id) + def is_systemwide_on() -> bool: """Returns a boolean if systemwide is connected. True is Yes, False is No""" state_repo = ProfileStateRepository() @@ -36,6 +61,125 @@ def is_connected(profile_id: int) -> bool: return state_repo.is_profile_active(profile_id) +def coordinate_evaluation(connection_observer: ConnectionObserver) -> Result: + """See if a connection can be salvaged. If not, rip it down & start over.""" + + # Claimed JSON state: + if not SystemStateController.exists(): + error_msg = "State vanished inbetween the previous error prompt and now. That state is an external file, so that is technically possible, but a critical error." + logger.error(error_msg) + return Result(valid=False, error_type=ResultError.MISSING_DATA, message=error_msg) + + json_state = SystemStateController.get() + protocol = json_state.protocol + profile_id = json_state.profile_id + firewalled = json_state.firewalled + + logger.info("Extracted JSON state data") + + + # full profile: + profile = ProfileController.get(profile_id) + if not profile: + error_msg = "Critical Failure where the profile to evaluate the connection can't be found. Resetting state." + logger.error(error_msg) + SystemStateController.dissolve() + killed_firewall = killswitch.disarm() + logger.info(f"killed_firewall: {killed_firewall.valid}") + reverted_dns = revert_dns() + logger.info(f"reverted_dns is {reverted_dns}") + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) + logger.info("Got the relevant profile. Doing a pre-flight check") + + #----------------------- + # INTERFACE + #----------------------- + if protocol == ProtocolChoice.WIREGUARD: + interface_up = interface_tools.exists("wg") + elif protocol in SINGBOX_PROTOCOLS: + interface_up = interface_tools.exists(Constants.SINGBOX_TUN_IF) + else: + interface_up = False + logger.error("Unsupported protocol in systemwide state.") + + # RULE OUT: + if not interface_up: + logger.info("Doing a new connection due to the interface being down") + return establish_connection(profile, ignore={}, connection_observer=connection_observer) + + #----------------------- + # FIREWALL + #----------------------- + # Previously User requested Configuration requests + user_wants_firewall = ConfigurationController.get_firewall_setting() + + # Wireguard is voluntary opt-in, so they might not have the script installed, + if not user_wants_firewall and protocol == ProtocolChoice.WIREGUARD: + firewall_on = False + # unless they explicitly opted out with WG, then check: + else: + if firewalled: + firewall_on = killswitch.status() + else: + firewall_on = False + + #----------------------- + # # DNS + #----------------------- + if protocol == ProtocolChoice.WIREGUARD: + target_interface = "wg" + elif protocol in SINGBOX_PROTOCOLS: + target_interface = Constants.SINGBOX_TUN_IF + else: + target_interface = "GARBAGE_FALSE" + logger.error("Unsupported protocol in systemwide state.") + + # this doesn't need sudo to run, and its just regular commands (no script setup): + dns_on = quick_domain_dns_check(target_interface=target_interface) + + logger.info("Evaluation Data Gathered") + + # ------------------ + # DO API CALL & EVALUATE + # ------------------ + test_result = test_ip_systemwide_for_any_protocol(profile_id=profile_id, protocol=protocol, connection_observer=connection_observer) + + if not test_result.valid: + logger.info("IP address check was not valid, so we're doing a new connection.") + return establish_connection(profile, ignore={}, connection_observer=connection_observer) + + if firewall_dns_in_line_with_expectations(firewalled=firewall_on, dns_set=dns_on, protocol=protocol): + logger.info("IP address is connected and firewall & DNS expectations match reality. Update/Verify Global State that it's on") + verify_systemwide_state(current_state=json_state) # this notifies the UI + return Result(valid=True) + + logger.error("Current connection did not meet the criteria to continue to use it as is.") + return establish_connection(profile, ignore={}, connection_observer=connection_observer) + + +def firewall_dns_in_line_with_expectations(firewalled: bool, dns_set: bool, protocol: ProtocolChoice) -> bool: + """We assume the IP address is okay. Is that enough?""" + + if protocol != ProtocolChoice.WIREGUARD: + if firewalled and dns_set: + return True + else: + logger.info("Only wireguard is optional firewall/dns.") + return False + + # Previously User requested Configuration requests + user_wants_firewall = ConfigurationController.get_firewall_setting() + user_wants_dns = ConfigurationController.get_dns_setting() + + if user_wants_firewall and not firewalled: + return False + + if user_wants_dns and not dns_set: + return False + + return True + + def evaluate_state_on_startup() -> Result: """When the app first starts, evaluate pre-existing state""" @@ -44,9 +188,10 @@ def evaluate_state_on_startup() -> Result: #----------------------- systemwide_active = SystemStateController.get() if systemwide_active: - error_msg = "A pre-existing systemwide state is active." - logger.error(error_msg) - return Result(valid=False, error_type=ResultError.STATE_CONFLICT, message=error_msg) + error_msg = "The previous system-wide connection was not correctly turned off before the app was killed. Do you want to evaluate it? (no blindly kills it)" + no_option = partial(kill_any_protocol, current_state=systemwide_active, active_protocol=systemwide_active.protocol) + return Result(valid=False, no_option=no_option, error_type=ResultError.ZOMBIE_CONNECTION, message=error_msg) + #----------------------- # SESSION @@ -58,5 +203,3 @@ def evaluate_state_on_startup() -> Result: SessionStateController.remove(each_profile.id) return Result(valid=True) - - diff --git a/core/infrastructure/plumbing.py b/core/infrastructure/plumbing.py index a353d98..359e7cb 100644 --- a/core/infrastructure/plumbing.py +++ b/core/infrastructure/plumbing.py @@ -126,6 +126,13 @@ class ThreadPlumbing: self.attempt_count += 1 self._spawn_action(passed_result.function_choice) else: + # if the result originally prescribed it's own "no option": + no_option = passed_result.no_option + if no_option: + no_option() + return + + # otherwise do the original UI call: if self.ui_caller: self.ui_caller(passed_result) @@ -136,10 +143,12 @@ class ThreadPlumbing: self.navigator.request_navigation.emit(auto_navigate_page) + def normalize_seq(s): """Ensures input is a tuple regardless of if it was a list or tuple.""" return tuple(s) if s is not None else () + def resolve_prescription(result: Result, ENUM_MAP: dict, *args, **kwargs): """ Find the correct solution to the type of enum problem, @@ -181,7 +190,30 @@ def resolve_prescription(result: Result, ENUM_MAP: dict, *args, **kwargs): if not function_requirements: return func - # Priority 3: Use original args/kwargs + # Priority 3: Does it want the result itself? + if 'result' in function_requirements: + logger.info("result required; baking result into partial.") + + # 1. Prepare the arguments we have available right now + # Start with default_args from the ENUM_MAP if they exist + current_args = {} + if isinstance(default_args, dict): + current_args.update(default_args) + + # Inject the result object + current_args['result'] = result + + # 2. Use bind_partial so it doesn't fail if other args are still missing + bound = sig.bind_partial(**current_args) + + # 3. Fill in any function-defined defaults + bound.apply_defaults() + + # 4. Return the partial with the bound arguments + return partial(func, **bound.arguments) + + + # Priority 4: Use original args/kwargs try: bound = sig.bind(*args, **kwargs) bound.apply_defaults() @@ -189,7 +221,7 @@ def resolve_prescription(result: Result, ENUM_MAP: dict, *args, **kwargs): except TypeError: pass # Not enough, move to priority 4 - # Priority 4: Use original args/kwargs PLUS defaults + # Priority 5: Use original args/kwargs PLUS defaults if default_args is None: return False diff --git a/core/models/Result.py b/core/models/Result.py index 3888b2b..66b6486 100644 --- a/core/models/Result.py +++ b/core/models/Result.py @@ -24,6 +24,7 @@ class ResultError(Enum): EXPIRED_SUB = "expired_sub" ACTIVE_BILLING = "active_billing" STATE_CONFLICT = "state_conflict" + ZOMBIE_CONNECTION = "zombie_connection" TICKET = "ticket" NO_TICKET = "no_ticket" RANDOM_TICKETS_OFF = "random_tickets_off" @@ -60,6 +61,7 @@ class Result(): message: Optional[str] = None goal: Optional[str] = None function_choice: Optional[partial] = None + no_option: Optional[partial] = None navigate_to: Optional[str] = None def user_message(self) -> str: diff --git a/core/models/system/SystemState.py b/core/models/system/SystemState.py index 532904f..43e2725 100644 --- a/core/models/system/SystemState.py +++ b/core/models/system/SystemState.py @@ -70,3 +70,16 @@ class SystemState: @staticmethod def __get_state_path(): return Constants.HV_STATE_HOME + + + +# @dataclass +# class StateEvaluation: +# interface: bool +# firewalled: bool +# dns_set: bool +# ip_address_correct: Optional[bool] = None +# firewall_killed_internet: Optional[bool] = None +# dns_killed_internet: Optional[bool] = None +# has_internet: Optional[bool] = None +# killed_vpn: Optional[bool] = None diff --git a/core/services/networking/systemwide/dns_tools/process_dns_result.py b/core/services/networking/systemwide/dns_tools/process_dns_result.py index 2d3f357..27efbe4 100644 --- a/core/services/networking/systemwide/dns_tools/process_dns_result.py +++ b/core/services/networking/systemwide/dns_tools/process_dns_result.py @@ -36,10 +36,11 @@ def process_ALL_dns_results(target_interface: str) -> Result: # ================= START LOOP ===================== for each_command in list_of_commands: - each_SearchResult = process_ONE_dns_result( - which_command=each_command, - target_interface=target_interface, - ) + if each_command: + each_SearchResult = process_ONE_dns_result( + which_command=each_command, + target_interface=target_interface, + ) # add the command to the SearchResult here, so it's only declared once, each_SearchResult.which_command_was_run = each_command @@ -93,6 +94,15 @@ def orchestrate_dns_check(target_interface: str) -> Result: return Result(valid=False, data=target_not_found) +def quick_domain_dns_check(target_interface: str) -> bool: + each_SearchResult = process_ONE_dns_result( + which_command="domain", + target_interface=target_interface, + ) + if each_SearchResult.valid: + return True + else: + return False diff --git a/core/services/networking/systemwide/general_tools/test_any_protocol.py b/core/services/networking/systemwide/general_tools/test_any_protocol.py new file mode 100644 index 0000000..9897e37 --- /dev/null +++ b/core/services/networking/systemwide/general_tools/test_any_protocol.py @@ -0,0 +1,64 @@ + +from core.services.networking.general_connection_tools.config_tools import extract_endpoint_ip +from core.services.networking.general_connection_tools import ip +from essentials.observers.ConnectionObserver import ConnectionObserver +from core.models.enum_models import ProtocolChoice +from core.utils.basic_operations.write_or_read_from_json import get_value_from_json_file +from core.models.Result import Result, ResultError +from core.errors.logger import logger +from core.Constants import Constants + +import os + + +def get_singbox_ip(profile_id: int) -> str|None: + """Get the IP address for a Singbox based profile based on local disk.""" + config_path = f'{Constants.HV_SYSTEM_PROFILE_CONFIG_PATH}/{profile_id}/proxy.json' + logger.info(f"config_path is {config_path}") + + if not os.path.isfile(config_path): + return None + + server_ip = get_value_from_json_file(filepath=config_path, category="outbounds", key="server") + return server_ip + + +def test_ip_systemwide_for_any_protocol(profile_id: int, protocol: ProtocolChoice, connection_observer: ConnectionObserver) -> Result: + """Compare the real vs expected IP address.""" + + # protocol = profile.type + # profile_id = profile.id + + # what's expected: + if protocol == ProtocolChoice.WIREGUARD: + expected_ip_address = extract_endpoint_ip(profile_id) + elif protocol in (ProtocolChoice.VLESS, ProtocolChoice.HYSTERIA2): + expected_ip_address = get_singbox_ip(profile_id=profile_id) + else: + expected_ip_address = None + + if not expected_ip_address: + error_msg = "Could not get the expected IP address from local files." + logger.error(error_msg) + return Result(valid=False, error_type=ResultError.MISSING_DATA, message=error_msg) + + # reality: + ip_result = ip.test(connection_observer) + + if not ip_result.valid: + logger.error(ip_result.message) + return ip_result # result object + + real_ip = ip_result.data + + print(f"Expected: {expected_ip_address} Reality: {real_ip}") + + if real_ip == expected_ip_address: + return Result(valid=True) + else: + error_msg = f"IP address does NOT match. Expected: {expected_ip_address} vs Reality: {real_ip}" + return Result(valid=False, error_type=ResultError.LEAK_ISSUE, message=error_msg) + + + + diff --git a/core/services/networking/systemwide/systemwide_wireguard.py b/core/services/networking/systemwide/systemwide_wireguard.py index c604724..58ce16f 100644 --- a/core/services/networking/systemwide/systemwide_wireguard.py +++ b/core/services/networking/systemwide/systemwide_wireguard.py @@ -212,7 +212,7 @@ def __establish_system_connection( current_state.dns_set = did_dns_work SystemStateController.update_or_create(current_state) - Notifications.emit(EventType.SYSTEM_ENABLE, profile_id) + Notifications.emit(EventType.SYSTEM_ENABLE, profile.id) return Result(valid=True, data=current_state) diff --git a/core/utils/type_checker.py b/core/utils/type_checker.py index bb645b5..7af5be8 100644 --- a/core/utils/type_checker.py +++ b/core/utils/type_checker.py @@ -1,49 +1,107 @@ -from core.models.Result import Result +from core.models.Result import Result, ResultError from core.errors.logger import logger import inspect from functools import wraps -from typing import Any +from typing import get_origin, get_args, Union +import types # Required for UnionType (the | operator) + +def _get_type_error_details(check_type, value, origin, args_types): + """ + Helper to construct human-readable names for the expected and actual types. + """ + # Case 1: User used Type[T] but passed an instance of T + if origin is type: + inner_name = getattr(args_types[0], '__name__', 'the specified type') + return f"a class/type (subclass of {inner_name})", f"an instance of {type(value).__name__}" + + # Case 2: check_type is a tuple (common for Unions) + if isinstance(check_type, tuple): + type_names = ", ".join([getattr(t, '__name__', str(t)) for t in check_type]) + return f"one of ({type_names})", f"type {type(value).__name__}" + + # Case 3: Standard single type + type_name = getattr(check_type, '__name__', str(check_type)) + return type_name, f"type {type(value).__name__}" + +def _validate_inner_types(value, origin, args_types, func, name): + """ + Helper to handle deep validation for generics like List[int] or Type[T]. + Returns (is_valid, error_message). + """ + if not args_types: + return True, None + + inner_type = args_types[0] + + # Handle Type[T]: value must be a subclass of T + if origin is type: + if not issubclass(value, inner_type): + inner_name = getattr(inner_type, '__name__', str(inner_type)) + return False, f"For function {func}, Argument '{name}' must be a subclass of {inner_name}, got {value.__name__}" + return True, None + + # Handle Collections: list, set, tuple + if isinstance(value, (list, set, tuple)): + for item in value: + inner_origin = get_origin(inner_type) + + # Handle Unions inside collections (e.g., List[Union[int, str]]) + if inner_origin is Union or (hasattr(types, "UnionType") and inner_origin is types.UnionType): + if not isinstance(item, get_args(inner_type)): + return False, f"For function {func}, Argument '{name}' contains element {type(item).__name__}, expected {inner_type}" + # Handle standard types inside collections + elif not isinstance(item, inner_type): + inner_name = getattr(inner_type, '__name__', str(inner_type)) + return False, f"For function {func}, Argument '{name}' contains element {type(item).__name__}, expected {inner_name}" + + return True, None def type_checker(func): - """ - Type check wrapper. - This type checks individual functions it's put on, and returns a Result object on failure. - """ @wraps(func) def wrapper(*args, **kwargs): - # 1. Introspect the wrapped function sig = inspect.signature(func) + # 1. Bind arguments to names to handle both positional and keyword args try: - # Bind the provided args/kwargs to the function's signature bound_args = sig.bind(*args, **kwargs) bound_args.apply_defaults() except TypeError as e: - # This handles cases where the wrong number of arguments are passed error_msg = f"For {func}, {str(e)}" logger.error(error_msg) return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) - # 2. Perform type validation + # 2. Validate each argument against its type hint for name, value in bound_args.arguments.items(): expected_type = sig.parameters[name].annotation - # Skip parameters that don't have a type hint + # Skip if no type hint is provided if expected_type is inspect.Parameter.empty: continue - if not isinstance(value, expected_type): - error_msg = f"For function {func}, Argument '{name}' must be {expected_type.__name__}, got type {type(value).__name__}" + origin = get_origin(expected_type) + args_types = get_args(expected_type) + + # Determine the type to use for isinstance() + # Handles both typing.Union and the Python 3.10+ | operator (UnionType) + is_union = origin is Union or (hasattr(types, "UnionType") and origin is types.UnionType) + check_type = args_types if is_union else (origin if origin is not None else expected_type) + + # Basic type validation + if not isinstance(value, check_type): + type_name, actual_name = _get_type_error_details(check_type, value, origin, args_types) + error_msg = f"For function {func}, Argument '{name}' must be {type_name}, got {actual_name}" logger.error(error_msg) - return Result( - valid=False, - error_type=ResultError.INVALID_INPUT, - message=error_msg - ) + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) + + # Deep validation for generics (e.g., List[int], Type[MyClass]) + # We only do this if it's not a Union, as Unions were handled by the basic isinstance check above + if origin is not None and not is_union: + is_valid, error_msg = _validate_inner_types(value, origin, args_types, func, name) + if not is_valid: + logger.error(error_msg) + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=error_msg) - # 3. If all checks pass, call the original function return func(*args, **kwargs) - return wrapper