diff --git a/core/services/networking/general_connection_tools/connection_enable.py b/core/services/networking/general_connection_tools/connection_enable.py index 0398191..6d04cde 100644 --- a/core/services/networking/general_connection_tools/connection_enable.py +++ b/core/services/networking/general_connection_tools/connection_enable.py @@ -1,15 +1,17 @@ -from core.services.networking.general_connection_tools.location_evaluation import check_if_location_is_up -from core.services.networking.api_requests.ApiResponseModel import BackoffStrategy +from core.services.networking.systemwide.encrypted_proxy.launch_proxy import launch_encrypted_proxy +# from core.services.networking.general_connection_tools.location_evaluation import check_if_location_is_up +# from core.services.networking.api_requests.ApiResponseModel import BackoffStrategy from core.services.networking.general_connection_tools.testing_evaluating import system_uses_wireguard_interface from core.services.networking.systemwide.systemwide_wireguard import establish_system_connection, terminate_system_connection from core.services.keys_and_verifications.wireguard_keys import register_wireguard_session -from core.services.subscriptions.subscriptions import activate_subscription, get_encrypted_proxy_billing_code -from core.services.networking.systemwide.encrypted_proxy.configure_singbox import configure_singbox -from core.services.networking.systemwide.encrypted_proxy.singbox_runner import start_singbox -from core.models.Result import Result, ResultError -from core.utils.basic_operations.write_or_read_from_json import get_value_from_json_file -from core.observers.TicketObserver import TicketObserver -from core.services.assassin import assassin_tools +from core.services.subscriptions.subscriptions import activate_subscription +#, get_encrypted_proxy_billing_code +# from core.services.networking.systemwide.encrypted_proxy.configure_singbox import configure_singbox +# from core.services.networking.systemwide.encrypted_proxy.singbox_runner import start_singbox +# from core.models.Result import Result, ResultError +# from core.utils.basic_operations.write_or_read_from_json import get_value_from_json_file +# from core.observers.TicketObserver import TicketObserver +# from core.services.assassin import assassin_tools # If refactored to enums: # from core.models.session.SessionConnection import SessionConnectionTypes @@ -17,7 +19,7 @@ from core.services.assassin import assassin_tools # wireguard_types = [SystemConnectionTypes.WIREGUARD, SessionConnectionTypes.WIREGUARD] from core.errors.logger import logger -from core.errors.exceptions import * +# from core.errors.exceptions import * from core.errors.exceptions import FirewallError from typing import Union, Optional, Callable from core.models.session.SessionProfile import SessionProfile @@ -30,10 +32,6 @@ from core.observers.ConnectionObserver import ConnectionObserver from core.controllers.SystemStateController import SystemStateController from core.observers.TicketObserver import TicketObserver -import os - -QUANTITY_OF_TRIES = 2 - def establish_connection( profile: Union[SessionProfile, SystemProfile], ignore: tuple[type[Exception]] = (), @@ -42,7 +40,7 @@ def establish_connection( ): """Establish a connection for the given profile.""" - logger.info(f"[CONNECTION] Checking subscription..") + 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) @@ -56,10 +54,10 @@ def establish_connection( # ========================================= # SOCKS5 & WIREGUARD # ========================================= - logger.info(f"[CONNECTION] Checking proxy configuration..") + logger.info("[CONNECTION] Checking proxy configuration..") _ensure_proxy_configured(profile, connection_observer) - logger.info(f"[CONNECTION] Checking Wireguard configuration..") + logger.info("[CONNECTION] Checking Wireguard configuration..") if profile.connection.code == "wireguard": _ensure_wireguard_configured(profile, connection_observer) @@ -71,81 +69,6 @@ def establish_connection( return _establish_with_renegotiation(profile, establish_fn, ignore, connection_observer, ticket_observer) - -def launch_encrypted_proxy( - profile: Union[SessionProfile, SystemProfile], - connection_observer: Optional[ConnectionObserver] -): - """ - Launch & Setup an Encrypted proxy. - - 1) Check if the config exists - 2) If not, set it up - 3) If it does, get the server IP from it - 4) Launch singbox with the config & IP - """ - # do they have a proxy config already? - config_path = f'{profile.get_system_config_path()}/proxy.json' - - tries = 0 - exists_already = os.path.isfile(config_path) - - while tries < QUANTITY_OF_TRIES: - tries = tries + 1 - - if exists_already: - logger.info("The config exists already, attempting to get the IP from it...") - server_ip = get_value_from_json_file(filepath=config_path, category="outbounds", key="server") - else: - server_ip = setup_config_and_return_ip( - profile=profile, - connection_observer=connection_observer - ) - - print(f"We will be aiming to establish a connection to {server_ip}") - - connection_result = start_singbox( - profile_id=profile.id, - server_ip=server_ip, - protocol_choice=profile.connection.code, - connection_observer=connection_observer - ) - - if connection_result.valid: - return connection_result - else: - logger.info("The existing config was invalid, trying a new config..") - exists_already = False - - -# removed = delete_file(config_path) -# if not removed: -# return Result(valid=False, error_type=ResultError.FILE_SYSTEM, message="Can't delete an invalid config.") - - -def setup_config_and_return_ip( - profile: Union[SessionProfile, SystemProfile], - connection_observer: Optional[ConnectionObserver], -) -> str | None: - - configured = configure_singbox( - profile=profile, - connection_observer=connection_observer - ) - - if configured.valid: - server_ip = configured.data - return server_ip - else: - reason = configured.error_type - logger.error(f"Could not setup Singbox config because {configured.error_type} and {configured.message}") - - if reason == ResultError.INVALID_API_REPLY: - raise InvalidSubscriptionError() - else: - raise ConnectionError('The connection could not be established.') - return None # redundant - def _ensure_proxy_configured( profile: Union[SessionProfile, SystemProfile], connection_observer: Optional[ConnectionObserver], @@ -156,11 +79,11 @@ def _ensure_proxy_configured( return if not profile.has_proxy_configuration(): - logger.info(f"[CONNECTION] You need a new proxy configuration") + logger.info("[CONNECTION] You need a new proxy configuration") if not profile.assassin: activate_subscription(profile, connection_observer) - logger.info(f"[CONNECTION] Getting a proxy config from web service..") + logger.info("[CONNECTION] Getting a proxy config from web service..") proxy_config = ConnectionController.with_preferred_connection( profile.subscription.billing_code, task=WebServiceApiService.get_proxy_configuration, @@ -186,7 +109,7 @@ def _ensure_wireguard_configured( if profile.type == ProfileType.SYSTEM: if system_uses_wireguard_interface() and SystemStateController.exists(): - logger.info(f"[CONNECTION] There is a systemwide profile that is already is active.") + logger.info("[CONNECTION] There is a systemwide profile that is already is active.") try: terminate_system_connection() except ConnectionTerminationError as e: @@ -194,9 +117,9 @@ def _ensure_wireguard_configured( if not profile.assassin: activate_subscription(profile, connection_observer) - logger.info(f"[CONNECTION] Checking if the profile already has a wg config..") + logger.info("[CONNECTION] Checking if the profile already has a wg config..") if not profile.has_wireguard_configuration(): - logger.info(f"[CONNECTION] Profile does NOT have WG keys and it needs it. Registering a NEW wg session..") + logger.info("[CONNECTION] Profile does NOT have WG keys and it needs it. Registering a NEW wg session..") registry_result = register_wireguard_session(profile, connection_observer=connection_observer) logger.info(f"registry_result is {registry_result}") @@ -212,12 +135,12 @@ def _establish_with_renegotiation( ): """Attempt connection, renegotiate WireGuard once if needed.""" try: - logger.info(f"[CONNECTION] Connecting..") + logger.info("[CONNECTION] Connecting..") return establish_fn(profile, ignore=ignore, connection_observer=connection_observer) except ConnectionError as e: logger.error(str(e)) if __should_renegotiate(profile): - logger.info(f"[CONNECTION] Renegotiating a new wg key session with API..") + logger.info("[CONNECTION] Renegotiating a new wg key session with API..") # KEY if connection_observer: @@ -230,7 +153,7 @@ def _establish_with_renegotiation( return establish_fn(profile, ignore=ignore, connection_observer=connection_observer, ticket_observer=ticket_observer) raise ConnectionError('The connection could not be established.') except FirewallError: - logger.error(f"Failed to turn on the firewall, passing the error up..") + logger.error("Failed to turn on the firewall, passing the error up..") raise diff --git a/core/services/networking/systemwide/encrypted_proxy/launch_proxy.py b/core/services/networking/systemwide/encrypted_proxy/launch_proxy.py new file mode 100644 index 0000000..736d51f --- /dev/null +++ b/core/services/networking/systemwide/encrypted_proxy/launch_proxy.py @@ -0,0 +1,84 @@ +from core.services.networking.systemwide.encrypted_proxy.configure_singbox import configure_singbox +from core.services.networking.systemwide.encrypted_proxy.singbox_runner import start_singbox +from core.utils.basic_operations.write_or_read_from_json import get_value_from_json_file +from core.observers.ConnectionObserver import ConnectionObserver +from core.models.BaseProfile import BaseProfile as Profile +from core.models.Result import Result, ResultError +from core.errors.logger import logger + +from typing import Optional +import os + +QUANTITY_OF_TRIES = 2 + +def get_existing_server_ip(profile: Profile) -> Optional[str]: + """Attempts to retrieve the server IP from the local config file.""" + config_path = f'{profile.get_system_config_path()}/proxy.json' + if os.path.isfile(config_path): + logger.info("Attempting to get the IP from existing config...") + return get_value_from_json_file(filepath=config_path, category="outbounds", key="server") + return None + +def refresh_server_config(profile: Profile, connection_observer: ConnectionObserver) -> Result: + """Fetches a new configuration from the server and returns the IP.""" + configured = configure_singbox(profile=profile, connection_observer=connection_observer) + if not configured.valid: + logger.error(configured.message) + return configured + + if configured.data: + return Result(valid=True, data=configured.data) + + return Result( + valid=False, + error_type=ResultError.MISSING_DATA, + message="Failed to setup a new singbox configuration and extract its IP" + ) + +def launch_encrypted_proxy( + profile: Profile, + connection_observer: Optional[ConnectionObserver] +) -> Result: + """Launch & Setup an Encrypted proxy with a single-refresh retry policy.""" + + server_ip = get_existing_server_ip(profile) + has_refreshed = False # Tracks if we have already fetched a new config this session + + tries = 0 + while tries < QUANTITY_OF_TRIES: + tries += 1 + + # 1. Handle missing IP (initial state if no config existed on disk) + if not server_ip: + logger.info("No existing config found. Fetching initial config...") + refresh_result = refresh_server_config(profile, connection_observer) + if not refresh_result.valid: + return refresh_result + server_ip = refresh_result.data + has_refreshed = True + + logger.info(f"Attempt {tries}: establishing connection to server {server_ip}") + + connection_result = start_singbox( + profile_id=profile.id, + server_ip=server_ip, + protocol_choice=profile.connection.code, + connection_observer=connection_observer + ) + + if connection_result.valid: + return connection_result + + # 2. Handle connection failure + if not has_refreshed: + logger.info("Existing config failed. Fetching a new config before retrying...") + refresh_result = refresh_server_config(profile, connection_observer) + if not refresh_result.valid: + return refresh_result + + server_ip = refresh_result.data + has_refreshed = True + else: + logger.info("Connection failed with new config. Retrying connection without re-fetching...") + + return Result(valid=False, message=f"Failed to connect after {QUANTITY_OF_TRIES} attempts") diff --git a/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py b/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py index 6fb7a42..559a8d6 100644 --- a/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py +++ b/core/services/networking/systemwide/encrypted_proxy/singbox_runner.py @@ -34,6 +34,39 @@ from typing import Optional QUANTITY_OF_ATTEMPTS = 2 +def verify_connection_integrity(connection_observer: ConnectionObserver, server_ip: str) -> Result: + """Checks if the connection is active and not leaking.""" + ip_result = ip.test(connection_observer) + + if not ip_result.valid: + return Result( + valid=False, + error_type=ResultError.CONNECTION, + message="IP address check is unable to connect." + ) + + observed_ip = ip_result.data + if str(observed_ip) != str(server_ip): + return Result( + valid=False, + error_type=ResultError.LEAK_ISSUE, + message=f"Connection is leaking! Observed {observed_ip} does NOT match intended {server_ip}." + ) + + return Result(valid=True, data=observed_ip) + + +def teardown_failed_connection(current_state: SystemState) -> None: + """Performs a full cleanup: closes the interface and dissolves system state.""" + logger.info("Performing full teardown of failed connection...") + + closed = orchestrate_closing(current_state=current_state, interface=Constants.SINGBOX_TUN_IF) + + if closed.valid: + logger.info("Closed singbox. Performing redundant state removal...") + SystemStateController.dissolve() + + def confirm_proxy_started(protocol_choice: str) -> bool: SINGBOX_OUTPUT = Constants.SINGBOX_OUTPUT @@ -190,168 +223,138 @@ def end_singbox() -> Result: return Result(valid=True) +def prepare_singbox_environment(protocol_choice: str) -> Result: + """Ends existing processes and clears logs.""" + killed = end_singbox() + if not killed.valid: + return killed + + wiped = write_string_to_text_file(content_to_write="", file_path=Constants.SINGBOX_OUTPUT) + if not wiped: + error_msg = f"Critical Error wiping log at {Constants.SINGBOX_OUTPUT}." + if protocol_choice == "vless": + logger.error(f"{error_msg} Vless relies on this to detect connection start.") + return Result(valid=False, error_type=ResultError.FILE_SYSTEM, message=error_msg) + logger.error(f"{error_msg} but {protocol_choice} does not require it.") + + return Result(valid=True) + +def ensure_interface_is_up(process_id: int, connection_observer: Optional[ConnectionObserver]) -> Result: + """Verifies the TUN interface is actually active.""" + if connection_observer: + connection_observer.notify("connecting", "Testing Interface..") + + interface_result = interface_tools.get_output(Constants.SINGBOX_TUN_IF) + if not interface_result.valid: + return Result(valid=False, error_type=ResultError.INTERFACE, + message=f"Could not get interface output for PID {process_id}", data=process_id) + + time.sleep(1) + if not interface_tools.is_up(interface_result.data): + msg = "Interface is DOWN" + if connection_observer: connection_observer.notify("connecting", msg) + return Result(valid=False, error_type=ResultError.INTERFACE, data=process_id) + + return Result(valid=True) + +def apply_network_settings(current_state, server_ip, connection_observer) -> Result: + """Handles Firewall and DNS configuration.""" + # 1. Firewall + firewall_result = enable_firewall_w_retry( + interface_name=str(Constants.SINGBOX_TUN_IF), + server_ip=str(server_ip), + internal_subnet=Constants.SINGBOX_INTERNAL_SUBNET, + max_retries=2 + ) + if not firewall_result.valid: + return Result(valid=False, error_type=ResultError.FIREWALL, message=firewall_result.message) + + if connection_observer: + connection_observer.notify("connecting", "Firewall Enabled.") + + # 2. DNS + dns_result = set_dns_for_singbox(current_state) + if not dns_result: # Assuming set_dns returns None/False on failure + return Result(valid=False, error_type=ResultError.DNS, message="Failed to set DNS") + + if connection_observer: + connection_observer.notify("connecting", "DNS Changed.") + + return Result(valid=True) + + def start_singbox( profile_id: int, server_ip: str, protocol_choice: str, connection_observer: Optional[ConnectionObserver] = None ) -> Result: - """ - Purpose: - Orchestrates the full flow of the Singbox connection. + """Orchestrates the full flow of the Singbox connection.""" + + # 1. Validation & Prep + if not all([profile_id, server_ip]): + return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="Invalid inputs") - Steps: - 1) Validate inputs - 2) Start the binary - 3) Evaluate if the process id is still up. - 4) Check if the interface is up (before we wait for timeout period for output). This gives immediate feedback. - 5) For a timeout period, we wait to confirm via singbox's output in a text file. - 6) If it's up, setup the State JSON - 7) Test the IP address to match the server via an external API - 8) Turn on Firewall - 9) Turn on DNS - 10) Update the State - """ - # ============= INPUT VALIDATION ============= - requirements = [profile_id, server_ip] - for each_requirement in requirements: - if each_requirement is None: - return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="Invalid inputs into singbox runner") - logger.info("All requirements met.") + prep_result = prepare_singbox_environment(protocol_choice) + if not prep_result.valid: + return prep_result - # ========== KILL IT IF ALREADY UP ========== - killed_pre_existing = end_singbox() - if not killed_pre_existing.valid: - return killed_pre_existing - logger.info("Pre-existing process ended.") - - # ========= WIPE PRE-EXISTING LOGS ======= - wiped = write_string_to_text_file(content_to_write="", file_path=Constants.SINGBOX_OUTPUT) - if not wiped: - error_msg = f"Critical Error with wiping the log at {Constants.SINGBOX_OUTPUT}." - if protocol_choice == "vless": - logger.error(f"{error_msg} which vless relies upon to see when the connection started.") - return Result(valid=False, error_type=ResultError.FILE_SYSTEM, message=error_msg) - logger.error(f"{error_msg} but that's okay, because {protocol_choice} doesn't use it.") - - # ============= START BINARY ============= + # 2. Launch Binary if connection_observer: connection_observer.notify("connecting", "Starting Software..") + launched = _attempt_start_with_retry(profile_id=profile_id, quantity_of_attempts=2) if not launched.valid: return launched - + process_id = launched.data - logger.info(f"Binary started with process id {process_id}.") - # ============= CHECK INTERFACE ============= - if connection_observer: - connection_observer.notify("connecting", "Testing Interface..") - print(f"The Constants.SINGBOX_TUN_IF is {Constants.SINGBOX_TUN_IF}") - interface_result = interface_tools.get_output(Constants.SINGBOX_TUN_IF) - if not interface_result.valid: - error_msg = f"We could not get the interface's output, despite having a process id of {process_id}" - logger.error(error_msg) - return Result(valid=False, error_type=ResultError.INTERFACE, message=error_msg, data=process_id) + # 3. Health Checks (Pre-State Cleanup) + # Note: Since SystemStateController.create hasn't been called yet, we use terminate_pid() for cleanup here. + health_check = ensure_interface_is_up(process_id, connection_observer) + if not health_check.valid: + terminate_pid(process_id) # <--- Lightweight cleanup + return health_check - time.sleep(1) - singbox_output = interface_result.data - interface_up = interface_tools.is_up(singbox_output) - logger.info(f"Is the interface up? {interface_up}") - - if interface_up: - logger.info("Interface is confirmed to be Up") - if connection_observer: - connection_observer.notify("connecting", "Interface looking healthy..") - else: - error_msg = "Interface is DOWN" - logger.error(error_msg) - if connection_observer: - connection_observer.notify("connecting", error_msg) - return Result(valid=False, error_type=ResultError.INTERFACE, data=process_id) - - # ======= CONFIRM VIA SINGBOX OUTPUT ========== - its_up = confirm_proxy_started(protocol_choice=protocol_choice) - if not its_up: - return Result(valid=False, error_type=ResultError.CONNECTION, message="After a timeout period, the singbox output is still not showing the connection having started.") - logger.info("Singbox output confirming it's up.") + if not confirm_proxy_started(protocol_choice=protocol_choice): + terminate_pid(process_id) # <--- Lightweight cleanup + return Result(valid=False, error_type=ResultError.CONNECTION, message="Singbox output timeout") + if connection_observer: connection_observer.notify("connecting", "Software looking healthy so far..") - # ============= SETUP STATE ============= - # 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") + # 4. State Persistence current_state = SystemStateController.create( profile_id=profile_id, - firewalled=True, # intended setting, not result yet - dns_set=True, # intended setting, not result yet + firewalled=True, + dns_set=True, protocol=get_protocol_enum(profile_id), process_id=process_id ) - logger.info("State is created.") - # ============= 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) - if closed.valid: - logger.info("Closed singbox. Redundant state removal now taking place..") - # redundant state removal. This is a repeat of the close_Singbox doing it. - SystemState.dissolve() + # 5. Network Configuration (Post-State Cleanup) + # From this point forward, we use the full teardown_failed_connection helper + + net_result = apply_network_settings(current_state, server_ip, connection_observer) + if not net_result.valid: + logger.error(f"Network Config Failed: {net_result.message}") + teardown_failed_connection(current_state) # <--- Full teardown + return net_result - # 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.") + # 6. Final IP Integrity Check + validation = verify_connection_integrity(connection_observer, server_ip) + if not validation.valid: + logger.error(validation.message) + teardown_failed_connection(current_state) # <--- Full teardown + if validation.error_type == ResultError.LEAK_ISSUE and connection_observer: + connection_observer.notify("connecting", "Failed Connection! Test showed leak!") + return validation - observed_ip_address = ip_result.data - logger.info(f""" - observed_ip_address: {observed_ip_address} - observed_ip_address type: {type(observed_ip_address)} - server_ip: {server_ip} - server_ip type: {type(server_ip)} - """) - if str(observed_ip_address) != str(server_ip): - error_msg = f"Connection didn't work, it's leaking! 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) - error_for_user = f"Failed Connection! Test showed leak! Closed Status: {closed}" - if connection_observer: - connection_observer.notify("connecting", error_for_user) - return Result(valid=False, error_type=ResultError.LEAK_ISSUE, message=error_msg) - - - logger.info("Observed IP matches the intended server IP. Proceeding..") + # 7. Success if connection_observer: - connection_observer.notify("connecting", f"IP Matched! {server_ip}") - - # ============= FIREWALL ============= - logger.info(f"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 - interface_name=str(Constants.SINGBOX_TUN_IF), - server_ip=str(server_ip), - internal_subnet = (Constants.SINGBOX_INTERNAL_SUBNET), - max_retries = 2 - ) - - if not firewall_result.valid: - # Update the state of the failure, - current_state.firewalled = False - current_state.dns_set = False - SystemStateController.update_or_create(current_state) - raise FirewallError(firewall_result) # Now end the party. - if connection_observer: - connection_observer.notify("connecting", "Firewall Enabled.") - - # ============= DNS ============= - dns_result = set_dns_for_singbox(current_state) - if dns_result and connection_observer: - connection_observer.notify("connecting", "DNS Changed.") - # raises error if not okay. - + connection_observer.notify("connected", "Connected.") + Notifications.emit(EventType.SYSTEM_ENABLE, profile_id) - notification = "Connected." - Notifications.emit(EventType.STATUS_UPDATE, notification) - # ============= CONCLUSION ============= + Notifications.emit(EventType.STATUS_UPDATE, "Connected.") + return Result(valid=True, data=process_id) -