Plumbing switched to regular thread (not executor). Introduced Port checking and changing as part of wireguard failure

core/controllers/ConfigurationController.py

@@ -177,3 +177,21 @@ class ConfigurationController:
         configuration = ConfigurationController.get()
         return configuration.tickets_setup
 
+    @staticmethod
+    def do_we_allow_port_prompts() -> bool:
+        configuration = ConfigurationController.get()
+
+        if configuration.allow_port_prompts is None:
+            print("Setting allow ports to true")
+            configuration.allow_port_prompts = True
+            configuration.save()
+
+        return configuration.allow_port_prompts
+
+    @staticmethod
+    def change_port_prompts(new_value) -> bool:
+        configuration = ConfigurationController.get()
+        configuration.allow_port_prompts = new_value
+        configuration.save()
+        return True
+

core/infrastructure/global_state.py

@@ -14,6 +14,7 @@ class GlobalState:
     systemwide_profile_id: Optional[int] = None
     session_profile_ids: list[int] = field(default_factory=list)
     verified: bool = False
+    changed_port: bool = False
 
 class ProfileStateRepository:
     _instance = None

core/infrastructure/global_state_coordinator.py

@@ -24,6 +24,7 @@ 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()

core/infrastructure/plumbing.py

@@ -2,19 +2,17 @@ from core.infrastructure.thread_dispatch import MainThreadDispatcher
 from core.infrastructure.error_popup import PopupProtocol, StackErrorProtocol
 from core.infrastructure.service_registry import ServiceLocator
 from core.errors.translate_error import get_traceback_line
-from core.errors.ErrorStack import ErrorStack
 from core.models.Result import Result, ResultError
 from core.errors.logger import logger
 
 # generic:
-from concurrent.futures import ThreadPoolExecutor, Future
+import threading
 from typing import Callable, Any, Optional, Type
 import inspect
 from functools import partial
 import sys
 
 MAX_ATTEMPTS = 2
