forked from Support/sp-hydra-veil-gui
359 lines
13 KiB
Python
359 lines
13 KiB
Python
# core
|
|
from core.controllers.ConfigurationController import ConfigurationController
|
|
from core.services.prepare_tickets.ticket_tracker import get_a_random_ticket
|
|
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, wipe_subscription, wipe_billing_cycle
|
|
from core.utils.check_if_time_lapsed import check_if_time_lapsed
|
|
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.actions.ticket_actions import get_ticket_data, use_a_ticket
|
|
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.")
|
|
|
|
|
|
@register_action(ResultError.EXPIRED_SUB)
|
|
def wipe_expired_subscription(profile_id: int) -> Result:
|
|
"""Call core to wipe a subscription, then navigate to the billing page"""
|
|
wiped = wipe_subscription(profile_id)
|
|
if wiped.valid:
|
|
return navigate_to_new_sub_page()
|
|
else:
|
|
return wiped
|
|
|
|
|
|
@register_action(ResultError.NO_TICKET)
|
|
def start_and_navigate_to_ticketing():
|
|
"""Starts a new ticketing cycle. (assumes the user opted in)."""
|
|
start_result = start_new_payment_cycle(category=PaymentCategory.TICKET)
|
|
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="plan_picker")
|
|
else:
|
|
logger.info(f"Returning the error of start_result {start_result}")
|
|
return start_result
|
|
|
|
|
|
def random_ticket_dispatch(profile_id: int, passed_ui_caller: Callable) -> Result:
|
|
"""
|
|
User explicitly wants a random ticket.
|
|
The reason this is in GUI repo, is because use_a_ticket is.
|
|
"""
|
|
|
|
random_ticket_result = get_a_random_ticket()
|
|
|
|
if not random_ticket_result.valid:
|
|
return random_ticket_result
|
|
|
|
which_ticket = random_ticket_result.data
|
|
|
|
# this is a redundant sanity check:
|
|
if not which_ticket:
|
|
Result(valid=False, error_type=ResultError.NO_TICKET, message="No tickets left. Would you like to get some?")
|
|
|
|
# calls GUI repo function:
|
|
return use_a_ticket(profile_id=profile_id, which_ticket=which_ticket, ui_caller=passed_ui_caller)
|
|
|
|
|
|
def dispatch_ticketing(profile_id: int, passed_ui_caller: Callable) -> Result:
|
|
"""Evaluate if a random ticket will be used."""
|
|
|
|
use_random_setting = ConfigurationController.use_random
|
|
|
|
if use_random_setting:
|
|
return random_ticket_dispatch(profile_id=profile_id, passed_ui_caller=passed_ui_caller)
|
|
|
|
# User does NOT want an automated selection, so they can pick from their tickets:
|
|
do_they_have_tickets = get_ticket_data()
|
|
|
|
if do_they_have_tickets.valid:
|
|
return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="ticket_or_billing_choice")
|
|
|
|
else:
|
|
# user can pick between tickets or individual:
|
|
return navigate_to_new_sub_page()
|
|
|
|
|
|
def filter_subscription(profile_id: int, passed_ui_caller: Callable) -> 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)
|
|
|
|
#-----------------------------
|
|
# No Subscription
|
|
#-----------------------------
|
|
if not subscription or not subscription.billing_code:
|
|
tickets_setup = ConfigurationController.are_tickets_setup()
|
|
if tickets_setup:
|
|
# already opted into ticketing:
|
|
return dispatch_ticketing(profile_id=profile_id, passed_ui_caller=passed_ui_caller)
|
|
else:
|
|
# Starting fresh. Pick a billing type:
|
|
return navigate_to_new_sub_page()
|
|
|
|
#-----------------------------
|
|
# Pending Subscription
|
|
#-----------------------------
|
|
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="Please finish your payment.")
|
|
else:
|
|
logger.info(f"Returning the error of start_result {start_result}")
|
|
return start_result
|
|
|
|
#-----------------------------
|
|
# Expired Subscription
|
|
#-----------------------------
|
|
expiration_date = subscription.expires_at
|
|
if not expiration_date:
|
|
# auto check tickets would be here.
|
|
return navigate_to_new_sub_page()
|
|
|
|
expired = check_if_time_lapsed(expiration_date)
|
|
if expired:
|
|
return Result(valid=False, error_type=ResultError.EXPIRED_SUB, message="This subscription is expired, would you like to wipe it and get a new one?")
|
|
|
|
#-----------------------------
|
|
# Proceed with 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, passed_ui_caller: Callable) -> 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, passed_ui_caller=passed_ui_caller)
|
|
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)
|
|
|
|
|
|
# def basic_test():
|
|
# print("returning the basic test")
|
|
# return Result(valid=False, error_type=ResultError.ACTIVE_BILLING, message="hello")
|
|
|
|
# @type_checker
|
|
# def connect_profile(profile_id: int, ui_caller: Callable) -> Result:
|
|
# """Type check before we thread"""
|
|
# take_action(
|
|
# action=basic_test,
|
|
# ui_caller=ui_caller
|
|
# )
|
|
# return Result(valid=True)
|
|
|
|
|
|
@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,
|
|
passed_ui_caller=ui_caller
|
|
)
|
|
return Result(valid=True)
|
|
|
|
|