diff --git a/apps/cic-ussd/cic_ussd/processor/menu.py b/apps/cic-ussd/cic_ussd/processor/menu.py index 24006a6c..15464e1e 100644 --- a/apps/cic-ussd/cic_ussd/processor/menu.py +++ b/apps/cic-ussd/cic_ussd/processor/menu.py @@ -104,6 +104,27 @@ class MenuProcessor: self.display_key, preferred_language, last_transaction_set=last_transaction_set ) + def add_guardian_pin_authorization(self): + guardian_information = self.guardian_metadata() + return self.pin_authorization(guardian_information=guardian_information) + + def guardian_list(self): + preferred_language = get_cached_preferred_language(self.account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + set_guardians = self.account.get_guardians() + if set_guardians: + guardians_list = '' + guardians_list_header = translation_for('helpers.guardians_list_header', preferred_language) + for phone_number in set_guardians: + guardian = Account.get_by_phone_number(phone_number, self.session) + guardian_information = guardian.standard_metadata_id() + guardians_list += f'{guardian_information}\n' + guardians_list = guardians_list_header + '\n' + guardians_list + else: + guardians_list = translation_for('helpers.no_guardians_list', preferred_language) + return translation_for(self.display_key, preferred_language, guardians_list=guardians_list) + def help(self): preferred_language = get_cached_preferred_language(self.account.blockchain_address) if not preferred_language: @@ -155,6 +176,20 @@ class MenuProcessor: f'{self.display_key}.retry', preferred_language, retry_pin_entry=retry_pin_entry ) + def guarded_account_metadata(self): + guarded_account_phone_number = self.ussd_session.get('data').get('guarded_account_phone_number') + guarded_account = Account.get_by_phone_number(guarded_account_phone_number, self.session) + return guarded_account.standard_metadata_id() + + def guardian_metadata(self): + guardian_phone_number = self.ussd_session.get('data').get('guardian_phone_number') + guardian = Account.get_by_phone_number(guardian_phone_number, self.session) + return guardian.standard_metadata_id() + + def reset_guarded_pin_authorization(self): + guarded_account_information = self.guarded_account_metadata() + return self.pin_authorization(guarded_account_information=guarded_account_information) + def start_menu(self): """ :return: @@ -205,6 +240,47 @@ class MenuProcessor: sender_information=tx_sender_information ) + def exit_guardian_addition_success(self) -> str: + guardian_information = self.guardian_metadata() + preferred_language = get_cached_preferred_language(self.account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return translation_for(self.display_key, + preferred_language, + guardian_information=guardian_information) + + def exit_guardian_removal_success(self): + guardian_information = self.guardian_metadata() + preferred_language = get_cached_preferred_language(self.account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return translation_for(self.display_key, + preferred_language, + guardian_information=guardian_information) + + def exit_invalid_guardian_addition(self): + failure_reason = self.ussd_session.get('data').get('failure_reason') + preferred_language = get_cached_preferred_language(self.account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return translation_for(self.display_key, preferred_language, error_exit=failure_reason) + + def exit_invalid_guardian_removal(self): + failure_reason = self.ussd_session.get('data').get('failure_reason') + preferred_language = get_cached_preferred_language(self.account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return translation_for(self.display_key, preferred_language, error_exit=failure_reason) + + def exit_pin_reset_initiated_success(self): + guarded_account_information = self.guarded_account_metadata() + preferred_language = get_cached_preferred_language(self.account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return translation_for(self.display_key, + preferred_language, + guarded_account_information=guarded_account_information) + def exit_insufficient_balance(self): """ :return: @@ -297,12 +373,36 @@ def response(account: Account, display_key: str, menu_name: str, session: Sessio if menu_name == 'exit_insufficient_balance': return menu_processor.exit_insufficient_balance() + if menu_name == 'exit_invalid_guardian_addition': + return menu_processor.exit_invalid_guardian_addition() + + if menu_name == 'exit_invalid_guardian_removal': + return menu_processor.exit_invalid_guardian_removal() + if menu_name == 'exit_successful_transaction': return menu_processor.exit_successful_transaction() + if menu_name == 'exit_guardian_addition_success': + return menu_processor.exit_guardian_addition_success() + + if menu_name == 'exit_guardian_removal_success': + return menu_processor.exit_guardian_removal_success() + + if menu_name == 'exit_pin_reset_initiated_success': + return menu_processor.exit_pin_reset_initiated_success() + if menu_name == 'account_balances': return menu_processor.account_balances() + if menu_name == 'guardian_list': + return menu_processor.guardian_list() + + if menu_name == 'add_guardian_pin_authorization': + return menu_processor.add_guardian_pin_authorization() + + if menu_name == 'reset_guarded_pin_authorization': + return menu_processor.reset_guarded_pin_authorization() + if 'pin_authorization' in menu_name: return menu_processor.pin_authorization() diff --git a/apps/cic-ussd/cic_ussd/state_machine/logic/pin_guard.py b/apps/cic-ussd/cic_ussd/state_machine/logic/pin_guard.py new file mode 100644 index 00000000..4e87e473 --- /dev/null +++ b/apps/cic-ussd/cic_ussd/state_machine/logic/pin_guard.py @@ -0,0 +1,200 @@ +# standard imports +import logging +from typing import Tuple + +# external imports +import celery +import i18n +from phonenumbers.phonenumberutil import NumberParseException +from sqlalchemy.orm.session import Session + +# local imports +from cic_ussd.account.metadata import get_cached_preferred_language +from cic_ussd.db.models.account import Account +from cic_ussd.db.models.base import SessionBase +from cic_ussd.phone_number import process_phone_number, E164Format +from cic_ussd.session.ussd_session import save_session_data +from cic_ussd.translation import translation_for + + +logg = logging.getLogger(__file__) + + +def save_guardian_to_session_data(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + session_data = ussd_session.get('data') or {} + guardian_phone_number = process_phone_number(phone_number=user_input, region=E164Format.region) + session_data['guardian_phone_number'] = guardian_phone_number + save_session_data('cic-ussd', session, session_data, ussd_session) + + +def save_guarded_account_session_data(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + session_data = ussd_session.get('data') or {} + guarded_account_phone_number = process_phone_number(phone_number=user_input, region=E164Format.region) + session_data['guarded_account_phone_number'] = guarded_account_phone_number + save_session_data('cic-ussd', session, session_data, ussd_session) + + +def retrieve_person_metadata(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + guardian_phone_number = process_phone_number(user_input, E164Format.region) + guardian = Account.get_by_phone_number(guardian_phone_number, session) + blockchain_address = guardian.blockchain_address + s_query_person_metadata = celery.signature( + 'cic_ussd.tasks.metadata.query_person_metadata', [blockchain_address], queue='cic-ussd') + s_query_person_metadata.apply_async() + + +def is_valid_guardian_addition(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + try: + phone_number = process_phone_number(user_input, E164Format.region) + except NumberParseException: + phone_number = None + + preferred_language = get_cached_preferred_language(account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + + is_valid_account = Account.get_by_phone_number(phone_number, session) is not None + is_initiator = phone_number == account.phone_number + is_existent_guardian = phone_number in account.get_guardians() + + failure_reason = '' + if not is_valid_account: + failure_reason = translation_for('helpers.error.no_matching_account', preferred_language) + + if is_initiator: + failure_reason = translation_for('helpers.error.is_initiator', preferred_language) + + if is_existent_guardian: + failure_reason = translation_for('helpers.error.is_existent_guardian', preferred_language) + + if failure_reason: + session_data = ussd_session.get('data') or {} + session_data['failure_reason'] = failure_reason + save_session_data('cic-ussd', session, session_data, ussd_session) + + return phone_number is not None and is_valid_account and not is_existent_guardian and not is_initiator + + +def add_pin_guardian(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + guardian_phone_number = ussd_session.get('data').get('guardian_phone_number') + account.add_guardian(guardian_phone_number) + session.add(account) + session.flush() + SessionBase.release_session(session=session) + + +def is_set_pin_guardian(account: Account, checked_number: str, preferred_language: str, session: Session, ussd_session: dict): + """""" + failure_reason = '' + set_guardians = [] + if account: + set_guardians = account.get_guardians() + else: + failure_reason = translation_for('helpers.error.no_matching_account', preferred_language) + + is_set_guardian = checked_number in set_guardians + is_initiator = checked_number == account.phone_number + + if not is_set_guardian: + failure_reason = translation_for('helpers.error.is_not_existent_guardian', preferred_language) + + if is_initiator: + failure_reason = translation_for('helpers.error.is_initiator', preferred_language) + + if failure_reason: + session_data = ussd_session.get('data') or {} + session_data['failure_reason'] = failure_reason + save_session_data('cic-ussd', session, session_data, ussd_session) + + return is_set_guardian and not is_initiator + + +def is_dialers_pin_guardian(state_machine_data: Tuple[str, dict, Account, Session]): + user_input, ussd_session, account, session = state_machine_data + phone_number = process_phone_number(phone_number=user_input, region=E164Format.region) + preferred_language = get_cached_preferred_language(account.blockchain_address) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return is_set_pin_guardian(account, phone_number, preferred_language, session, ussd_session) + + +def is_others_pin_guardian(state_machine_data: Tuple[str, dict, Account, Session]): + user_input, ussd_session, account, session = state_machine_data + preferred_language = get_cached_preferred_language(account.blockchain_address) + phone_number = process_phone_number(phone_number=user_input, region=E164Format.region) + guarded_account = Account.get_by_phone_number(phone_number, session) + if not preferred_language: + preferred_language = i18n.config.get('fallback') + return is_set_pin_guardian(guarded_account, account.phone_number, preferred_language, session, ussd_session) + + +def remove_pin_guardian(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + guardian_phone_number = ussd_session.get('data').get('guardian_phone_number') + account.remove_guardian(guardian_phone_number) + session.add(account) + session.flush() + SessionBase.release_session(session=session) + + +def initiate_pin_reset(state_machine_data: Tuple[str, dict, Account, Session]): + """ + :param state_machine_data: + :type state_machine_data: + :return: + :rtype: + """ + user_input, ussd_session, account, session = state_machine_data + session_data = ussd_session.get('data') + quorum_count = session_data['quorum_count'] if session_data.get('quorum_count') else 0 + quorum_count += 1 + session_data['quorum_count'] = quorum_count + save_session_data('cic-ussd', session, session_data, ussd_session) + guarded_account_phone_number = session_data.get('guarded_account_phone_number') + guarded_account = Account.get_by_phone_number(guarded_account_phone_number, session) + if quorum_count >= guarded_account.guardian_quora: + guarded_account.reset_pin(session) + logg.debug(f'Reset initiated for: {guarded_account.phone_number}') + session_data['quorum_count'] = 0 + save_session_data('cic-ussd', session, session_data, ussd_session)