-executor = ThreadPoolExecutor(max_workers=1)
 
 class ThreadPlumbing:
     """
@@ -33,50 +31,55 @@ class ThreadPlumbing:
         self.args = args
         self.kwargs = kwargs
         self.attempt_count = 0
-    
-    def execute(self):
-        """Entry point: kick off the first attempt"""
-        self.attempt_count += 1
-        self._spawn_action(action=self.action, first_time=True)
-    
-    def _spawn_action(self, action: Callable, first_time: bool = None):
-        """Spawn action on executor"""
-        if first_time:
-            future = executor.submit(action, *self.args, **self.kwargs) # original args/kwargs
-        else:
-            future = executor.submit(action) # this is a pre-loaded partial
-        future.add_done_callback(self._on_complete)
-    
-    def _on_complete(self, future: Future):
-        """Handle thread completion"""
-        try:
-            logger.info("The plumbing module started it's 'on complete' phase with this result:")
-            result = future.result()
-            if not isinstance(result, Result):
-                result = convert_to_result(result)
 
-            logger.info(result)
+    def execute(self, task = None):
-            if not result.valid:
+        """Kicks off the action in a background thread"""
-                if not result.function_choice:
-                    result.function_choice = resolve_prescription(result, self.ENUM_MAP, *self.args, **self.kwargs)
-                    logger.info(f"result.function_choice is {result.function_choice}")
-                else:
-                    logger.info("Using the pre-existing function choice for this error.")
 
-        except Exception as e:
+        # 0. If there's no task passed in, get it from the object initalization,
-            error_msg = f"There was an error: {str(e)}"
+        if not task:
-            logger.error(error_msg)
+            # 1. Create the partial immediately in the main thread
-            error_stack = get_traceback_line(e)
+            task = partial(self.action, *self.args, **self.kwargs)
-            error_msg = error_stack.message_for_user
+        
-            logger.error(error_msg)
+        # 2. Define a simple wrapper to run the task and then handle the result
-            result = Result(valid=False, error_stack=error_stack, error_type=ResultError.UNKNOWN, message=error_msg)
+        def wrapper():
+            try:
+                result = task()        # Run the partial
+                logger.info(f"post complete result here is {result}")
+            except Exception as e:
+                error_msg = f"There was an error: {str(e)}"
+                logger.error(error_msg)
+                error_stack = get_traceback_line(e)
+                error_msg = error_stack.message_for_user
+                logger.error(error_msg)
+                result = Result(valid=False, error_stack=error_stack, error_type=ResultError.UNKNOWN, message=error_msg)
+            finally:
+                self._on_complete(result) # Process the result
 
-        finally:
+        # 3. Spin up a single thread and start it
-            # Back to main thread
+        thread = threading.Thread(target=wrapper, daemon=True)
-            print(f"result in plumbing is {result}")
+        thread.start()
-            self.ui_dispatcher.run_on_main_thread(self._on_result_ready, result)
+
-    
+    def _on_complete(self, result: Result):
-    def _on_result_ready(self, result: Result):
+        """Get a potential function prescription and return it to the main thread."""
+
+        #----------------------------
+        # Function Prescription:
+        #----------------------------
+        if not result.valid:
+            if not result.function_choice:
+                result.function_choice = resolve_prescription(result, self.ENUM_MAP, *self.args, **self.kwargs)
+                logger.info(f"result.function_choice is {result.function_choice}")
+            else:
+                logger.info("Using the pre-existing function choice for this error.")
+
+        #----------------------------
+        # Back to main thread
+        #----------------------------
+        # logger.info(f"result in plumbing background thread is {result}")
+        self.ui_dispatcher.run_on_main_thread(self._back_on_main_thread, result)
+        # print("ran on main thread")
+
+    def _back_on_main_thread(self, result: Result):
         """Decide: retry, show popup, or final callback"""
 
         logger.info(f"We are on the main thread with result {result}")
@@ -92,6 +95,14 @@ class ThreadPlumbing:
                 sys.exit()
             return
 
+        #----------------------------
+        # NAVIGATE AWAY
+        #----------------------------
+        # If the result sender explicitly asked us to navigate to a page with a ui callback, then bypass the popup:
+        if result.error_type == ResultError.NAVIGATE_TO and self.ui_caller:
+            logger.info("Plumbing was asked to NAVIGATE_TO so skipping to the ui_caller..")
+            self.ui_caller(result)
+
         #----------------------------
         # MAX TRIES
         #----------------------------
@@ -103,16 +114,9 @@ class ThreadPlumbing:
                 self.ui_caller(result)
             return
 
-        # does this have an auto-navigate page?
+        #----------------------------
-        # if self.navigator and self.auto_navigate_dict:
+        # SHOW CHOICE & POPUP
-        #     logger.info("checking auto-navigate..")
+        #----------------------------
-        #     self.check_auto_navigation(result)
-
-        # If the result sender explicitly asked us to navigate to a page with a ui callback, then bypass the popup:
-        if result.error_type == ResultError.NAVIGATE_TO and self.ui_caller:
-            logger.info("Plumbing was asked to NAVIGATE_TO so skipping to the ui_caller..")
-            self.ui_caller(result)
-
         # do we need to decide on an action?
         choice = bool(result.function_choice)
         logger.info(f"Do we have to make a choice? {choice}")
@@ -133,6 +137,9 @@ class ThreadPlumbing:
             popup.show()
             logger.info("Showed the error popup.")
 
+        #----------------------------
+        # COMPLETE
+        #----------------------------
         # otherwise return it as is:
         if self.ui_caller:
             logger.info("Returning from plumbing..")
@@ -141,26 +148,34 @@ class ThreadPlumbing:
 
     def _on_popup_answer(self, user_choice: bool, passed_result: Result):
         """User responded to popup"""
+
+        #---------------------
+        # RUN THE CHOICE:
+        #---------------------
         if user_choice and passed_result.function_choice:
             self.attempt_count += 1
-            self._spawn_action(passed_result.function_choice)
+            self.execute(passed_result.function_choice)
+
         else:
+            #---------------------
+            # OR RUN THE "NO" CHOICE:
+            #---------------------
             # if the result originally prescribed it's own "no option":
             no_option = passed_result.no_option
             if no_option:
                 no_option()
                 return
 
+            #---------------------
+            # OR Just return back,
+            #---------------------
             # otherwise do the original UI call:
             if self.ui_caller:
                 self.ui_caller(passed_result)
 
-    def check_auto_navigation(self, result: Result):
-        auto_navigate_page = self.auto_navigate_dict.get(result.error_type)
-        if auto_navigate_page is None:
-            return
 
-        self.navigator.request_navigation.emit(auto_navigate_page)
+
+# STANDALONE HELPERS:
 
 
 def normalize_seq(s):
@@ -302,3 +317,44 @@ def convert_to_result(result: Any) -> Result:
 def take_plumbing_action(action: Callable, ui_dispatcher: MainThreadDispatcher, error_implementation: Type[PopupProtocol], ui_caller: Callable, *args, **kwargs):
     controller = ThreadPlumbing(action, ui_dispatcher, ui_caller, *args, **kwargs)
     controller.execute()
+
+
+
+
+
+# def execute(self):
+#     """Entry point: kick off the first attempt"""
+#     self.attempt_count += 1
+#     self._spawn_action(action=self.action, first_time=True)
+
+# def _spawn_action(self, action: Callable, first_time: bool = None):
+#     """Spawn action on executor"""
+#     if first_time:
+#         future = executor.submit(action, *self.args, **self.kwargs) # original args/kwargs
+#     else:
+#         future = executor.submit(action) # this is a pre-loaded partial
+#     future.add_done_callback(self._on_complete)
+
+
+# result = future.result()
+# if not isinstance(result, Result):
+#     result = convert_to_result(result)
+
+
+
+
+
+# Auto navigate:
+
+# does this have an auto-navigate page?
+# if self.navigator and self.auto_navigate_dict:
+#     logger.info("checking auto-navigate..")
+#     self.check_auto_navigation(result)
+
+
+# def check_auto_navigation(self, result: Result):
+#     auto_navigate_page = self.auto_navigate_dict.get(result.error_type)
+#     if auto_navigate_page is None:
+#         return
+
+#     self.navigator.request_navigation.emit(auto_navigate_page)

core/models/Configuration.py

@@ -35,6 +35,7 @@ class Configuration(BaseModel):
     singbox: Optional[str] = None
     tickets_setup: Optional[bool] = False
     use_random: Optional[bool] = False
+    allow_port_prompts: Optional[bool] = True
 
 
     @field_validator('last_synced_at', mode='before')

core/models/Result.py

@@ -56,6 +56,7 @@ class ResultError(Enum):
     API_DOWN = "api_down"
     NODE_ERROR = "node_error"
     NO_INTERNET = "no_internet"
+    PORT = "port"
     NAVIGATE_TO = "navigate_to"
     UNKNOWN = "unknown"
 

core/services/keys_and_verifications/wireguard_keys.py

@@ -1,3 +1,5 @@
+from core.controllers.ConfigurationController import ConfigurationController
+from core.infrastructure.global_state import ProfileStateRepository
 from core.utils.get_raw_string import get_string
 from core.models.manage.server_reply_tools import api_call_and_extraction, ApiCall
 # from core.services.networking.httpx import connect
@@ -17,13 +19,18 @@ from core.errors.logger import logger
 #from core.models.manage.server_reply_tools import format_server_response
 # from core.models.pydantic_models.ServerReply import ServerReply, ServerError
 from core.models.Result import Result, ResultError
+from core.Errors import ProfileModificationError, CommandNotFoundError
 
 from typing import Optional
 import base64
 import re
+import random
+from functools import partial
 
 MIN_MTU = 1280
 
+PORT_CHOICES = [48271, 52428, 51820]
+
 def register_wireguard_session(
     profile: Profile, 
     connection_observer: Optional[ConnectionObserver] = None
@@ -181,3 +188,113 @@ def expand_newlines(text):
     empty line between every non-empty line of text.
     """
     return "\n\n".join([line for line in text.splitlines() if line.strip()])
