# core from core.services.profiles.profile_connections import enable_profile #, disable_profile from core.controllers.ProfileController import ProfileController from core.models.Configuration import Configuration, ConnectionChoice from core.services.subscriptions.subscriptions_by_protocol import get_profile_subscription, start_new_payment_cycle from core.models.enum_models import PaymentCategory from core.utils.type_checker import type_checker from core.models.Result import Result, ResultError from core.errors.logger import logger from core.Errors import ( CommandNotFoundError, EndpointVerificationError, FileIntegrityError, InvalidSubscriptionError, MissingSubscriptionError, ProfileActivationError, ProfileModificationError, ProfileStateConflictError, UnsupportedApplicationVersionError, ) # gui from gui.v2.infrastructure.action_registry import register_action from gui.v2.infrastructure.take_plumbing_action import take_action from gui.v2.infrastructure.screen_size import get_max_screensize from gui.v2.infrastructure.setup_observers import ( application_version_observer, connection_observer, profile_observer, ticket_observer, ) # generic from typing import Callable, Optional, Type ERROR_TO_UI_MESSAGE = { ProfileActivationError: "The profile could not be enabled", UnsupportedApplicationVersionError: "The application version in question is not supported", FileIntegrityError: "Application version file integrity could not be verified.", ProfileModificationError: "WireGuard configuration could not be attached.", } def get_profile(profile_id: int): profile = ProfileController.get(int(profile_id)) if not profile: return Result(valid=False, error_type=ResultError.INVALID_INPUT, message=f"There is no profile with the id {profile_id}") else: return Result(valid=True, data=profile) def get_ignored_exceptions(ignore_state_conflict: Optional[bool] = False): ignored_as_list = [] config = Configuration.get() endpoint_verification_enabled = config.endpoint_verification_enabled if endpoint_verification_enabled: ignored_as_list.append(EndpointVerificationError) if ignore_state_conflict: ignored_as_list.append(ProfileStateConflictError) ignored = tuple(ignored_as_list) return ignored @register_action(ResultError.SUBSCRIPTION) def navigate_to_new_sub_page() -> Result: return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="id", message="No subscription. Pick a payment method.") def filter_subscription(profile_id: int) -> Result: """Evaluate if this profile has a subscription that's ready to connect.""" logger.info(f"Checking the subscription on profile id {profile_id}") subscription = get_profile_subscription(profile_id=profile_id) if not subscription or not subscription.billing_code: # auto check tickets would be here. return navigate_to_new_sub_page() if subscription.billing_pending: start_result = start_new_payment_cycle(category=PaymentCategory.INDIVIDUAL, profile_id=profile_id) if start_result.valid: logger.info("creating a new payment cycle was valid, it worked.") return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="payment_details", message="Finish the subscription payment") else: logger.info(f"Returning the error of start_result {start_result}") return start_result # has a valid sub: logger.info(f"Profile id {profile_id} has a valid subscription ready to go..") return Result(valid=True) @register_action(ResultError.TOR_NEEDED) def switch_to_tor_connection() -> Result: config = Configuration.get() config.connection = ConnectionChoice.TOR config.save() return Result(valid=True) def NOT_need_tor_warnings(profile_id: int) -> Result: """ Purpose: Filters if we need to warn the user about a potential Tor leak. Context: If it's a Tor profile, and the user does NOT have Tor connection enabled, do we need to warn them? Returns: If we need it, it emits Result as False, with ResultError.TOR_NEEDED If not, returns Result as True, to move on. """ # PROFILE profile_result = get_profile(profile_id) profile = profile_result.data # Actual Profile is a Tor Profile: if profile.connection.code != "tor": return Result(valid=True) # no warning needed, it's not even Tor # Config shows a non-Tor connection to download apps: config = Configuration.get() if config.connection != ConnectionChoice.TOR: return Result(valid=True) # no warning needed, it's already using Tor to download apps # Lacks Application: if profile.application_version: return Result(valid=True) # no warning needed, they have the app already. error_msg = "This app is not yet installed. Since it's a Tor profile, do you want to switch to Tor to download it? (Hit 'No' to proceed over clearweb)" return Result(valid=False, error_type=ResultError.TOR_NEEDED, message=error_msg) @register_action(ResultError.ENDPOINT_VERIFICATION) def literally_connect( profile_id: int, ignore_state_conflict: Optional[bool] = False ) -> Result: # Step 1: Get the Profile profile_result = get_profile(profile_id) if not profile_result: return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="This profile does not match one of your configuration files.") profile = profile_result.data # EXCEPTIONS ignored = get_ignored_exceptions(ignore_state_conflict=ignore_state_conflict) # SCREEN SIZE: max_resolution = get_max_screensize() try: final_result = enable_profile( profile=profile, ignore=ignored, profile_observer=profile_observer, application_version_observer=application_version_observer, connection_observer=connection_observer, ticket_observer=ticket_observer, max_resolution=max_resolution ) if final_result: return final_result else: return Result(valid=True) except Exception as error: return _enable_error(error=error, profile_id=profile_id) def _enable_error(error: Type[Exception], profile_id: int) -> Result: """Translate core errors into actions and/or messages via Result Objects.""" logger.error(str(error)) #---------------------- # PURE MESSAGES (NO SOLUTION FUNCTION) #---------------------- if error in ERROR_TO_UI_MESSAGE: error_msg = ERROR_TO_UI_MESSAGE[error] return Result(valid=False, error_type=ResultError.UNKNOWN, message=error_msg) #---------------------- # ENDPOINT VERIFICATION #---------------------- if isinstance(error, EndpointVerificationError): return Result(valid=False, error_type=ResultError.ENDPOINT_VERIFICATION, message="Operator verification failed. Do you want to continue anyway?") #---------------------- # STATE CONFLICT #---------------------- if isinstance(error, ProfileStateConflictError): function_choice = literally_connect( profile_id=profile_id, ignore_state_conflict=True ) error_msg = "The profile is already enabled. Do you want to force enable it anyway?" return Result(valid=False, function_choice=function_choice, error_type=ResultError.STATE_CONFLICT, message=error_msg) #---------------------- # SUBSCRIPTION #---------------------- if isinstance(error, (InvalidSubscriptionError, MissingSubscriptionError)): error_msg = f"Subscription missing or invalid for profile {profile_id}. Do you want to get a new sub?" return Result(valid=False, error_type=ResultError.SUBSCRIPTION, message=error_msg) #---------------------- # COMMAND NOT FOUND #---------------------- if isinstance(error, CommandNotFoundError): # add install later!! name_of_package = str(error.subject) error_msg = f"You're missing the {name_of_package} package. Please install it." 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.") def _connect_profile(profile_id: int) -> Result: """Background Threaded Orchestration to Start a Connection""" # # Step 1: Get the Profile # profile_result = get_profile(profile_id) # if not profile_result: # return Result(valid=False, error_type=ResultError.INVALID_INPUT, message="This profile does not match one of your configuration files.") # profile = profile_result.data # Step 2: Subscription filter: sub_result = filter_subscription(profile_id=profile_id) if not sub_result.valid: return sub_result # Step 3: Tor filter check: result = NOT_need_tor_warnings(profile_id=profile_id) if not result.valid: return result # Actual Connection: logger.info(f"Profile id {profile_id} is ready to connect..") return literally_connect(profile_id=profile_id) @type_checker def connect_profile(profile_id: int, ui_caller: Callable) -> Result: """Type check before we thread""" take_action( action=_connect_profile, ui_caller=ui_caller, profile_id=profile_id ) return Result(valid=True)