Adds logic for social pin recovery.
This commit is contained in:
parent
860a27f8d5
commit
97a0e42501
GPG Key ID:
B00CE9034DA19FB7
|
|
@ -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()
|
||||
|
||||
|
|
|
|||
|
|
@ -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)
|
||||
Loading…
Reference in New Issue