+
+
+def get_port(config: str) -> int:
+    """Get the wireguard port if you have a config already."""
+    # This regex searches for the "Endpoint" key, ignores optional whitespace around the '=',
+    # skips the IP/hostname part until the colon, and captures the digits following it.
+    match = re.search(r"Endpoint\s*=\s*[^:]+:(\d+)", config, re.IGNORECASE)
+    if match:
+        return int(match.group(1))
+    return None
+
+def change_port(new_port: int, config: str) -> str:
+    # This regex captures everything from the start of the line up to the colon 
+    # in the first group, then matches the existing digits of the port.
+    pattern = r"(Endpoint\s*=\s*[^:]+):(\d+)"
+    
+    # We use a lambda function as the replacement to insert the new_port variable 
+    # while preserving the captured IP/hostname in group 1 (\1).
+    return re.sub(pattern, lambda m: f"{m.group(1)}:{new_port}", config, flags=re.IGNORECASE)
+
+
+def get_wg_port(profile: Profile) -> Result:
+    """Get the Wireguard port with a given profile."""
+    config_path = profile.get_wireguard_configuration_path()
+    if not config_path:
+        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="Can't get the path of the config")
+
+    config_data = get_string(config_path)
+    if not config_data:
+        return Result(valid=False, error_type=ResultError.MISSING_DATA, message="Missing the config data, or could not read it.")
+
+    port = get_port(config_data)
+    if port:
+        return Result(valid=True, data=port)
+    else:
+        return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="Config is not in the correct format to be read.")
+
+
+def get_random_different_port(current_port: int) -> int|None:
+    """Get a different port (that's available) from the given choices."""
+    while True:
+        new_port = random.choice(PORT_CHOICES)
+        if str(new_port) != str(current_port):
+            return new_port
+
+
+def change_wg_port(profile: Profile = None, desired_port: Optional[int] = None) -> Result:
+    """Change a profile's Wireguard port."""
+
+    logger.info(f"desired_port is {desired_port}")
+    config_path = profile.get_wireguard_configuration_path()
+    config_data = get_string(config_path)
+    current_port = get_port(config_data)
+    logger.info(f"current_port is {current_port}")
+    
+    if desired_port and desired_port != current_port and desired_port in PORT_CHOICES:
+        change_to = desired_port
+    else:
+        change_to = get_random_different_port(current_port)
+        logger.info(f"Changing the port to {change_to}")
+
+    new_config = change_port(new_port=change_to, config=config_data)
+
+    # sudo prompts:
+    try:
+        profile.attach_wireguard_configuration(new_config)
+        logger.info("changed the port.  updating state..")
+
+        # UPDATE STATE:
+        state_repo = ProfileStateRepository()
+        state_repo.global_state.changed_port = True
+
+        return Result(valid=True, data=change_to)
+    except ProfileModificationError:
+        return Result(valid=False, error_type=ResultError.PERMISSON, message="Failed to swap the port due to permissions")
+    except CommandNotFoundError:
+        return Result(valid=False, error_type=ResultError.MISSING_SOFTWARE, message="Missing the command")
+
+
+def port_was_changed_already_today() -> bool:
+    state_repo = ProfileStateRepository()
+    return state_repo.global_state.changed_port
+
+
+def offer_port_change(profile: Profile) -> Result:
+    """Offer the user the option to change the port"""
+    port_result = get_wg_port(profile)
+
+    if not port_result.valid:
+        return port_result
+
+    current_port = port_result.data
+    desired_port = get_random_different_port(current_port)
+    
+    func_choice = partial(change_wg_port, profile=profile, desired_port=desired_port)
+    return Result(valid=False, function_choice=func_choice, error_type=ResultError.PORT, message=f"Failed to connect.  Do you want to try to change the port to {desired_port}?")
+
+
+def should_we_try_to_change_port():
+    """Returns a boolean if we should change the port"""
+    if not ConfigurationController.do_we_allow_port_prompts():
+        logger.info("We do not allow port changes")
+        return False
+
+    state_repo = ProfileStateRepository()
+    if state_repo.global_state.changed_port:
+        return False
+    else:
+        return True
+

