SimplifiedPrivacy/hydraveil-gui
1
0
Fork 1
forked from Support/sp-hydra-veil-gui
Code Pull requests Activity
hydraveil-gui/gui/v2/actions/menu_coordinator.py

225 lines
8.5 KiB
Python
Raw Blame History

# 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, evaluate_repeat_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.translate_error import translate_error, get_ignored_exceptions
from core.services.payment_phase.payment_state import OverallPaymentCycle
# gui
# from gui.v2.actions.payment_actions import wipe_expired_subscription
from gui.v2.actions.ticket_actions import dispatch_ticketing
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,
)
from gui.v2.actions.error_handlers import navigate_to_new_sub_page
# generic
from typing import Callable, Optional
from functools import partial
@type_checker
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 dispatch_individual(profile_id: int, ui_caller: Callable) -> Result:
payment_cycle = OverallPaymentCycle()
print(f"payment_cycle is {payment_cycle}")
if payment_cycle.active:
same_result = evaluate_repeat_cycle(payment_cycle=payment_cycle, profile_id=profile_id)
if not same_result.valid:
return same_result
else:
error_msg = "There's an error in the dispatch_individual function"
logger.error(f"This shouldn't be going off. {error_msg}")
return Result(valid=False, error_type=ResultError.NAVIGATE_TO, navigate_to="home", message=error_msg)
else:
# Starting fresh. Pick a billing type:
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:
return dispatch_individual(profile_id=profile_id, ui_caller=passed_ui_caller)
#-----------------------------
# 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:
function_choice = partial(wipe_subscription, profile_id=profile_id)
return Result(valid=False, function_choice=function_choice, 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:
ConfigurationController.set_connection_enum(ConnectionChoice.TOR)
return Result(valid=True)
@type_checker
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 translate_error(error=error, profile_id=profile.id)
@register_action(ResultError.STATE_CONFLICT)
def ignore_state_conflict(profile_id: int) -> Result:
return literally_connect(
profile_id=profile_id,
ignore_state_conflict=True
)
def _connect_profile(profile_id: int, passed_ui_caller: Callable) -> Result:
"""Background Threaded Orchestration to Start a Connection"""
# Step 1: 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 2: Tor filter check:
result = NOT_need_tor_warnings(profile_id=profile_id)
if not result.valid:
return result
# Step 3: 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,
passed_ui_caller=ui_caller
)
return Result(valid=True)
View git blame Copy permalink