core/services/networking/general_connection_tools/connection_enable.py

@@ -32,6 +32,7 @@ from core.observers.ConnectionObserver import ConnectionObserver
 from core.controllers.SystemStateController import SystemStateController
 from core.observers.TicketObserver import TicketObserver
 
+
 def establish_connection(
     profile: Union[SessionProfile, SystemProfile], 
     ignore: tuple[type[Exception]] = (), 
@@ -39,7 +40,7 @@ def establish_connection(
     ticket_observer: Optional[TicketObserver] = None # Passed in only for assassin's link to tickets in ConnectionController.establish_session_connection
 ):
     """Establish a connection for the given profile."""
-    
+
     logger.info("[CONNECTION] Checking subscription..")
     if not profile.assassin: # Note: Assassin does it's own billing. This would be redundant
         activate_subscription(profile=profile, connection_observer=connection_observer)

core/services/networking/systemwide/systemwide_wireguard.py

@@ -1,4 +1,4 @@
-from core.services.keys_and_verifications.wireguard_keys import get_mtu, insert_mtu_into_config_on_disk
+from core.services.keys_and_verifications.wireguard_keys import get_mtu, insert_mtu_into_config_on_disk, offer_port_change, should_we_try_to_change_port, port_was_changed_already_today
 from core.services.networking.systemwide.wireguard.mtu_tools import loop_through_testing_mtu, mtu_is_fine, MIN_MTU, MAX_MTU
 from core.services.networking.general_connection_tools.location_evaluation import check_if_location_is_up
 from core.services.networking.api_requests.ApiResponseModel import BackoffStrategy
@@ -11,7 +11,6 @@ from core.services.networking.systemwide.systemwide_utils import extract_wg_inte
 from core.observers.Notifications import Notifications, EventType
 from core.services.networking.systemwide.general_tools.test_any_protocol import test_ip_systemwide_for_any_protocol
 
-
 # from core.services.networking.general_connection_tools.config_tools import extract_endpoint_ip, extract_dns
 from core.services.networking.systemwide import killswitch
 from core.services.networking.systemwide.dns_tools.process_dns_result import orchestrate_dns_check
@@ -292,7 +291,7 @@ def establish_system_connection(
     ignore: tuple[type[Exception]] = (),
     connection_observer: Optional[ConnectionObserver] = None,
     ticket_observer: Optional[TicketObserver] = None # Passed in only for assassin's link to tickets in ConnectionController.establish_session_connection
-    ):
+    ) -> Result|None:
     """
     Establish connection with a maximum of 2 retries.
 
@@ -339,11 +338,16 @@ def establish_system_connection(
             raise ConnectionError(notification)
 
         except (ConnectionError, CalledProcessError) as e:
-            _cleanup_on_error(firewall_setting, dns_setting)
+
-            
+            # single time port is offered forever.
+            if port_was_changed_already_today():
+                # permanent config modify:
+                ConfigurationController.change_port_prompts(False)
+
             if attempt == 1:
                 # First failure: Evaluate Location and Internet without connection,
-                terminate_system_connection(firewall_setting, dns_setting)
+                _cleanup_on_error(firewall_setting, dns_setting)
+
                 location_result = evaluate_location_and_internet(
                     profile=profile, 
                     connection_observer=connection_observer
@@ -364,11 +368,12 @@ def establish_system_connection(
                 continue # Proceed to attempt 2
 
             elif attempt == 2:
-                # Second failure: Evaluate MTU
+                # Second failure: Evaluate MTU WITH connection on
                 current_mtu = get_mtu(profile_id=profile.id)
 
                 if current_mtu != MAX_MTU or not current_mtu:
                     logger.info("MTU is already at the max, no point to change it.")
+                    _cleanup_on_error(firewall_setting, dns_setting)
                     raise ConnectionError("Failed to connect.")
                     # return Result(valid=False, error_type=ResultError.CONNECTION, message="Failed to connect.")
 
@@ -381,6 +386,7 @@ def establish_system_connection(
                         Notifications.emit(EventType.STATUS_UPDATE, notification)
                         logger.info(notification)
                         # If MTU wasn't changed or wasn't applicable, don't bother with attempt 3
+                        _cleanup_on_error(firewall_setting, dns_setting)
                         raise ConnectionError("Failed to connect.")
                         # return Result(valid=False, error_type=ResultError.CONNECTION, message="Failed to connect.")
 
@@ -394,16 +400,23 @@ def establish_system_connection(
                 else:
                     logger.info("Skipping MTU testing, it's not relevant here.")
                 
-                # If MTU wasn't changed or wasn't applicable, don't bother with attempt 3
+                # If Port & MTU wasn't changed or wasn't applicable, don't bother with attempt 3
-                notification = "Failed to connect."
+                # PORT CHANGE:
-                Notifications.emit(EventType.STATUS_UPDATE, notification)
+                if should_we_try_to_change_port():
-                raise ConnectionError(notification)
+                    print("We should try to change port")
-                # return Result(valid=False, error_type=ResultError.CONNECTION, message="Failed to connect.")
+                    _cleanup_on_error(firewall_setting, dns_setting)
+                    return offer_port_change(profile)
+                else:
+                    notification = "Failed to connect."
+                    Notifications.emit(EventType.STATUS_UPDATE, notification)
+                    _cleanup_on_error(firewall_setting, dns_setting)
+                    raise ConnectionError("Failed to connect.")
 
             else:
                 # Attempt 3 failed
                 logger.error("Maximum retries reached. Connection failed.")
                 notification = "Failed to connect."
                 Notifications.emit(EventType.STATUS_UPDATE, notification)
+                _cleanup_on_error(firewall_setting, dns_setting)
                 raise ConnectionError(notification)
                 # return Result(valid=False, error_type=ResultError.CONNECTION, message="Failed to connect.")

core/services/networking/tor_tools/ports.py

@@ -218,4 +218,3 @@ async def kill_9050() -> bool:
         return True
 
     return kill_via_lsof(Constants.DEFAULT_TOR_PORT)
-

core/services/profiles/profile_connections.py

@@ -120,7 +120,6 @@ def enable_profile(
     ticket_observer: TicketObserver = None,
     max_resolution: Optional[str] = None
 ) -> Result:
-
     """Orchestrate Enabling both types of profiles"""
 
     # =========== ALREADY ENABLED ============
@@ -161,20 +160,14 @@ def enable_profile(
                 Notifications.emit(EventType.STATUS_UPDATE, notification)
                 return
 
-            if connection_result.valid:
+            if not connection_result.valid:
-                if profile_observer is not None:
+                return connection_result
-                    profile_observer.notify('enabled', profile)
-                notification = "Enabled."
-                Notifications.emit(EventType.STATUS_UPDATE, notification)
 
-                return Result(valid=True, data=profile.type)
+            if profile_observer is not None:
-            else:
+                profile_observer.notify('enabled', profile)
-                error_msg = connection_result.message
+            notification = "Enabled."
-                logger.error(error_msg)
+            Notifications.emit(EventType.STATUS_UPDATE, notification)
-                notification = "Connection Failed."
+            return Result(valid=True, data=profile.type)
-                Notifications.emit(EventType.STATUS_UPDATE, notification)
-                return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_msg)
-                # raise ProfileActivationError(error_msg)
         except FirewallError:
             return Result(valid=False, error_type=ResultError.FIREWALL, message="Firewall Error.")
         except ConnectionError as e:

core/utils/basic_operations/file_tools.py

@@ -26,7 +26,7 @@ def read_a_file(path_of_file: str) -> str | bool:
     file_path = Path(path_of_file)
 
     try:
-        with open(file_path, "r") as file:
+        with open(file_path, "r", encoding="utf-8") as file:
             content = file.read()
         return content