Merge remote-tracking branch 'origin/master' into lash/custom-meta

cic-eth: Add sanity checks for emergency shutdown / liveness tests

See merge request grassrootseconomics/cic-internal-integration!110

.gitignore

@@ -1,2 +1,10 @@
 service-configs/*
 !service-configs/.gitkeep
+**/node_modules/
+__pycache__
+*.pyc
+*.o
+gmon.out
+*.egg-info
+dist/
+build/

apps/cic-cache/cic_cache/runnable/daemons/filters/erc20.py

@@ -67,6 +67,7 @@ class ERC20TransferFilter(SyncFilter):
                 tx.status == Status.SUCCESS,
                 block.timestamp,
                 )
-        db_session.flush()
+        #db_session.flush()
+        db_session.commit()
 
         return True

apps/cic-cache/cic_cache/runnable/daemons/tracker.py

@@ -26,9 +26,10 @@ from chainlib.eth.block import (
 from hexathon import (
         strip_0x,
         )
-from chainsyncer.backend import SyncerBackend
+from chainsyncer.backend.sql import SQLBackend
 from chainsyncer.driver import (
         HeadSyncer,
+        HistorySyncer,
         )
 from chainsyncer.db.models.base import SessionBase
 
@@ -70,19 +71,21 @@ def main():
 
     syncers = []
 
-    #if SyncerBackend.first(chain_spec):
-    #    backend = SyncerBackend.initial(chain_spec, block_offset)
-    syncer_backends = SyncerBackend.resume(chain_spec, block_offset)
+    #if SQLBackend.first(chain_spec):
+    #    backend = SQLBackend.initial(chain_spec, block_offset)
+    syncer_backends = SQLBackend.resume(chain_spec, block_offset)
 
     if len(syncer_backends) == 0:
         logg.info('found no backends to resume')
-        syncer_backends.append(SyncerBackend.initial(chain_spec, block_offset))
+        syncer_backends.append(SQLBackend.initial(chain_spec, block_offset))
     else:
         for syncer_backend in syncer_backends:
             logg.info('resuming sync session {}'.format(syncer_backend))
 
-    syncer_backends.append(SyncerBackend.live(chain_spec, block_offset+1))
+    for syncer_backend in syncer_backends:
+        syncers.append(HistorySyncer(syncer_backend))
 
+    syncer_backend = SQLBackend.live(chain_spec, block_offset+1)
     syncers.append(HeadSyncer(syncer_backend))
 
     trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')

apps/cic-cache/docker/Dockerfile

@@ -17,7 +17,7 @@ RUN apt-get update && \
 
 # Copy shared requirements from top of mono-repo
 RUN echo "copying root req file ${root_requirement_file}"
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a58
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a76
 
 COPY cic-cache/requirements.txt ./
 COPY cic-cache/setup.cfg \
@@ -47,6 +47,9 @@ RUN git clone https://gitlab.com/grassrootseconomics/cic-contracts.git && \
 	mkdir -p /usr/local/share/cic/solidity && \
 	cp -R cic-contracts/abis /usr/local/share/cic/solidity/abi
 
+COPY cic-cache/docker/start_tracker.sh ./start_tracker.sh
+COPY cic-cache/docker/db.sh ./db.sh
+RUN chmod 755 ./*.sh
 # Tracker
 # ENTRYPOINT ["/usr/local/bin/cic-cache-tracker", "-vv"]
 # Server

apps/cic-cache/docker/db.sh

@@ -0,0 +1,6 @@
+#!/bin/bash
+
+set -e
+>&2 echo executing database migration
+python scripts/migrate.py -c /usr/local/etc/cic-cache --migrations-dir /usr/local/share/cic-cache/alembic -vv
+set +e

apps/cic-cache/docker/start_tracker.sh

@@ -0,0 +1,10 @@
+#!/bin/bash
+
+. ./db.sh
+
+if [ $? -ne "0" ]; then
+	>&2 echo db migrate fail
+	exit 1
+fi
+
+/usr/local/bin/cic-cache-trackerd $@

apps/cic-cache/requirements.txt

@@ -1,13 +1,12 @@
-cic-base~=0.1.2a58
+cic-base~=0.1.2a77
 alembic==1.4.2
 confini~=0.3.6rc3
 uwsgi==2.0.19.1
 moolb~=0.1.0
-cic-eth-registry~=0.5.4a10
+cic-eth-registry~=0.5.4a16
 SQLAlchemy==1.3.20
 semver==2.13.0
 psycopg2==2.8.6
 celery==4.4.7
 redis==3.5.3
-chainlib~=0.0.2a2
-chainsyncer~=0.0.1a21
+chainsyncer[sql]~=0.0.2a2

apps/cic-eth/cic_eth/admin/ctrl.py

@@ -32,7 +32,9 @@ def lock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.AL
     :returns: New lock state for address
     :rtype: number
     """
-    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    chain_str = '::'
+    if chain_spec_dict != None:
+        chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.set(chain_str, flags, address=address, tx_hash=tx_hash)
     logg.debug('Locked {} for {}, flag now {}'.format(flags, address, r))
     return chained_input
@@ -51,7 +53,9 @@ def unlock(chained_input, chain_spec_dict, address=ZERO_ADDRESS, flags=LockEnum.
     :returns: New lock state for address
     :rtype: number
     """
-    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    chain_str = '::'
+    if chain_spec_dict != None:
+        chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     r = Lock.reset(chain_str, flags, address=address)
     logg.debug('Unlocked {} for {}, flag now {}'.format(flags, address, r))
     return chained_input
@@ -127,7 +131,9 @@ def unlock_queue(chained_input, chain_spec_dict, address=ZERO_ADDRESS):
 
 @celery_app.task(base=CriticalSQLAlchemyTask)
 def check_lock(chained_input, chain_spec_dict, lock_flags, address=None):
-    chain_str = str(ChainSpec.from_dict(chain_spec_dict))
+    chain_str = '::'
+    if chain_spec_dict != None:
+        chain_str = str(ChainSpec.from_dict(chain_spec_dict))
     session = SessionBase.create_session()
     r = Lock.check(chain_str, lock_flags, address=ZERO_ADDRESS, session=session)
     if address != None:

apps/cic-eth/cic_eth/api/api_task.py

@@ -62,6 +62,18 @@ class Api:
                     )       
 
 
+    def default_token(self):
+        s_token = celery.signature(
+                'cic_eth.admin.token.default_token',
+                [],
+                queue=self.queue,
+                )
+        if self.callback_param != None:
+            s_token.link(self.callback_success)
+
+        return s_token.apply_async()
+
+
     def convert_transfer(self, from_address, to_address, target_return, minimum_return, from_token_symbol, to_token_symbol):
         """Executes a chain of celery tasks that performs conversion between two ERC20 tokens, and transfers to a specified receipient after convert has completed.
 

apps/cic-eth/cic_eth/check/db.py

@@ -0,0 +1,8 @@
+from cic_eth.db.models.base import SessionBase
+
+
+def health(*args, **kwargs):
+    session = SessionBase.create_session()
+    session.execute('SELECT count(*) from alembic_version')
+    session.close()
+    return True

apps/cic-eth/cic_eth/check/gas.py

@@ -0,0 +1,48 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.connection import RPCConnection
+from chainlib.chain import ChainSpec
+from chainlib.eth.gas import balance
+
+# local imports
+from cic_eth.db.models.role import AccountRole
+from cic_eth.db.models.base import SessionBase
+from cic_eth.db.enum import LockEnum
+from cic_eth.error import LockedError
+from cic_eth.admin.ctrl import check_lock
+
+logg = logging.getLogger().getChild(__name__)
+
+
+def health(*args, **kwargs):
+
+    session = SessionBase.create_session()
+
+    config = kwargs['config']
+    chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+    logg.debug('check gas balance of gas gifter for chain {}'.format(chain_spec))
+
+    try:
+        check_lock(None, None, LockEnum.INIT)
+    except LockedError:
+        logg.warning('INIT lock is set, skipping GAS GIFTER balance check.')
+        return True
+
+    gas_provider = AccountRole.get_address('GAS_GIFTER', session=session)
+    session.close()
+
+    rpc = RPCConnection.connect(chain_spec, 'default')
+    o = balance(gas_provider)
+    r = rpc.do(o)
+    try: 
+        r = int(r, 16)
+    except TypeError:
+        r = int(r)
+    gas_min = int(config.get('ETH_GAS_GIFTER_MINIMUM_BALANCE'))
+    if r < gas_min:
+        logg.error('EEK! gas gifter has balance {}, below minimum {}'.format(r, gas_min))
+        return False
+
+    return True

apps/cic-eth/cic_eth/check/signer.py

@@ -0,0 +1,37 @@
+# standard imports
+import time
+import logging
+from urllib.error import URLError
+
+# external imports
+from chainlib.connection import RPCConnection
+from chainlib.eth.constant import ZERO_ADDRESS
+from chainlib.eth.sign import sign_message
+from chainlib.error import JSONRPCException
+
+logg = logging.getLogger().getChild(__name__)
+
+
+def health(*args, **kwargs):
+    blocked = True
+    max_attempts = 5
+    conn = RPCConnection.connect(kwargs['config'].get('CIC_CHAIN_SPEC'), tag='signer')
+    for i in range(max_attempts):
+        idx = i + 1
+        logg.debug('attempt signer connection check {}/{}'.format(idx, max_attempts))
+        try:
+            conn.do(sign_message(ZERO_ADDRESS, '0x2a'))
+        except FileNotFoundError:
+            pass
+        except ConnectionError:
+            pass
+        except URLError:
+            pass
+        except JSONRPCException:
+            logg.debug('signer connection succeeded')
+            return True
+
+        if idx < max_attempts:
+            time.sleep(0.5)
+
+    return False

apps/cic-eth/cic_eth/db/enum.py

@@ -74,10 +74,11 @@ class LockEnum(enum.IntEnum):
     QUEUE: Disable queueing new or modified transactions
     """
     STICKY=1
-    CREATE=2
-    SEND=4
-    QUEUE=8
-    QUERY=16
+    INIT=2
+    CREATE=4
+    SEND=8
+    QUEUE=16
+    QUERY=32
     ALL=int(0xfffffffffffffffe)
 
 

apps/cic-eth/cic_eth/db/migrations/default/versions/75d4767b3031_lock.py

@@ -5,8 +5,11 @@ Revises: 1f1b3b641d08
 Create Date: 2021-04-02 18:41:20.864265
 
 """
+import datetime
 from alembic import op
 import sqlalchemy as sa
+from chainlib.eth.constant import ZERO_ADDRESS
+from cic_eth.db.enum import LockEnum
 
 
 # revision identifiers, used by Alembic.
@@ -23,10 +26,11 @@ def upgrade():
             sa.Column("address", sa.String(42), nullable=True),
             sa.Column('blockchain', sa.String),
             sa.Column("flags", sa.BIGINT(), nullable=False, default=0),
-            sa.Column("date_created", sa.DateTime, nullable=False),
+            sa.Column("date_created", sa.DateTime, nullable=False, default=datetime.datetime.utcnow),
             sa.Column("otx_id", sa.Integer, sa.ForeignKey('otx.id'), nullable=True),
             )
     op.create_index('idx_chain_address', 'lock', ['blockchain', 'address'], unique=True)
+    op.execute("INSERT INTO lock (address, date_created, blockchain, flags) VALUES('{}', '{}', '::', {})".format(ZERO_ADDRESS, datetime.datetime.utcnow(), LockEnum.INIT | LockEnum.SEND | LockEnum.QUEUE))
 
 
 def downgrade():

apps/cic-eth/cic_eth/db/models/base.py

@@ -10,6 +10,7 @@ from sqlalchemy.pool import (
         StaticPool,
         QueuePool,
         AssertionPool,
+        NullPool,
         )
 
 logg = logging.getLogger()
@@ -64,6 +65,7 @@ class SessionBase(Model):
         if SessionBase.poolable:
             poolclass = QueuePool
             if pool_size > 1:
+                logg.info('db using queue pool')
                 e = create_engine(
                         dsn,
                         max_overflow=pool_size*3,
@@ -74,17 +76,22 @@ class SessionBase(Model):
                         echo=debug,
                     )
             else:
-                if debug:
+                if pool_size == 0:
+                    logg.info('db using nullpool')
+                    poolclass = NullPool
+                elif debug:
+                    logg.info('db using assertion pool')
                     poolclass = AssertionPool
                 else:
+                    logg.info('db using static pool')
                     poolclass = StaticPool
-
                 e = create_engine(
                         dsn,
                         poolclass=poolclass,
                         echo=debug,
                     )
         else:
+            logg.info('db not poolable')
             e = create_engine(
                     dsn,
                     echo=debug,

apps/cic-eth/cic_eth/eth/meta.py

@@ -1,6 +1,10 @@
-# extended imports
+# external imports
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.status import Status as TxStatus
+from cic_eth_registry.erc20 import ERC20Token
+
+# local imports
+from cic_eth.ext.address import translate_address
 
 
 class ExtendedTx:
@@ -27,12 +31,12 @@ class ExtendedTx:
         self.status_code = TxStatus.PENDING.value
 
 
-    def set_actors(self, sender, recipient, trusted_declarator_addresses=None):
+    def set_actors(self, sender, recipient, trusted_declarator_addresses=None, caller_address=ZERO_ADDRESS):
         self.sender = sender
         self.recipient = recipient
         if trusted_declarator_addresses != None:
-            self.sender_label = translate_address(sender, trusted_declarator_addresses, self.chain_spec)
-            self.recipient_label = translate_address(recipient, trusted_declarator_addresses, self.chain_spec)
+            self.sender_label = translate_address(sender, trusted_declarator_addresses, self.chain_spec, sender_address=caller_address)
+            self.recipient_label = translate_address(recipient, trusted_declarator_addresses, self.chain_spec, sender_address=caller_address)
 
 
     def set_tokens(self, source, source_value, destination=None, destination_value=None):
@@ -40,8 +44,8 @@ class ExtendedTx:
             destination = source
         if destination_value == None:
             destination_value = source_value
-        st = ERC20Token(self.rpc, source)
-        dt = ERC20Token(self.rpc, destination)
+        st = ERC20Token(self.chain_spec, self.rpc, source)
+        dt = ERC20Token(self.chain_spec, self.rpc, destination)
         self.source_token = source
         self.source_token_symbol = st.symbol
         self.source_token_name = st.name
@@ -62,10 +66,10 @@ class ExtendedTx:
         self.status_code = n
 
 
-    def to_dict(self):
+    def asdict(self):
         o = {}
         for attr in dir(self):
-            if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'to_dict']:
+            if attr[0] == '_' or attr in ['set_actors', 'set_tokens', 'set_status', 'asdict', 'rpc']:
                 continue
             o[attr] = getattr(self, attr)
         return o 

apps/cic-eth/cic_eth/runnable/ctrl.py

@@ -59,6 +59,7 @@ args_override = {
         'CIC_CHAIN_SPEC': getattr(args, 'i'),
         }
 # override args
+config.dict_override(args_override, 'cli')
 config.censor('PASSWORD', 'DATABASE')
 config.censor('PASSWORD', 'SSL')
 logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
@@ -67,7 +68,9 @@ celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=confi
 
 queue = args.q
 
-chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+chain_spec = None
+if config.get('CIC_CHAIN_SPEC') != None and config.get('CIC_CHAIN_SPEC') != '::':
+    chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
 admin_api = AdminApi(None)
 
 
@@ -82,6 +85,9 @@ def lock_names_to_flag(s):
 
 # TODO: move each command to submodule
 def main():
+    chain_spec_dict = None
+    if chain_spec != None:
+        chain_spec_dict = chain_spec.asdict()
     if args.command == 'unlock':
         flags = lock_names_to_flag(args.flags)
         if not is_checksum_address(args.address):
@@ -91,7 +97,7 @@ def main():
             'cic_eth.admin.ctrl.unlock',
             [
                 None,
-                chain_spec.asdict(),
+                chain_spec_dict,
                 args.address,
                 flags,
                 ],
@@ -110,7 +116,7 @@ def main():
             'cic_eth.admin.ctrl.lock',
             [
                 None,
-                chain_spec.asdict(),
+                chain_spec_dict,
                 args.address,
                 flags,
                 ],

apps/cic-eth/cic_eth/runnable/daemons/dispatcher.py

@@ -15,7 +15,6 @@ from cic_eth_registry import CICRegistry
 from chainlib.chain import ChainSpec
 from chainlib.eth.tx import unpack
 from chainlib.connection import RPCConnection
-from chainsyncer.error import SyncDone
 from hexathon import strip_0x
 from chainqueue.db.enum import (
     StatusEnum,
@@ -153,10 +152,7 @@ class DispatchSyncer:
 def main(): 
     syncer = DispatchSyncer(chain_spec)
     conn = RPCConnection.connect(chain_spec, 'default')
-    try:
-        syncer.loop(conn, float(config.get('DISPATCHER_LOOP_INTERVAL')))
-    except SyncDone as e:
-        sys.stderr.write("dispatcher done at block {}\n".format(e))
+    syncer.loop(conn, float(config.get('DISPATCHER_LOOP_INTERVAL')))
 
     sys.exit(0)
 

apps/cic-eth/cic_eth/runnable/daemons/filters/callback.py

@@ -1,7 +1,7 @@
 # standard imports
 import logging
 
-# third-party imports
+# external imports
 import celery
 from cic_eth_registry.error import UnknownContractError
 from chainlib.status import Status as TxStatus
@@ -9,7 +9,13 @@ from chainlib.eth.address import to_checksum_address
 from chainlib.eth.error import RequestMismatchException
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.eth.erc20 import ERC20
-from hexathon import strip_0x
+from hexathon import (
+        strip_0x,
+        add_0x,
+        )
+# TODO: use sarafu_Faucet for both when inheritance has been implemented
+from erc20_single_shot_faucet import SingleShotFaucet
+from sarafu_faucet import MinterFaucet as Faucet
 
 # local imports
 from .base import SyncFilter
@@ -18,65 +24,73 @@ from cic_eth.eth.meta import ExtendedTx
 logg = logging.getLogger().getChild(__name__)
 
 
-def parse_transfer(tx):
-    r = ERC20.parse_transfer_request(tx.payload)
-    transfer_data = {}
-    transfer_data['to'] = r[0]
-    transfer_data['value'] = r[1]
-    transfer_data['from'] = tx['from']
-    transfer_data['token_address'] = tx['to']
-    return ('transfer', transfer_data)
-
-
-def parse_transferfrom(tx):
-    r = ERC20.parse_transfer_request(tx.payload)
-    transfer_data = unpack_transferfrom(tx.payload)
-    transfer_data['from'] = r[0]
-    transfer_data['to'] = r[1]
-    transfer_data['value'] = r[2]
-    transfer_data['token_address'] = tx['to']
-    return ('transferfrom', transfer_data)
-
-
-def parse_giftto(tx):
-    # TODO: broken
-    logg.error('broken')
-    return
-    transfer_data = unpack_gift(tx.payload)
-    transfer_data['from'] = tx.inputs[0]
-    transfer_data['value'] = 0
-    transfer_data['token_address'] = ZERO_ADDRESS
-    # TODO: would be better to query the gift amount from the block state
-    for l in tx.logs:
-        topics = l['topics']
-        logg.debug('topixx {}'.format(topics))
-        if strip_0x(topics[0]) == '45c201a59ac545000ead84f30b2db67da23353aa1d58ac522c48505412143ffa':
-            #transfer_data['value'] = web3.Web3.toInt(hexstr=strip_0x(l['data']))
-            transfer_data['value'] = int.from_bytes(bytes.fromhex(strip_0x(l_data)))
-            #token_address_bytes = topics[2][32-20:]
-            token_address = strip_0x(topics[2])[64-40:]
-            transfer_data['token_address'] = to_checksum_address(token_address)
-    return ('tokengift', transfer_data)
-
 
 class CallbackFilter(SyncFilter):
 
     trusted_addresses = []
 
-    def __init__(self, chain_spec, method, queue):
+    def __init__(self, chain_spec, method, queue, caller_address=ZERO_ADDRESS):
         self.queue = queue
         self.method = method
         self.chain_spec = chain_spec
+        self.caller_address = caller_address
+
+
+    def parse_transfer(self, tx, conn):
+        if not tx.payload:
+            return (None, None)
+        r = ERC20.parse_transfer_request(tx.payload)
+        transfer_data = {}
+        transfer_data['to'] = r[0]
+        transfer_data['value'] = r[1]
+        transfer_data['from'] = tx.outputs[0]
+        transfer_data['token_address'] = tx.inputs[0]
+        return ('transfer', transfer_data)
+
+
+    def parse_transferfrom(self, tx, conn):
+        if not tx.payload:
+            return (None, None)
+        r = ERC20.parse_transfer_from_request(tx.payload)
+        transfer_data = {}
+        transfer_data['from'] = r[0]
+        transfer_data['to'] = r[1]
+        transfer_data['value'] = r[2]
+        transfer_data['token_address'] = tx.inputs[0]
+        return ('transferfrom', transfer_data)
+
+
+    def parse_giftto(self, tx, conn):
+        if not tx.payload:
+            return (None, None)
+        r = Faucet.parse_give_to_request(tx.payload)
+        transfer_data = {}
+        transfer_data['to'] = r[0]
+        transfer_data['value'] = tx.value
+        transfer_data['from'] = tx.outputs[0]
+        #transfer_data['token_address'] = tx.inputs[0]
+        faucet_contract = tx.inputs[0]
+
+        c = SingleShotFaucet(self.chain_spec)
+        o = c.token(faucet_contract, sender_address=self.caller_address)
+        r = conn.do(o)
+        transfer_data['token_address'] = add_0x(c.parse_token(r))
+
+        o = c.amount(faucet_contract, sender_address=self.caller_address)
+        r = conn.do(o)
+        transfer_data['value'] = c.parse_amount(r)
+
+        return ('tokengift', transfer_data)
 
 
     def call_back(self, transfer_type, result):
-        logg.debug('result {}'.format(result))
+        result['chain_spec'] = result['chain_spec'].asdict()
         s = celery.signature(
             self.method,
             [
                 result,
                 transfer_type,
-                int(result['status_code'] == 0),
+                int(result['status_code'] != 0),
             ],
             queue=self.queue,
             )
@@ -92,26 +106,29 @@ class CallbackFilter(SyncFilter):
 #        s_translate.link(s)
 #        s_translate.apply_async()
         t = s.apply_async()
-        return s
+        return t
 
 
-    def parse_data(self, tx):
+    def parse_data(self, tx, conn):
         transfer_type = None
         transfer_data = None
         # TODO: what's with the mix of attributes and dict keys
         logg.debug('have payload {}'.format(tx.payload))
-        method_signature = tx.payload[:8]
 
         logg.debug('tx status {}'.format(tx.status))
 
         for parser in [
-                parse_transfer,
-                parse_transferfrom,
-                parse_giftto,
+                self.parse_transfer,
+                self.parse_transferfrom,
+                self.parse_giftto,
                 ]:
             try:
-                (transfer_type, transfer_data) = parser(tx)
-                break
+                if tx:
+                    (transfer_type, transfer_data) = parser(tx, conn)
+                    if transfer_type == None:
+                        continue
+                else:
+                    pass
             except RequestMismatchException:
                 continue
 
@@ -128,7 +145,7 @@ class CallbackFilter(SyncFilter):
         transfer_data = None
         transfer_type = None
         try:
-            (transfer_type, transfer_data) = self.parse_data(tx)
+            (transfer_type, transfer_data) = self.parse_data(tx, conn)
         except TypeError:
             logg.debug('invalid method data length for tx {}'.format(tx.hash))
             return
@@ -144,16 +161,17 @@ class CallbackFilter(SyncFilter):
             result = None
             try:
                 tokentx = ExtendedTx(conn, tx.hash, self.chain_spec)
-                tokentx.set_actors(transfer_data['from'], transfer_data['to'], self.trusted_addresses)
+                tokentx.set_actors(transfer_data['from'], transfer_data['to'], self.trusted_addresses, caller_address=self.caller_address)
                 tokentx.set_tokens(transfer_data['token_address'], transfer_data['value'])
                 if transfer_data['status'] == 0:
                     tokentx.set_status(1)
                 else:
                     tokentx.set_status(0)
-                t = self.call_back(transfer_type, tokentx.to_dict())
-                logg.info('callback success task id {} tx {}'.format(t, tx.hash))
+                result = tokentx.asdict()
+                t = self.call_back(transfer_type, result)
+                logg.info('callback success task id {} tx {} queue {}'.format(t, tx.hash, t.queue))
             except UnknownContractError:
-                logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tc.queue, tc.method, transfer_data['to'], tx.hash))
+                logg.debug('callback filter {}:{} skipping "transfer" method on unknown contract {} tx {}'.format(tx.queue, tx.method, transfer_data['to'], tx.hash))
 
 
     def __str__(self):

apps/cic-eth/cic_eth/runnable/daemons/retry.py

@@ -20,6 +20,7 @@ from cic_eth.admin.ctrl import lock_send
 from cic_eth.db.enum import LockEnum
 from cic_eth.runnable.daemons.filters.straggler import StragglerFilter
 from cic_eth.sync.retry import RetrySyncer
+from cic_eth.stat import init_chain_stat
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -71,57 +72,21 @@ RPCConnection.register_location(config.get('ETH_PROVIDER'), chain_spec, tag='def
 dsn = dsn_from_config(config)
 SessionBase.connect(dsn, debug=config.true('DATABASE_DEBUG'))
 
-straggler_delay = int(config.get('CIC_TX_RETRY_DELAY'))
-
-## TODO: we already have the signed raw tx in get, so its a waste of cycles to get_tx here
-#def sendfail_filter(w3, tx_hash, rcpt, chain_spec):
-#    tx_dict = get_tx(tx_hash)
-#    tx = unpack(tx_dict['signed_tx'], chain_spec)
-#    logg.debug('submitting tx {} for retry'.format(tx_hash))
-#    s_check = celery.signature(
-#            'cic_eth.admin.ctrl.check_lock',
-#            [
-#                tx_hash,
-#                chain_str,
-#                LockEnum.QUEUE,
-#                tx['from'],
-#                ],
-#            queue=queue,
-#            )
-##    s_resume = celery.signature(
-##            'cic_eth.eth.tx.resume_tx',
-##            [
-##                chain_str,
-##                ],
-##            queue=queue,
-##            )
-#    
-##    s_retry_status = celery.signature(
-##            'cic_eth.queue.state.set_ready',
-##            [],
-##            queue=queue,
-##    )
-#    s_resend = celery.signature(
-#            'cic_eth.eth.gas.resend_with_higher_gas',
-#            [
-#                chain_str,
-#                ],
-#            queue=queue,
-#            )
-#
-#    #s_resume.link(s_retry_status)
-#    #s_check.link(s_resume)
-#    s_check.link(s_resend)
-#    s_check.apply_async()
-
 
 def main(): 
     conn = RPCConnection.connect(chain_spec, 'default')
+
+    straggler_delay = int(config.get('CIC_TX_RETRY_DELAY'))
+    loop_interval = config.get('SYNCER_LOOP_INTERVAL')
+    if loop_interval == None:
+        stat = init_chain_stat(conn)
+        loop_interval = stat.block_average()
+
     syncer = RetrySyncer(conn, chain_spec, straggler_delay, batch_size=config.get('_BATCH_SIZE'))
     syncer.backend.set(0, 0)
     fltr = StragglerFilter(chain_spec, queue=queue)
     syncer.add_filter(fltr)
-    syncer.loop(float(straggler_delay), conn)
+    syncer.loop(int(loop_interval), conn)
 
 
 if __name__ == '__main__':

apps/cic-eth/cic_eth/runnable/daemons/tasker.py

@@ -15,6 +15,9 @@ from chainlib.connection import RPCConnection
 from chainlib.eth.connection import EthUnixSignerConnection
 from chainlib.chain import ChainSpec
 from chainqueue.db.models.otx import Otx
+from cic_eth_registry.error import UnknownContractError
+import liveness.linux
+
 
 # local imports
 from cic_eth.eth import (
@@ -27,6 +30,7 @@ from cic_eth.eth import (
 from cic_eth.admin import (
         debug,
         ctrl,
+        token,
         )
 from cic_eth.queue import (
         query,
@@ -39,6 +43,7 @@ from cic_eth.queue import (
 from cic_eth.callbacks import (
         Callback,
         http,
+        noop,
         #tcp,
         redis,
         )
@@ -50,6 +55,8 @@ from cic_eth.registry import (
         connect_declarator,
         connect_token_registry,
         )
+from cic_eth.task import BaseTask
+
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -61,6 +68,7 @@ argparser.add_argument('-p', '--provider', dest='p', type=str, help='rpc provide
 argparser.add_argument('-c', type=str, default=config_dir, help='config file')
 argparser.add_argument('-q', type=str, default='cic-eth', help='queue name for worker tasks')
 argparser.add_argument('-r', type=str, help='CIC registry address')
+argparser.add_argument('--default-token-symbol', dest='default_token_symbol', type=str, help='Symbol of default token to use')
 argparser.add_argument('--abi-dir', dest='abi_dir', type=str, help='Directory containing bytecode and abi')
 argparser.add_argument('--trace-queue-status', default=None, dest='trace_queue_status', action='store_true', help='set to perist all queue entry status changes to storage')
 argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')
@@ -80,6 +88,7 @@ config.process()
 args_override = {
         'CIC_CHAIN_SPEC': getattr(args, 'i'),
         'CIC_REGISTRY_ADDRESS': getattr(args, 'r'),
+        'CIC_DEFAULT_TOKEN_SYMBOL': getattr(args, 'default_token_symbol'),
         'ETH_PROVIDER': getattr(args, 'p'),
         'TASKS_TRACE_QUEUE_STATUS': getattr(args, 'trace_queue_status'),
         }
@@ -89,14 +98,15 @@ config.censor('PASSWORD', 'DATABASE')
 config.censor('PASSWORD', 'SSL')
 logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 
+health_modules = config.get('CIC_HEALTH_MODULES', [])
+if len(health_modules) != 0:
+    health_modules = health_modules.split(',')
+logg.debug('health mods {}'.format(health_modules))
+
 # connect to database
 dsn = dsn_from_config(config)
-SessionBase.connect(dsn, pool_size=50, debug=config.true('DATABASE_DEBUG'))
+SessionBase.connect(dsn, pool_size=int(config.get('DATABASE_POOL_SIZE')), debug=config.true('DATABASE_DEBUG'))
 
-# verify database connection with minimal sanity query
-session = SessionBase.create_session()
-session.execute('select version_num from alembic_version')
-session.close()
 
 # set up celery
 current_app = celery.Celery(__name__)
@@ -134,10 +144,15 @@ else:
 
 chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
 RPCConnection.register_location(config.get('ETH_PROVIDER'), chain_spec, 'default')
-RPCConnection.register_location(config.get('SIGNER_SOCKET_PATH'), chain_spec, 'signer', constructor=EthUnixSignerConnection)
+#RPCConnection.register_location(config.get('SIGNER_SOCKET_PATH'), chain_spec, 'signer', constructor=EthUnixSignerConnection)
+RPCConnection.register_location(config.get('SIGNER_SOCKET_PATH'), chain_spec, 'signer')
 
 Otx.tracing = config.true('TASKS_TRACE_QUEUE_STATUS')
 
+#import cic_eth.checks.gas
+#if not cic_eth.checks.gas.health(config=config):
+#    raise RuntimeError()
+liveness.linux.load(health_modules, rundir=config.get('CIC_RUN_DIR'), config=config)
 
 def main():
     argv = ['worker']
@@ -161,7 +176,11 @@ def main():
 
     rpc = RPCConnection.connect(chain_spec, 'default')
 
-    connect_registry(rpc, chain_spec, config.get('CIC_REGISTRY_ADDRESS'))
+    try:
+        registry = connect_registry(rpc, chain_spec, config.get('CIC_REGISTRY_ADDRESS'))
+    except UnknownContractError as e:
+        logg.exception('Registry contract connection failed for {}: {}'.format(config.get('CIC_REGISTRY_ADDRESS'), e))
+        sys.exit(1)
 
     trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
     if trusted_addresses_src == None:
@@ -170,10 +189,17 @@ def main():
     trusted_addresses = trusted_addresses_src.split(',')
     for address in trusted_addresses:
         logg.info('using trusted address {}'.format(address))
+
     connect_declarator(rpc, chain_spec, trusted_addresses)
     connect_token_registry(rpc, chain_spec)
-    
+
+    BaseTask.default_token_symbol = config.get('CIC_DEFAULT_TOKEN_SYMBOL')
+    BaseTask.default_token_address = registry.by_name(BaseTask.default_token_symbol)
+    logg.info('default token set to {} {}'.format(BaseTask.default_token_symbol, BaseTask.default_token_address))
+   
+    liveness.linux.set()
     current_app.worker_main(argv)
+    liveness.linux.reset()
 
 
 @celery.signals.eventlet_pool_postshutdown.connect

apps/cic-eth/cic_eth/runnable/daemons/tracker.py

@@ -7,7 +7,7 @@ import argparse
 import sys
 import re
 
-# third-party imports
+# external imports
 import confini
 import celery
 import rlp
@@ -15,7 +15,6 @@ import cic_base.config
 import cic_base.log
 import cic_base.argparse
 import cic_base.rpc
-from cic_eth_registry import CICRegistry
 from cic_eth_registry.error import UnknownContractError
 from chainlib.chain import ChainSpec
 from chainlib.eth.constant import ZERO_ADDRESS
@@ -26,7 +25,7 @@ from chainlib.eth.block import (
 from hexathon import (
         strip_0x,
         )
-from chainsyncer.backend import SyncerBackend
+from chainsyncer.backend.sql import SQLBackend
 from chainsyncer.driver import (
         HeadSyncer,
         HistorySyncer,
@@ -42,6 +41,13 @@ from cic_eth.runnable.daemons.filters import (
         RegistrationFilter,
         TransferAuthFilter,
         )
+from cic_eth.stat import init_chain_stat
+from cic_eth.registry import (
+        connect as connect_registry,
+        connect_declarator,
+        connect_token_registry,
+        )
+
 
 script_dir = os.path.realpath(os.path.dirname(__file__))
 
@@ -78,24 +84,27 @@ def main():
     block_current = int(r, 16)
     block_offset = block_current + 1
 
-    stat = init_chain_stat(rpc, block_current)
+    loop_interval = config.get('SYNCER_LOOP_INTERVAL')
+    if loop_interval == None:
+        stat = init_chain_stat(rpc, block_start=block_current)
+        loop_interval = stat.block_average()
 
     logg.debug('starting at block {}'.format(block_offset))
 
     syncers = []
 
-    #if SyncerBackend.first(chain_spec):
-    #    backend = SyncerBackend.initial(chain_spec, block_offset)
-    syncer_backends = SyncerBackend.resume(chain_spec, block_offset)
+    #if SQLBackend.first(chain_spec):
+    #    backend = SQLBackend.initial(chain_spec, block_offset)
+    syncer_backends = SQLBackend.resume(chain_spec, block_offset)
 
     if len(syncer_backends) == 0:
         logg.info('found no backends to resume')
-        syncer_backends.append(SyncerBackend.initial(chain_spec, block_offset))
+        syncer_backends.append(SQLBackend.initial(chain_spec, block_offset))
     else:
         for syncer_backend in syncer_backends:
             logg.info('resuming sync session {}'.format(syncer_backend))
 
-    syncer_backends.append(SyncerBackend.live(chain_spec, block_offset+1))
+    syncer_backends.append(SQLBackend.live(chain_spec, block_offset+1))
 
     for syncer_backend in syncer_backends:
         try:
@@ -105,6 +114,8 @@ def main():
             logg.info('Initializing HEAD syncer on backend {}'.format(syncer_backend))
             syncers.append(HeadSyncer(syncer_backend))
 
+    connect_registry(rpc, chain_spec, config.get('CIC_REGISTRY_ADDRESS'))
+
     trusted_addresses_src = config.get('CIC_TRUST_ADDRESS')
     if trusted_addresses_src == None:
         logg.critical('At least one trusted address must be declared in CIC_TRUST_ADDRESS')
@@ -112,6 +123,8 @@ def main():
     trusted_addresses = trusted_addresses_src.split(',')
     for address in trusted_addresses:
         logg.info('using trusted address {}'.format(address))
+    connect_declarator(rpc, chain_spec, trusted_addresses)
+    connect_token_registry(rpc, chain_spec)
     CallbackFilter.trusted_addresses = trusted_addresses
 
     callback_filters = []
@@ -142,7 +155,8 @@ def main():
         for cf in callback_filters:
             syncer.add_filter(cf)
 
-        r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
+        #r = syncer.loop(int(config.get('SYNCER_LOOP_INTERVAL')), rpc)
+        r = syncer.loop(int(loop_interval), rpc)
         sys.stderr.write("sync {} done at block {}\n".format(syncer, r))
 
         i += 1

apps/cic-eth/cic_eth/stat.py

@@ -0,0 +1,33 @@
+# standard imports
+import logging
+
+# external imports
+from chainlib.stat import ChainStat
+from chainlib.eth.block import (
+        block_latest,
+        block_by_number,
+        Block,
+        )
+
+logg = logging.getLogger().getChild(__name__)
+
+BLOCK_SAMPLES = 10
+
+
+def init_chain_stat(rpc, block_start=0):
+    stat = ChainStat()
+
+    if block_start == 0:
+        o = block_latest()
+        r = rpc.do(o)
+        block_start = int(r, 16)
+
+    for i in range(BLOCK_SAMPLES):
+        o = block_by_number(block_start-10+i)
+        block_src = rpc.do(o)
+        logg.debug('block {}'.format(block_src))
+        block = Block(block_src)
+        stat.block_apply(block)
+
+    logg.debug('calculated block time {} from {} block samples'.format(stat.block_average(), BLOCK_SAMPLES))
+    return stat

apps/cic-eth/cic_eth/sync/retry.py

@@ -4,7 +4,7 @@ import datetime
 
 # external imports
 from chainsyncer.driver import HeadSyncer
-from chainsyncer.backend import MemBackend
+from chainsyncer.backend.memory import MemBackend
 from chainsyncer.error import NoBlockForYou
 from chainlib.eth.block import (
         block_by_number,

apps/cic-eth/cic_eth/task.py

@@ -29,6 +29,8 @@ class BaseTask(celery.Task):
     call_address = ZERO_ADDRESS
     create_nonce_oracle = RPCNonceOracle
     create_gas_oracle = RPCGasOracle
+    default_token_address = None
+    default_token_symbol = None
 
     def create_session(self):
         return BaseTask.session_func()
@@ -94,3 +96,8 @@ class CriticalWeb3AndSignerTask(CriticalTask):
 def hello(self):
     time.sleep(0.1)
     return id(SessionBase.create_session)
+
+
+@celery_app.task()
+def check_health(self):
+    celery.app.control.shutdown()

apps/cic-eth/cic_eth/version.py

@@ -10,7 +10,7 @@ version = (
         0,
         11,
         0,
-        'beta.1',
+        'beta.8',
         )
 
 version_object = semver.VersionInfo(

apps/cic-eth/config/cic.ini

@@ -3,3 +3,6 @@ registry_address =
 chain_spec = evm:bloxberg:8996
 tx_retry_delay = 
 trust_address =
+default_token_symbol = GFT
+health_modules = cic_eth.check.db,cic_eth.check.signer,cic_eth.check.gas
+run_dir = /run

apps/cic-eth/config/database.ini

@@ -6,4 +6,5 @@ HOST=localhost
 PORT=5432
 ENGINE=postgresql
 DRIVER=psycopg2
+POOL_SIZE=50
 DEBUG=0

apps/cic-eth/config/docker/cic.ini

@@ -2,3 +2,7 @@
 registry_address = 
 chain_spec = evm:bloxberg:8996
 trust_address = 0xEb3907eCad74a0013c259D5874AE7f22DcBcC95C
+tx_retry_delay = 20
+default_token_symbol = GFT
+health_modules = cic_eth.check.db,cic_eth.check.signer,cic_eth.check.gas
+run_dir = /run

apps/cic-eth/config/docker/database.ini

@@ -6,4 +6,5 @@ HOST=localhost
 PORT=63432
 ENGINE=postgresql
 DRIVER=psycopg2
+POOL_SIZE=50
 DEBUG=0

apps/cic-eth/config/docker/eth.ini

@@ -1,8 +1,4 @@
 [eth]
-#ws_provider = ws://localhost:8546
-#ttp_provider = http://localhost:8545
 provider = http://localhost:63545
-gas_provider_address = 
-#chain_id = 
-abi_dir = /home/lash/src/ext/cic/grassrootseconomics/cic-contracts/abis
-account_accounts_index_writer = 
+health_modules = cic_eth.check.db,cic_eth.check.gas
+gas_gifter_minimum_balance = 10000000000000000000000

apps/cic-eth/config/docker/signer.ini

@@ -1,5 +1,5 @@
 [signer]
-socket_path = /tmp/crypto-dev-signer/jsonrpc.ipc
+socket_path = ipc:///tmp/crypto-dev-signer/jsonrpc.ipc
 secret = deedbeef
 database_name = signer_test
 dev_keys_path = 

apps/cic-eth/config/docker/syncer.ini

@@ -1,2 +1,2 @@
 [SYNCER]
-loop_interval = 1
+loop_interval = 

apps/cic-eth/config/eth.ini

@@ -1,8 +1,4 @@
 [eth]
-#ws_provider = ws://localhost:8546
-#ttp_provider = http://localhost:8545
 provider = http://localhost:8545
-gas_provider_address = 
-#chain_id = 
-abi_dir = /usr/local/share/cic/solidity/abi
-account_accounts_index_writer = 
+gas_gifter_minimum_balance = 10000000000000000000000
+health_modules = cic_eth.check.db,cic_eth.check.gas

apps/cic-eth/config/syncer.ini

@@ -1,2 +1,2 @@
 [SYNCER]
-loop_interval = 1
+loop_interval = 

apps/cic-eth/docker/Dockerfile

@@ -29,7 +29,7 @@ RUN /usr/local/bin/python -m pip install --upgrade pip
 #	python merge_requirements.py | tee merged_requirements.txt 
 #RUN cd cic-base && \
 #	pip install $pip_extra_index_url_flag -r ./merged_requirements.txt 
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a61
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a77
 
 COPY cic-eth/scripts/ scripts/
 COPY cic-eth/setup.cfg cic-eth/setup.py ./
@@ -53,3 +53,5 @@ COPY cic-eth/crypto_dev_signer_config/ /usr/local/etc/crypto-dev-signer/
 RUN git clone https://gitlab.com/grassrootseconomics/cic-contracts.git && \
 	mkdir -p /usr/local/share/cic/solidity && \
 	cp -R cic-contracts/abis /usr/local/share/cic/solidity/abi
+
+COPY util/liveness/health.sh /usr/local/bin/health.sh

apps/cic-eth/requirements.txt

@@ -1,25 +1,25 @@
-cic-base~=0.1.2a60
+cic-base==0.1.2b1
 celery==4.4.7
-crypto-dev-signer~=0.4.14a17
+crypto-dev-signer~=0.4.14b3
 confini~=0.3.6rc3
-cic-eth-registry~=0.5.4a12
+cic-eth-registry~=0.5.4a16
 #cic-bancor~=0.0.6
 redis==3.5.3
 alembic==1.4.2
 websockets==8.1
 requests~=2.24.0
-eth_accounts_index~=0.0.11a7
-erc20-transfer-authorization~=0.3.1a3
-#simple-rlp==0.1.2
+eth_accounts_index~=0.0.11a9
+erc20-transfer-authorization~=0.3.1a5
 uWSGI==2.0.19.1
 semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
-eth-address-index~=0.1.1a7
-chainlib~=0.0.2a4
+eth-address-index~=0.1.1a9
+chainlib~=0.0.2a13
 hexathon~=0.0.1a7
-chainsyncer~=0.0.1a21
+chainsyncer[sql]~=0.0.2a2
 chainqueue~=0.0.1a7
 pysha3==1.0.2
 coincurve==15.0.0
-sarafu-faucet~=0.0.2a19
+sarafu-faucet==0.0.2a28
+potaahto~=0.0.1a1

apps/cic-eth/setup.cfg

@@ -38,6 +38,7 @@ packages =
 	cic_eth.runnable.daemons.filters
 	cic_eth.callbacks
 	cic_eth.sync
+	cic_eth.check
 scripts =
 	./scripts/migrate.py
 
@@ -52,6 +53,7 @@ console_scripts =
 	cic-eth-create = cic_eth.runnable.create:main 
 	cic-eth-inspect = cic_eth.runnable.view:main
 	cic-eth-ctl = cic_eth.runnable.ctrl:main
+	cic-eth-info = cic_eth.runnable.info:main
 	# TODO: Merge this with ctl when subcmds sorted to submodules
 	cic-eth-tag = cic_eth.runnable.tag:main
 	cic-eth-resend = cic_eth.runnable.resend:main

apps/cic-eth/test_requirements.txt

@@ -4,4 +4,4 @@ pytest-mock==3.3.1
 pytest-cov==2.10.1
 eth-tester==0.5.0b3
 py-evm==0.3.0a20
-giftable-erc20-token==0.0.8a4
+giftable-erc20-token==0.0.8a9

apps/cic-eth/tests/filters/test_callback_filter.py

@@ -0,0 +1,223 @@
+# standard import
+import logging
+import datetime
+import os
+
+# external imports
+import pytest
+from chainlib.connection import RPCConnection
+from chainlib.eth.nonce import RPCNonceOracle
+from chainlib.eth.gas import OverrideGasOracle
+from chainlib.eth.tx import (
+        receipt,
+        transaction,
+        Tx,
+        )
+from chainlib.eth.block import Block
+from chainlib.eth.erc20 import ERC20
+from sarafu_faucet import MinterFaucet
+from eth_accounts_index import AccountRegistry
+from potaahto.symbols import snake_and_camel
+from hexathon import add_0x
+
+# local imports
+from cic_eth.runnable.daemons.filters.callback import CallbackFilter
+
+logg = logging.getLogger()
+
+
+@pytest.mark.skip()
+def test_transfer_tx(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        foo_token,
+        agent_roles,
+        token_roles,
+        contract_roles,
+        celery_session_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(token_roles['FOO_TOKEN_OWNER'], rpc)
+    gas_oracle = OverrideGasOracle(conn=rpc, limit=200000)
+   
+    txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = txf.transfer(foo_token, token_roles['FOO_TOKEN_OWNER'], agent_roles['ALICE'], 1024)
+    r = rpc.do(o)
+    
+    o = transaction(tx_hash_hex)
+    r = rpc.do(o)
+    logg.debug(r)
+    tx_src = snake_and_camel(r)
+    tx = Tx(tx_src)
+
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+
+    rcpt = snake_and_camel(r)
+    tx.apply_receipt(rcpt)
+
+    fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
+    (transfer_type, transfer_data) = fltr.parse_transfer(tx, eth_rpc)
+
+    assert transfer_type == 'transfer'
+
+
+@pytest.mark.skip()
+def test_transfer_from_tx(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        foo_token,
+        agent_roles,
+        token_roles,
+        contract_roles,
+        celery_session_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(token_roles['FOO_TOKEN_OWNER'], rpc)
+    gas_oracle = OverrideGasOracle(conn=rpc, limit=200000)
+   
+    txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+
+    (tx_hash_hex, o) = txf.approve(foo_token, token_roles['FOO_TOKEN_OWNER'], agent_roles['ALICE'], 1024)
+    r = rpc.do(o)
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], rpc)
+    txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = txf.transfer_from(foo_token, agent_roles['ALICE'], token_roles['FOO_TOKEN_OWNER'], agent_roles['BOB'], 1024)
+    r = rpc.do(o)
+    
+    o = transaction(tx_hash_hex)
+    r = rpc.do(o)
+    tx_src = snake_and_camel(r)
+    tx = Tx(tx_src)
+
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+
+    rcpt = snake_and_camel(r)
+    tx.apply_receipt(rcpt)
+
+    fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
+    (transfer_type, transfer_data) = fltr.parse_transferfrom(tx, eth_rpc)
+
+    assert transfer_type == 'transferfrom'
+
+
+def test_faucet_gift_to_tx(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        foo_token,
+        agent_roles,
+        contract_roles,
+        faucet,
+        account_registry,
+        celery_session_worker,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    gas_oracle = OverrideGasOracle(conn=rpc, limit=800000)
+
+    nonce_oracle = RPCNonceOracle(contract_roles['ACCOUNT_REGISTRY_WRITER'], rpc)
+    txf = AccountRegistry(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = txf.add(account_registry, contract_roles['ACCOUNT_REGISTRY_WRITER'], agent_roles['ALICE'])
+    r = rpc.do(o)
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+   
+    nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], rpc)
+    txf = MinterFaucet(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = txf.give_to(faucet, agent_roles['ALICE'], agent_roles['ALICE'])
+    r = rpc.do(o)
+
+    o = transaction(tx_hash_hex)
+    r = rpc.do(o)
+    tx_src = snake_and_camel(r)
+    tx = Tx(tx_src)
+
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+
+    rcpt = snake_and_camel(r)
+    tx.apply_receipt(rcpt)
+
+    fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
+    (transfer_type, transfer_data) = fltr.parse_giftto(tx, eth_rpc)
+
+    assert transfer_type == 'tokengift'
+    assert transfer_data['token_address'] == foo_token
+
+
+def test_callback_filter(
+        default_chain_spec,
+        init_database,
+        eth_rpc,
+        eth_signer,
+        foo_token,
+        token_roles,
+        agent_roles,
+        contract_roles,
+        register_lookups,
+        ):
+
+    rpc = RPCConnection.connect(default_chain_spec, 'default')
+    nonce_oracle = RPCNonceOracle(token_roles['FOO_TOKEN_OWNER'], rpc)
+    gas_oracle = OverrideGasOracle(conn=rpc, limit=200000)
+   
+    txf = ERC20(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
+    (tx_hash_hex, o) = txf.transfer(foo_token, token_roles['FOO_TOKEN_OWNER'], agent_roles['ALICE'], 1024)
+    r = rpc.do(o)
+    
+    o = transaction(tx_hash_hex)
+    r = rpc.do(o)
+    logg.debug(r)
+
+    mockblock_src = {
+        'hash': add_0x(os.urandom(32).hex()),
+        'number': '0x2a',
+        'transactions': [tx_hash_hex],
+        'timestamp': datetime.datetime.utcnow().timestamp(),
+            }
+    mockblock = Block(mockblock_src)
+
+    tx_src = snake_and_camel(r)
+    tx = Tx(tx_src, block=mockblock)
+
+    o = receipt(tx_hash_hex)
+    r = rpc.do(o)
+    assert r['status'] == 1
+
+    rcpt = snake_and_camel(r)
+    tx.apply_receipt(rcpt)
+
+    fltr = CallbackFilter(default_chain_spec, None, None, caller_address=contract_roles['CONTRACT_DEPLOYER'])
+
+    class CallbackMock:
+
+        def __init__(self):
+            self.results = {}
+
+        def call_back(self, transfer_type, result):
+            self.results[transfer_type] = result
+
+    mock = CallbackMock()
+    fltr.call_back = mock.call_back
+
+    fltr.filter(eth_rpc, mockblock, tx, init_database)
+
+    assert mock.results.get('transfer') != None
+    assert mock.results['transfer']['destination_token'] == foo_token

apps/cic-meta/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "cic-client-meta",
-	"version": "0.0.7-alpha.3",
+	"version": "0.0.7-alpha.6",
 	"description": "Signed CRDT metadata graphs for the CIC network",
 	"main": "dist/index.js",
 	"types": "dist/index.d.ts",
@@ -40,6 +40,6 @@
 	],
 	"license": "GPL-3.0-or-later",
 	"engines": {
-		"node": "~15.3.0"
+		"node": "~14.16.1"
 	}
 }

apps/cic-meta/scripts/server/server.ts

@@ -114,6 +114,7 @@ async function processRequest(req, res) {
 		return;
 	}
 
+
 	if (!['PUT', 'GET', 'POST'].includes(req.method)) {
 		res.writeHead(405, {"Content-Type": "text/plain"});
 		res.end();
@@ -123,6 +124,7 @@ async function processRequest(req, res) {
 	try {
 		digest = parseDigest(req.url);
 	} catch(e) {
+		console.error('digest error: ' + e)
 		res.writeHead(400, {"Content-Type": "text/plain"});
 		res.end();
 		return;

apps/cic-meta/src/assets/phone.ts

@@ -1,12 +1,12 @@
 import { ArgPair, Syncable } from '../sync';
-import { Addressable, addressToBytes, bytesToHex, toKey } from '../digest';
+import { Addressable, mergeKey } from '../digest';
 
 class Phone extends Syncable implements Addressable {
 
 	address:	string
 	value:		number
 
-	constructor(address:string, v:number) {
+	constructor(address:string, v:string) {
 		const o = {
 			msisdn: v,
 		}
@@ -17,8 +17,8 @@ class Phone extends Syncable implements Addressable {
 		});
 	}
 
-	public static async toKey(msisdn:number) {
-		return await toKey(msisdn.toString(), ':cic.msisdn');
+	public static async toKey(msisdn:string) {
+		return await mergeKey(Buffer.from(msisdn), Buffer.from(':cic.phone'));
 	}
 
 	public key(): string {

apps/cic-meta/src/digest.ts

@@ -61,6 +61,7 @@ function addressToBytes(s:string) {
 export {
 	toKey,
 	toAddressKey,
+	mergeKey,
 	bytesToHex,
 	addressToBytes,
 	Addressable,

apps/cic-notify/cic_notify/api.py

@@ -3,6 +3,7 @@ import logging
 import re
 
 # third-party imports
+from celery.app.control import Inspect
 import celery
 
 # local imports
@@ -15,6 +16,29 @@ logg = logging.getLogger()
 sms_tasks_matcher = r"^(cic_notify.tasks.sms)(\.\w+)?"
 
 
+re_q = r'^cic-notify'
+def get_sms_queue_tasks(app, task_prefix='cic_notify.tasks.sms.'):
+    host_queues = []
+
+    i = Inspect(app=app)
+    qs = i.active_queues()
+    for host in qs.keys():
+        for q in qs[host]:
+            if re.match(re_q, q['name']):
+                host_queues.append((host, q['name'],))
+   
+    task_prefix_len = len(task_prefix)
+    queue_tasks = []
+    for (host, queue) in host_queues:
+        i = Inspect(app=app, destination=[host])
+        for tasks in i.registered_tasks().values():
+            for task in tasks:
+                if len(task) >= task_prefix_len and task[:task_prefix_len] == task_prefix:
+                    queue_tasks.append((queue, task,))
+    
+    return queue_tasks
+
+
 class Api:
     # TODO: Implement callback strategy
     def __init__(self, queue='cic-notify'):
@@ -22,17 +46,9 @@ class Api:
         :param queue: The queue on which to execute notification tasks
         :type queue: str
         """
-        registered_tasks = app.tasks
-        self.sms_tasks = []
+        self.sms_tasks = get_sms_queue_tasks(app)
+        logg.debug('sms tasks {}'.format(self.sms_tasks))
 
-        for task in registered_tasks.keys():
-            logg.debug(f'Found: {task} {registered_tasks[task]}')
-            match = re.match(sms_tasks_matcher, task)
-            if match:
-                self.sms_tasks.append(task)
-
-        self.queue = queue
-        logg.info(f'api using queue: {self.queue}')
 
     def sms(self, message, recipient):
         """This function chains all sms tasks in order to send a message, log and persist said data to disk
@@ -44,12 +60,17 @@ class Api:
         :rtype: Celery.Task
         """
         signatures = []
-        for task in self.sms_tasks:
-            signature = celery.signature(task)
+        for q in self.sms_tasks:
+            signature = celery.signature(
+                    q[1],
+                    [
+                        message,
+                        recipient,
+                        ],
+                    queue=q[0],
+                    )
             signatures.append(signature)
-            signature_group = celery.group(signatures)
-            result = signature_group.apply_async(
-                args=[message, recipient],
-                queue=self.queue
-            )
-            return result
+
+        t = celery.group(signatures)()
+
+        return t

apps/cic-notify/cic_notify/runnable/send.py

@@ -0,0 +1,76 @@
+# standard imports
+import sys
+import os
+import logging
+import argparse
+import tempfile
+
+# external imports
+import celery
+import confini
+
+# local imports
+from cic_notify.api import Api
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+config_dir = os.path.join('/usr/local/etc/cic-notify')
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-c', type=str, default=config_dir, help='config file')
+argparser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
+argparser.add_argument('-v', action='store_true', help='be verbose')
+argparser.add_argument('-vv', action='store_true', help='be more verbose')
+argparser.add_argument('recipient', type=str, help='notification recipient')
+argparser.add_argument('message', type=str, help='message text')
+args = argparser.parse_args()
+
+if args.vv:
+    logging.getLogger().setLevel(logging.DEBUG)
+elif args.v:
+    logging.getLogger().setLevel(logging.INFO)
+
+config = confini.Config(args.c, args.env_prefix)
+config.process()
+config.censor('PASSWORD', 'DATABASE')
+config.add(args.recipient, '_RECIPIENT', True)
+config.add(args.message, '_MESSAGE', True)
+
+# set up celery
+app = celery.Celery(__name__)
+
+broker = config.get('CELERY_BROKER_URL')
+if broker[:4] == 'file':
+    bq = tempfile.mkdtemp()
+    bp = tempfile.mkdtemp()
+    app.conf.update({
+            'broker_url': broker,
+            'broker_transport_options': {
+                'data_folder_in': bq,
+                'data_folder_out': bq,
+                'data_folder_processed': bp,
+            },
+            },
+            )
+    logg.warning('celery broker dirs queue i/o {} processed {}, will NOT be deleted on shutdown'.format(bq, bp))
+else:
+    app.conf.update({
+        'broker_url': broker,
+        })
+
+result = config.get('CELERY_RESULT_URL')
+if result[:4] == 'file':
+    rq = tempfile.mkdtemp()
+    app.conf.update({
+        'result_backend': 'file://{}'.format(rq),
+        })
+    logg.warning('celery backend store dir {} created, will NOT be deleted on shutdown'.format(rq))
+else:
+    app.conf.update({
+        'result_backend': result,
+        })
+
+if __name__ == '__main__':
+    a = Api()
+    t = a.sms(config.get('_RECIPIENT'), config.get('_MESSAGE'))

apps/cic-notify/cic_notify/runnable/tasker.py

@@ -33,7 +33,9 @@ elif args.v:
 
 config = confini.Config(args.c, args.env_prefix)
 config.process()
+config.add(args.q, '_CELERY_QUEUE', True)
 config.censor('PASSWORD', 'DATABASE')
+logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 
 # connect to database
 dsn = dsn_from_config(config)

apps/cic-notify/cic_notify/version.py

@@ -6,11 +6,10 @@ import time
 import semver
 
 # local imports
-from cic_notify.error import PleaseCommitFirstError
 
 logg = logging.getLogger()
 
-version = (0, 4, 0, 'alpha.3')
+version = (0, 4, 0, 'alpha.4')
 
 version_object = semver.VersionInfo(
         major=version[0],
@@ -18,27 +17,4 @@ version_object = semver.VersionInfo(
         patch=version[2],
         prerelease=version[3],
         )
-
 version_string = str(version_object)
-
-
-def git_hash():
-    import subprocess
-
-    git_hash = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True)
-    git_hash_brief = git_hash.stdout.decode('utf-8')[:8]
-    return git_hash_brief
-
-
-try:
-    version_git = git_hash()
-    version_string += '+build.{}'.format(version_git)
-except FileNotFoundError:
-    time_string_pair = str(time.time()).split('.')
-    version_string += '+build.{}{:<09d}'.format(
-        time_string_pair[0],
-        int(time_string_pair[1]),
-    )
-logg.info(f'Final version string will be {version_string}')
-
-__version_string__ = version_string

apps/cic-notify/docker/Dockerfile

@@ -1,4 +1,4 @@
-FROM python:3.8.6
+FROM python:3.8.6-slim-buster
 
 RUN apt-get update && \
 	apt install -y gcc gnupg libpq-dev wget make g++ gnupg bash procps
@@ -6,7 +6,7 @@ RUN apt-get update && \
 WORKDIR /usr/src/cic-notify
 
 ARG pip_extra_index_url_flag='--index https://pypi.org/simple --extra-index-url https://pip.grassrootseconomics.net:8433'
-RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a44
+RUN pip install $pip_extra_index_url_flag cic-base[full_graph]==0.1.2a62
 
 COPY cic-notify/setup.cfg \
 	 cic-notify/setup.py \

apps/cic-notify/requirements.txt

@@ -1 +1 @@
-cic_base[full_graph]~=0.1.2a46
+cic_base[full_graph]~=0.1.2a61

apps/cic-notify/setup.cfg

@@ -1,6 +1,5 @@
 [metadata]
 name = cic-notify
-version= 0.4.0a2
 description = CIC notifications service
 author = Louis Holbrook
 author_email = dev@holbrook.no
@@ -45,3 +44,4 @@ testing =
 [options.entry_points]
 console_scripts =
 	cic-notify-tasker = cic_notify.runnable.tasker:main
+	cic-notify-send = cic_notify.runnable.send:main

apps/cic-notify/setup.py

@@ -1,9 +1,31 @@
 # standard imports
+import logging
+import subprocess
+import time
 from setuptools import setup
 
-# third-party imports
-
 # local imports
+from cic_notify.version import version_string
+
+logg = logging.getLogger()
+
+
+def git_hash():
+    git_hash = subprocess.run(['git', 'rev-parse', 'HEAD'], capture_output=True)
+    git_hash_brief = git_hash.stdout.decode('utf-8')[:8]
+    return git_hash_brief
+
+
+try:
+    version_git = git_hash()
+    version_string += '+build.{}'.format(version_git)
+except FileNotFoundError:
+    time_string_pair = str(time.time()).split('.')
+    version_string += '+build.{}{:<09d}'.format(
+            time_string_pair[0],
+            int(time_string_pair[1]),
+            )
+logg.info(f'Final version string will be {version_string}')
 
 
 requirements = []
@@ -25,6 +47,6 @@ while True:
 test_requirements_file.close()
 
 setup(
+    version=version_string,
     install_requires=requirements,
-    tests_require=test_requirements,
-)
+    tests_require=test_requirements)

apps/cic-ussd/.config/app.ini

@@ -1,14 +1,24 @@
 [app]
-ALLOWED_IP=127.0.0.1
+ALLOWED_IP=0.0.0.0/0
 LOCALE_FALLBACK=en
-LOCALE_PATH=var/lib/locale/
+LOCALE_PATH=/usr/src/cic-ussd/var/lib/locale/
 MAX_BODY_LENGTH=1024
 PASSWORD_PEPPER=QYbzKff6NhiQzY3ygl2BkiKOpER8RE/Upqs/5aZWW+I=
 SERVICE_CODE=*483*46#
 
+[phone_number]
+REGION=KE
+
 [ussd]
 MENU_FILE=/usr/src/data/ussd_menu.json
+user = 
+pass = 
 
 [statemachine]
 STATES=/usr/src/cic-ussd/states/
 TRANSITIONS=/usr/src/cic-ussd/transitions/
+
+[client]
+host =
+port = 
+ssl =

apps/cic-ussd/.config/database.ini

@@ -6,3 +6,5 @@ HOST=localhost
 PORT=5432
 ENGINE=postgresql
 DRIVER=psycopg2
+DEBUG=0
+POOL_SIZE=1

apps/cic-ussd/.config/redis.ini

@@ -3,7 +3,7 @@ BROKER_URL=redis://
 RESULT_URL=redis://
 
 [redis]
-HOSTNAME=localhost
+HOSTNAME=redis
 PASSWORD=
 PORT=6379
 DATABASE=0

apps/cic-ussd/cic_ussd/account.py

@@ -8,12 +8,12 @@ from cic_types.processor import generate_metadata_pointer
 
 # local imports
 from cic_ussd.chain import Chain
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.redis import get_cached_data
 
 
-def define_account_tx_metadata(user: User):
+def define_account_tx_metadata(user: Account):
     # get sender metadata
     identifier = blockchain_address_to_metadata_pointer(
         blockchain_address=user.blockchain_address
@@ -27,7 +27,7 @@ def define_account_tx_metadata(user: User):
     if account_metadata:
         account_metadata = json.loads(account_metadata)
         person = Person()
-        deserialized_person = person.deserialize(metadata=account_metadata)
+        deserialized_person = person.deserialize(person_data=account_metadata)
         given_name = deserialized_person.given_name
         family_name = deserialized_person.family_name
         phone_number = deserialized_person.tel
@@ -46,4 +46,4 @@ def retrieve_account_statement(blockchain_address: str):
         callback_task='cic_ussd.tasks.callback_handler.process_statement_callback',
         callback_param=blockchain_address
     )
-    result = cic_eth_api.list(address=blockchain_address, limit=9)
+    cic_eth_api.list(address=blockchain_address, limit=9)

apps/cic-ussd/cic_ussd/db/migrations/default/versions/f289e8510444_.py

@@ -1,4 +1,4 @@
-"""Create user table
+"""Create account table
 
 Revision ID: f289e8510444
 Revises: 
@@ -17,7 +17,7 @@ depends_on = None
 
 
 def upgrade():
-    op.create_table('user',
+    op.create_table('account',
                     sa.Column('id', sa.Integer(), nullable=False),
                     sa.Column('blockchain_address', sa.String(), nullable=False),
                     sa.Column('phone_number', sa.String(), nullable=False),
@@ -29,11 +29,11 @@ def upgrade():
                     sa.Column('updated', sa.DateTime(), nullable=False),
                     sa.PrimaryKeyConstraint('id')
                     )
-    op.create_index(op.f('ix_user_phone_number'), 'user', ['phone_number'], unique=True)
-    op.create_index(op.f('ix_user_blockchain_address'), 'user', ['blockchain_address'], unique=True)
+    op.create_index(op.f('ix_account_phone_number'), 'account', ['phone_number'], unique=True)
+    op.create_index(op.f('ix_account_blockchain_address'), 'account', ['blockchain_address'], unique=True)
 
 
 def downgrade():
-    op.drop_index(op.f('ix_user_blockchain_address'), table_name='user')
-    op.drop_index(op.f('ix_user_phone_number'), table_name='user')
-    op.drop_table('user')
+    op.drop_index(op.f('ix_account_blockchain_address'), table_name='account')
+    op.drop_index(op.f('ix_account_phone_number'), table_name='account')
+    op.drop_table('account')

apps/cic-ussd/cic_ussd/db/models/account.py

@@ -16,12 +16,12 @@ class AccountStatus(IntEnum):
     RESET = 4
 
 
-class User(SessionBase):
+class Account(SessionBase):
     """
     This class defines a user record along with functions responsible for hashing the user's corresponding password and
      subsequently verifying a password's validity given an input to compare against the persisted hash.
     """
-    __tablename__ = 'user'
+    __tablename__ = 'account'
 
     blockchain_address = Column(String)
     phone_number = Column(String)
@@ -38,7 +38,7 @@ class User(SessionBase):
         self.account_status = AccountStatus.PENDING.value
 
     def __repr__(self):
-        return f'<User: {self.blockchain_address}>'
+        return f'<Account: {self.blockchain_address}>'
 
     def create_password(self, password):
         """This method takes a password value and hashes the value before assigning it to the corresponding

apps/cic-ussd/cic_ussd/db/models/base.py

@@ -1,47 +1,129 @@
-# standard imports
+# stanard imports
+import logging
 import datetime
 
-# third-party imports
+# external imports
 from sqlalchemy import Column, Integer, DateTime
 from sqlalchemy.ext.declarative import declarative_base
 from sqlalchemy import create_engine
 from sqlalchemy.orm import sessionmaker
+from sqlalchemy.pool import (
+        StaticPool,
+        QueuePool,
+        AssertionPool,
+        NullPool,
+        )
+
+logg = logging.getLogger().getChild(__name__)
 
 Model = declarative_base(name='Model')
 
 
 class SessionBase(Model):
+    """The base object for all SQLAlchemy enabled models. All other models must extend this.
+    """
     __abstract__ = True
 
-    id = Column(Integer, primary_key=True)
     created = Column(DateTime, default=datetime.datetime.utcnow)
     updated = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
 
+    id = Column(Integer, primary_key=True)
+
     engine = None
-    session = None
-    query = None
+    """Database connection engine of the running aplication"""
+    sessionmaker = None
+    """Factory object responsible for creating sessions from the  connection pool"""
+    transactional = True
+    """Whether the database backend supports query transactions. Should be explicitly set by initialization code"""
+    poolable = True
+    """Whether the database backend supports connection pools. Should be explicitly set by initialization code"""
+    procedural = True
+    """Whether the database backend supports stored procedures"""
+    localsessions = {}
+    """Contains dictionary of sessions initiated by db model components"""
+
 
     @staticmethod
     def create_session():
-        session = sessionmaker(bind=SessionBase.engine)
-        return session()
+        """Creates a new database session.
+        """
+        return SessionBase.sessionmaker()
+
 
     @staticmethod
     def _set_engine(engine):
+        """Sets the database engine static property
+        """
         SessionBase.engine = engine
+        SessionBase.sessionmaker = sessionmaker(bind=SessionBase.engine)
+
 
     @staticmethod
-    def build():
-        Model.metadata.create_all(bind=SessionBase.engine)
+    def connect(dsn, pool_size=16, debug=False):
+        """Create new database connection engine and connect to database backend.
+
+        :param dsn: DSN string defining connection.
+        :type dsn: str
+        """
+        e = None
+        if SessionBase.poolable:
+            poolclass = QueuePool
+            if pool_size > 1:
+                logg.info('db using queue pool')
+                e = create_engine(
+                        dsn,
+                        max_overflow=pool_size*3,
+                        pool_pre_ping=True,
+                        pool_size=pool_size,
+                        pool_recycle=60,
+                        poolclass=poolclass,
+                        echo=debug,
+                    )
+            else:
+                if pool_size == 0:
+                    poolclass = NullPool
+                elif debug:
+                    poolclass = AssertionPool
+                else:
+                    poolclass = StaticPool
+                e = create_engine(
+                        dsn,
+                        poolclass=poolclass,
+                        echo=debug,
+                    )
+        else:
+            logg.info('db connection not poolable')
+            e = create_engine(
+                    dsn,
+                    echo=debug,
+                    )
+
+        SessionBase._set_engine(e)
 
-    @staticmethod
-    # https://docs.sqlalchemy.org/en/13/core/pooling.html#pool-disconnects
-    def connect(data_source_name):
-        engine = create_engine(data_source_name, pool_pre_ping=True)
-        SessionBase._set_engine(engine)
 
     @staticmethod
     def disconnect():
+        """Disconnect from database and free resources.
+        """
         SessionBase.engine.dispose()
         SessionBase.engine = None
 
+
+    @staticmethod
+    def bind_session(session=None):
+        localsession = session
+        if localsession == None:
+            localsession = SessionBase.create_session()
+            localsession_key = str(id(localsession))
+            logg.debug('creating new session {}'.format(localsession_key))
+            SessionBase.localsessions[localsession_key] = localsession
+        return localsession
+
+
+    @staticmethod
+    def release_session(session=None):
+        session_key = str(id(session))
+        if SessionBase.localsessions.get(session_key) != None:
+            logg.debug('commit and destroy session {}'.format(session_key))
+            session.commit()
+            session.close()

apps/cic-ussd/cic_ussd/db/ussd_menu.json

@@ -128,8 +128,8 @@
         },
         "22": {
             "description": "Pin entry menu.",
-            "display_key": "ussd.kenya.standard_pin_authorization",
-            "name": "standard_pin_authorization",
+            "display_key": "ussd.kenya.display_metadata_pin_authorization",
+            "name": "display_metadata_pin_authorization",
             "parent": "start"
         },
         "23": {
@@ -230,9 +230,22 @@
         },
         "39": {
             "description": "Menu to instruct users to call the office.",
-            "display_key": "ussd.key.help",
+            "display_key": "ussd.kenya.help",
             "name": "help",
             "parent": null
+        },
+        "40": {
+            "description": "Menu to display a user's entire profile",
+            "display_key": "ussd.kenya.display_user_metadata",
+            "name": "display_user_metadata",
+            "parent": "account_management"
+        },
+        "41": {
+            "description": "The recipient is not in the system",
+            "display_key": "ussd.kenya.exit_invalid_recipient",
+            "name": "exit_invalid_recipient",
+            "parent": null
         }
+
     }
 }

apps/cic-ussd/cic_ussd/error.py

@@ -18,7 +18,7 @@ class ActionDataNotFoundError(OSError):
     pass
 
 
-class UserMetadataNotFoundError(OSError):
+class MetadataNotFoundError(OSError):
     """Raised when metadata is expected but not available in cache."""
     pass
 
@@ -31,3 +31,10 @@ class UnsupportedMethodError(OSError):
 class CachedDataNotFoundError(OSError):
     """Raised when the method passed to the make request function is unsupported."""
     pass
+
+
+class MetadataStoreError(Exception):
+    """Raised when metadata storage fails"""
+    pass
+
+

apps/cic-ussd/cic_ussd/metadata/__init__.py

@@ -3,7 +3,10 @@
 # third-party imports
 import requests
 from chainlib.eth.address import to_checksum
-from hexathon import add_0x
+from hexathon import (
+        add_0x,
+        strip_0x,
+        )
 
 # local imports
 from cic_ussd.error import UnsupportedMethodError
@@ -40,4 +43,4 @@ def blockchain_address_to_metadata_pointer(blockchain_address: str):
     :return:
     :rtype:
     """
-    return bytes.fromhex(blockchain_address[2:])
+    return bytes.fromhex(strip_0x(blockchain_address))

apps/cic-ussd/cic_ussd/metadata/base.py

@@ -0,0 +1,126 @@
+# standard imports
+import json
+import logging
+import os
+from typing import Dict, Union
+
+# third-part imports
+import requests
+from cic_types.models.person import generate_metadata_pointer, Person
+
+# local imports
+from cic_ussd.metadata import make_request
+from cic_ussd.metadata.signer import Signer
+from cic_ussd.redis import cache_data
+from cic_ussd.error import MetadataStoreError
+
+
+logg = logging.getLogger().getChild(__name__)
+
+
+class Metadata:
+    """
+    :cvar base_url: The base url or the metadata server.
+    :type base_url: str
+    """
+
+    base_url = None
+
+
+def metadata_http_error_handler(result: requests.Response):
+    """ This function handles and appropriately raises errors from http requests interacting with the metadata server.
+    :param result: The response object from a http request.
+    :type result: requests.Response
+    """
+    status_code = result.status_code
+
+    if 100 <= status_code < 200:
+        raise MetadataStoreError(f'Informational errors: {status_code}, reason: {result.reason}')
+
+    elif 300 <= status_code < 400:
+        raise MetadataStoreError(f'Redirect Issues: {status_code}, reason: {result.reason}')
+
+    elif 400 <= status_code < 500:
+        raise MetadataStoreError(f'Client Error: {status_code}, reason: {result.reason}')
+
+    elif 500 <= status_code < 600:
+        raise MetadataStoreError(f'Server Error: {status_code}, reason: {result.reason}')
+
+
+class MetadataRequestsHandler(Metadata):
+
+    def __init__(self, cic_type: str, identifier: bytes, engine: str = 'pgp'):
+        """
+        :param cic_type: The salt value with which to hash a specific metadata identifier.
+        :type cic_type: str
+        :param engine: Encryption used for sending data to the metadata server.
+        :type engine: str
+        :param identifier: A unique element of data in bytes necessary for creating a metadata pointer.
+        :type identifier: bytes
+        """
+        self.cic_type = cic_type
+        self.engine = engine
+        self.headers = {
+            'X-CIC-AUTOMERGE': 'server',
+            'Content-Type': 'application/json'
+        }
+        self.identifier = identifier
+        self.metadata_pointer = generate_metadata_pointer(
+            identifier=self.identifier,
+            cic_type=self.cic_type
+        )
+        if self.base_url:
+            self.url = os.path.join(self.base_url, self.metadata_pointer)
+
+    def create(self, data: Union[Dict, str]):
+        """ This function is responsible for posting data to the metadata server with a corresponding metadata pointer
+        for storage.
+        :param data: The data to be stored in the metadata server.
+        :type data: dict|str
+        """
+        data = json.dumps(data).encode('utf-8')
+        result = make_request(method='POST', url=self.url, data=data, headers=self.headers)
+        metadata_http_error_handler(result=result)
+        metadata = result.content
+        self.edit(data=metadata)
+
+    def edit(self, data: bytes):
+        """ This function is responsible for editing data in the metadata server corresponding to a unique pointer.
+        :param data: The data to be edited in the metadata server.
+        :type data: bytes
+        """
+        cic_meta_signer = Signer()
+        signature = cic_meta_signer.sign_digest(data=data)
+        algorithm = cic_meta_signer.get_operational_key().get('algo')
+        decoded_data = data.decode('utf-8')
+        formatted_data = {
+            'm': data.decode('utf-8'),
+            's': {
+                'engine': self.engine,
+                'algo': algorithm,
+                'data': signature,
+                'digest': json.loads(data).get('digest'),
+            }
+        }
+        formatted_data = json.dumps(formatted_data).encode('utf-8')
+        result = make_request(method='PUT', url=self.url, data=formatted_data, headers=self.headers)
+        logg.info(f'signed metadata submission status: {result.status_code}.')
+        metadata_http_error_handler(result=result)
+        try:
+            decoded_identifier = self.identifier.decode("utf-8")
+        except UnicodeDecodeError:
+            decoded_identifier = self.identifier.hex()
+        logg.info(f'identifier: {decoded_identifier}. metadata pointer: {self.metadata_pointer} set to: {decoded_data}.')
+
+    def query(self):
+        """This function is responsible for querying the metadata server for data corresponding to a unique pointer."""
+        result = make_request(method='GET', url=self.url)
+        metadata_http_error_handler(result=result)
+        response_data = result.content
+        data = json.loads(response_data.decode('utf-8'))
+        if result.status_code == 200 and self.cic_type == 'cic.person':
+            person = Person()
+            deserialized_person = person.deserialize(person_data=json.loads(data))
+            data = json.dumps(deserialized_person.serialize())
+            cache_data(self.metadata_pointer, data=data)
+            logg.debug(f'caching: {data} with key: {self.metadata_pointer}')

apps/cic-ussd/cic_ussd/metadata/person.py

@@ -0,0 +1,12 @@
+# standard imports
+
+# third-party imports
+
+# local imports
+from .base import MetadataRequestsHandler
+
+
+class PersonMetadata(MetadataRequestsHandler):
+
+    def __init__(self, identifier: bytes):
+        super().__init__(cic_type='cic.person', identifier=identifier)

apps/cic-ussd/cic_ussd/metadata/phone.py

@@ -0,0 +1,13 @@
+# standard imports
+import logging
+
+# external imports
+
+# local imports
+from .base import MetadataRequestsHandler
+
+
+class PhonePointerMetadata(MetadataRequestsHandler):
+
+    def __init__(self, identifier: bytes):
+        super().__init__(cic_type='cic.msisdn', identifier=identifier)

apps/cic-ussd/cic_ussd/metadata/signer.py

@@ -44,7 +44,7 @@ class Signer:
         gpg_keys = self.gpg.list_keys()
         key_algorithm = gpg_keys[0].get('algo')
         key_id = gpg_keys[0].get("keyid")
-        logg.info(f'using signing key: {key_id}, algorithm: {key_algorithm}')
+        logg.debug(f'using signing key: {key_id}, algorithm: {key_algorithm}')
         return gpg_keys[0]
 
     def sign_digest(self, data: bytes):

apps/cic-ussd/cic_ussd/metadata/user.py

@@ -1,102 +0,0 @@
-# standard imports
-import json
-import logging
-import os
-
-# third-party imports
-import requests
-from cic_types.models.person import generate_metadata_pointer, Person
-
-# local imports
-from cic_ussd.chain import Chain
-from cic_ussd.metadata import make_request
-from cic_ussd.metadata.signer import Signer
-from cic_ussd.redis import cache_data
-
-logg = logging.getLogger()
-
-
-class UserMetadata:
-    """
-    :cvar base_url:
-    :type base_url:
-    """
-    base_url = None
-
-    def __init__(self, identifier: bytes):
-        """
-        :param identifier:
-        :type identifier:
-        """
-        self. headers = {
-            'X-CIC-AUTOMERGE': 'server',
-            'Content-Type': 'application/json'
-        }
-        self.identifier = identifier
-        self.metadata_pointer = generate_metadata_pointer(
-                identifier=self.identifier,
-                cic_type='cic.person'
-        )
-        if self.base_url:
-            self.url = os.path.join(self.base_url, self.metadata_pointer)
-
-    def create(self, data: dict):
-        try:
-            data = json.dumps(data).encode('utf-8')
-            result = make_request(method='POST', url=self.url, data=data, headers=self.headers)
-            metadata = result.content
-            self.edit(data=metadata, engine='pgp')
-            logg.info(f'Get sign material response status: {result.status_code}')
-            result.raise_for_status()
-        except requests.exceptions.HTTPError as error:
-            raise RuntimeError(error)
-
-    def edit(self, data: bytes, engine: str):
-        """
-        :param data:
-        :type data:
-        :param engine:
-        :type engine:
-        :return:
-        :rtype:
-        """
-        cic_meta_signer = Signer()
-        signature = cic_meta_signer.sign_digest(data=data)
-        algorithm = cic_meta_signer.get_operational_key().get('algo')
-        formatted_data = {
-            'm': data.decode('utf-8'),
-            's': {
-                'engine': engine,
-                'algo': algorithm,
-                'data': signature,
-                'digest': json.loads(data).get('digest'),
-            }
-        }
-        formatted_data = json.dumps(formatted_data).encode('utf-8')
-
-        try:
-            result = make_request(method='PUT', url=self.url, data=formatted_data, headers=self.headers)
-            logg.info(f'Signed content submission status: {result.status_code}.')
-            result.raise_for_status()
-        except requests.exceptions.HTTPError as error:
-            raise RuntimeError(error)
-
-    def query(self):
-        result = make_request(method='GET', url=self.url)
-        status = result.status_code
-        logg.info(f'Get latest data status: {status}')
-        try:
-            if status == 200:
-                response_data = result.content
-                data = json.loads(response_data.decode())
-
-                # validate data
-                person = Person()
-                deserialized_person = person.deserialize(metadata=json.loads(data))
-
-                cache_data(key=self.metadata_pointer, data=json.dumps(deserialized_person.serialize()))
-            elif status == 404:
-                logg.info('The data is not available and might need to be added.')
-            result.raise_for_status()
-        except requests.exceptions.HTTPError as error:
-            raise RuntimeError(error)

apps/cic-ussd/cic_ussd/operations.py

@@ -10,7 +10,7 @@ from tinydb.table import Document
 from typing import Optional
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 from cic_ussd.db.models.ussd_session import UssdSession
 from cic_ussd.db.models.task_tracker import TaskTracker
 from cic_ussd.menu.ussd_menu import UssdMenu
@@ -143,10 +143,10 @@ def get_account_status(phone_number) -> str:
     :return: The user account status.
     :rtype: str
     """
-    user = User.session.query(User).filter_by(phone_number=phone_number).first()
+    user = Account.session.query(Account).filter_by(phone_number=phone_number).first()
     status = user.get_account_status()
-    User.session.add(user)
-    User.session.commit()
+    Account.session.add(user)
+    Account.session.commit()
 
     return status
 
@@ -269,12 +269,12 @@ def cache_account_creation_task_id(phone_number: str, task_id: str):
     redis_cache.persist(name=task_id)
 
 
-def process_current_menu(ussd_session: Optional[dict], user: User, user_input: str) -> Document:
+def process_current_menu(ussd_session: Optional[dict], user: Account, user_input: str) -> Document:
     """This function checks user input and returns a corresponding ussd menu
     :param ussd_session: An in db ussd session object.
     :type ussd_session: UssdSession
     :param user: A user object.
-    :type user: User
+    :type user: Account
     :param user_input: The user's input.
     :type user_input: str
     :return: An in memory ussd menu object.
@@ -324,7 +324,7 @@ def process_menu_interaction_requests(chain_str: str,
 
     else:
         # get user
-        user = User.session.query(User).filter_by(phone_number=phone_number).first()
+        user = Account.session.query(Account).filter_by(phone_number=phone_number).first()
 
         # find any existing ussd session
         existing_ussd_session = UssdSession.session.query(UssdSession).filter_by(
@@ -390,10 +390,10 @@ def reset_pin(phone_number: str) -> str:
     :return: The status of the pin reset.
     :rtype: str
     """
-    user = User.session.query(User).filter_by(phone_number=phone_number).first()
+    user = Account.session.query(Account).filter_by(phone_number=phone_number).first()
     user.reset_account_pin()
-    User.session.add(user)
-    User.session.commit()
+    Account.session.add(user)
+    Account.session.commit()
 
     response = f'Pin reset for user {phone_number} is successful!'
     return response

apps/cic-ussd/cic_ussd/phone_number.py

@@ -5,7 +5,7 @@ from typing import Optional
 import phonenumbers
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 
 
 def process_phone_number(phone_number: str, region: str):
@@ -30,14 +30,14 @@ def process_phone_number(phone_number: str, region: str):
     return parsed_phone_number
 
 
-def get_user_by_phone_number(phone_number: str) -> Optional[User]:
+def get_user_by_phone_number(phone_number: str) -> Optional[Account]:
     """This function queries the database for a user based on the provided phone number.
     :param phone_number: A valid phone number.
     :type phone_number: str
     :return: A user object matching a given phone number
-    :rtype: User|None
+    :rtype: Account|None
     """
     # consider adding region to user's metadata
     phone_number = process_phone_number(phone_number=phone_number, region='KE')
-    user = User.session.query(User).filter_by(phone_number=phone_number).first()
+    user = Account.session.query(Account).filter_by(phone_number=phone_number).first()
     return user

apps/cic-ussd/cic_ussd/processor.py

@@ -13,8 +13,9 @@ from tinydb.table import Document
 from cic_ussd.account import define_account_tx_metadata, retrieve_account_statement
 from cic_ussd.balance import BalanceManager, compute_operational_balance, get_cached_operational_balance
 from cic_ussd.chain import Chain
-from cic_ussd.db.models.user import AccountStatus, User
+from cic_ussd.db.models.account import AccountStatus, Account
 from cic_ussd.db.models.ussd_session import UssdSession
+from cic_ussd.error import MetadataNotFoundError
 from cic_ussd.menu.ussd_menu import UssdMenu
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.phone_number import get_user_by_phone_number
@@ -22,17 +23,18 @@ from cic_ussd.redis import cache_data, create_cached_data_key, get_cached_data
 from cic_ussd.state_machine import UssdStateMachine
 from cic_ussd.conversions import to_wei, from_wei
 from cic_ussd.translation import translation_for
+from cic_types.models.person import generate_metadata_pointer, get_contact_data_from_vcard
 
 logg = logging.getLogger(__name__)
 
 
-def process_pin_authorization(display_key: str, user: User, **kwargs) -> str:
+def process_pin_authorization(display_key: str, user: Account, **kwargs) -> str:
     """
     This method provides translation for all ussd menu entries that follow the pin authorization pattern.
     :param display_key: The path in the translation files defining an appropriate ussd response
     :type display_key: str
     :param user: The user in a running USSD session.
-    :type user: User
+    :type user: Account
     :param kwargs: Any additional information required by the text values in the internationalization files.
     :type kwargs
     :return: A string value corresponding the ussd menu's text value.
@@ -53,13 +55,13 @@ def process_pin_authorization(display_key: str, user: User, **kwargs) -> str:
         )
 
 
-def process_exit_insufficient_balance(display_key: str, user: User, ussd_session: dict):
+def process_exit_insufficient_balance(display_key: str, user: Account, ussd_session: dict):
     """This function processes the exit menu letting users their account balance is insufficient to perform a specific
     transaction.
     :param display_key: The path in the translation files defining an appropriate ussd response
     :type display_key: str
     :param user: The user requesting access to the ussd menu.
-    :type user: User
+    :type user: Account
     :param ussd_session: A JSON serialized in-memory ussd session object
     :type ussd_session: dict
     :return: Corresponding translation text response
@@ -88,12 +90,12 @@ def process_exit_insufficient_balance(display_key: str, user: User, ussd_session
     )
 
 
-def process_exit_successful_transaction(display_key: str, user: User, ussd_session: dict):
+def process_exit_successful_transaction(display_key: str, user: Account, ussd_session: dict):
     """This function processes the exit menu after a successful initiation for a transfer of tokens.
     :param display_key: The path in the translation files defining an appropriate ussd response
     :type display_key: str
     :param user: The user requesting access to the ussd menu.
-    :type user: User
+    :type user: Account
     :param ussd_session: A JSON serialized in-memory ussd session object
     :type ussd_session: dict
     :return: Corresponding translation text response
@@ -116,11 +118,11 @@ def process_exit_successful_transaction(display_key: str, user: User, ussd_sessi
     )
 
 
-def process_transaction_pin_authorization(user: User, display_key: str, ussd_session: dict):
+def process_transaction_pin_authorization(user: Account, display_key: str, ussd_session: dict):
     """This function processes pin authorization where making a transaction is concerned. It constructs a
     pre-transaction response menu that shows the details of the transaction.
     :param user: The user requesting access to the ussd menu.
-    :type user: User
+    :type user: Account
     :param display_key: The path in the translation files defining an appropriate ussd response
     :type display_key: str
     :param ussd_session: The USSD session determining what user data needs to be extracted and added to the menu's
@@ -136,7 +138,7 @@ def process_transaction_pin_authorization(user: User, display_key: str, ussd_ses
     tx_sender_information = define_account_tx_metadata(user=user)
 
     token_symbol = 'SRF'
-    user_input = ussd_session.get('user_input').split('*')[-1]
+    user_input = ussd_session.get('session_data').get('transaction_amount')
     transaction_amount = to_wei(value=int(user_input))
     logg.debug('Requires integration to determine user tokens.')
     return process_pin_authorization(
@@ -149,7 +151,7 @@ def process_transaction_pin_authorization(user: User, display_key: str, ussd_ses
     )
 
 
-def process_account_balances(user: User, display_key: str, ussd_session: dict):
+def process_account_balances(user: Account, display_key: str, ussd_session: dict):
     """
     :param user:
     :type user:
@@ -187,22 +189,56 @@ def format_transactions(transactions: list, preferred_language: str):
             value = transaction.get('to_value')
             timestamp = transaction.get('timestamp')
             action_tag = transaction.get('action_tag')
+            direction = transaction.get('direction')
             token_symbol = 'SRF'
 
             if action_tag == 'SENT' or action_tag == 'ULITUMA':
-                formatted_transactions += f'{action_tag} {value} {token_symbol} {recipient_phone_number} {timestamp}.\n'
+                formatted_transactions += f'{action_tag} {value} {token_symbol} {direction} {recipient_phone_number} {timestamp}.\n'
             else:
-                formatted_transactions += f'{action_tag} {value} {token_symbol} {sender_phone_number} {timestamp}. \n'
+                formatted_transactions += f'{action_tag} {value} {token_symbol} {direction} {sender_phone_number} {timestamp}. \n'
         return formatted_transactions
     else:
         if preferred_language == 'en':
-            formatted_transactions = 'Empty'
+            formatted_transactions = 'NO TRANSACTION HISTORY'
         else:
-            formatted_transactions = 'Hamna historia'
+            formatted_transactions = 'HAMNA RIPOTI YA MATUMIZI'
         return formatted_transactions
 
 
-def process_account_statement(user: User, display_key: str, ussd_session: dict):
+def process_display_user_metadata(user: Account, display_key: str):
+    """
+    :param user:
+    :type user:
+    :param display_key:
+    :type display_key:
+    """
+    key = generate_metadata_pointer(
+        identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
+        cic_type='cic.person'
+    )
+    user_metadata = get_cached_data(key)
+    if user_metadata:
+        user_metadata = json.loads(user_metadata)
+        contact_data = get_contact_data_from_vcard(vcard=user_metadata.get('vcard'))
+        logg.debug(f'{contact_data}')
+        full_name = f'{contact_data.get("given")} {contact_data.get("family")}'
+        gender = user_metadata.get('gender')
+        products = ', '.join(user_metadata.get('products'))
+        location = user_metadata.get('location').get('area_name')
+
+        return translation_for(
+            key=display_key,
+            preferred_language=user.preferred_language,
+            full_name=full_name,
+            gender=gender,
+            location=location,
+            products=products
+        )
+    else:
+        raise MetadataNotFoundError(f'Expected person metadata but found none in cache for key: {key}')
+
+
+def process_account_statement(user: Account, display_key: str, ussd_session: dict):
     """
     :param user:
     :type user:
@@ -229,7 +265,7 @@ def process_account_statement(user: User, display_key: str, ussd_session: dict):
         middle_transaction_set += transactions[3:][:3]
         first_transaction_set += transactions[:3]
     # there are probably much cleaner and operational inexpensive ways to do this so find them
-    elif 4 < len(transactions) < 7:
+    elif 3 < len(transactions) < 7:
         middle_transaction_set += transactions[3:]
         first_transaction_set += transactions[:3]
     else:
@@ -265,12 +301,12 @@ def process_account_statement(user: User, display_key: str, ussd_session: dict):
         )
 
 
-def process_start_menu(display_key: str, user: User):
+def process_start_menu(display_key: str, user: Account):
     """This function gets data on an account's balance and token in order to append it to the start of the start menu's
     title. It passes said arguments to the translation function and returns the appropriate corresponding text from the
     translation files.
     :param user: The user requesting access to the ussd menu.
-    :type user: User
+    :type user: Account
     :param display_key: The path in the translation files defining an appropriate ussd response
     :type display_key: str
     :return: Corresponding translation text response
@@ -295,11 +331,11 @@ def process_start_menu(display_key: str, user: User):
     operational_balance = compute_operational_balance(balances=balances_data)
 
     # retrieve and cache account's metadata
-    s_query_user_metadata = celery.signature(
-        'cic_ussd.tasks.metadata.query_user_metadata',
+    s_query_person_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.query_person_metadata',
         [blockchain_address]
     )
-    s_query_user_metadata.apply_async(queue='cic-ussd')
+    s_query_person_metadata.apply_async(queue='cic-ussd')
 
     # retrieve and cache account's statement
     retrieve_account_statement(blockchain_address=blockchain_address)
@@ -325,13 +361,13 @@ def retrieve_most_recent_ussd_session(phone_number: str) -> UssdSession:
     return last_ussd_session
 
 
-def process_request(user_input: str, user: User, ussd_session: Optional[dict] = None) -> Document:
+def process_request(user_input: str, user: Account, ussd_session: Optional[dict] = None) -> Document:
     """This function assesses a request based on the user from the request comes, the session_id and the user's
     input. It determines whether the request translates to a return to an existing session by checking whether the
     provided  session id exists in the database or whether the creation of a new ussd session object is warranted.
     It then returns the appropriate ussd menu text values.
     :param user: The user requesting access to the ussd menu.
-    :type user: User
+    :type user: Account
     :param user_input: The value a user enters in the ussd menu.
     :type user_input: str
     :param ussd_session: A JSON serialized in-memory ussd session object
@@ -349,18 +385,24 @@ def process_request(user_input: str, user: User, ussd_session: Optional[dict] =
         if user.has_valid_pin():
             last_ussd_session = retrieve_most_recent_ussd_session(phone_number=user.phone_number)
 
-            key = create_cached_data_key(
+            key = generate_metadata_pointer(
                 identifier=blockchain_address_to_metadata_pointer(blockchain_address=user.blockchain_address),
-                salt='cic.person'
+                cic_type='cic.person'
             )
-            user_metadata = get_cached_data(key=key)
+            person_metadata = get_cached_data(key=key)
 
             if last_ussd_session:
                 # get last state
                 last_state = last_ussd_session.state
-                logg.debug(f'LAST USSD SESSION STATE: {last_state}')
                 # if last state is account_creation_prompt and metadata exists, show start menu
-                if last_state == 'account_creation_prompt' and user_metadata is not None:
+                if last_state in [
+                    'account_creation_prompt',
+                    'exit',
+                    'exit_invalid_pin',
+                    'exit_invalid_new_pin',
+                    'exit_pin_mismatch',
+                    'exit_invalid_request'
+                ] and person_metadata is not None:
                     return UssdMenu.find_by_name(name='start')
                 else:
                     return UssdMenu.find_by_name(name=last_state)
@@ -373,14 +415,14 @@ def process_request(user_input: str, user: User, ussd_session: Optional[dict] =
                 return UssdMenu.find_by_name(name='initial_pin_entry')
 
 
-def next_state(ussd_session: dict, user: User, user_input: str) -> str:
+def next_state(ussd_session: dict, user: Account, user_input: str) -> str:
     """This function navigates the state machine based on the ussd session object and user inputs it receives.
     It checks the user input and provides the successive state in the state machine. It then updates the session's
     state attribute with the new state.
     :param ussd_session: A JSON serialized in-memory ussd session object
     :type ussd_session: dict
     :param user: The user requesting access to the ussd menu.
-    :type user: User
+    :type user: Account
     :param user_input: The value a user enters in the ussd menu.
     :type user_input: str
     :return: A string value corresponding the successive give a specific state in the state machine.
@@ -396,7 +438,7 @@ def custom_display_text(
         display_key: str,
         menu_name: str,
         ussd_session: dict,
-        user: User) -> str:
+        user: Account) -> str:
     """This function extracts the appropriate session data based on the current menu name. It then inserts them as
     keywords in the i18n function.
     :param display_key: The path in the translation files defining an appropriate ussd response
@@ -404,7 +446,7 @@ def custom_display_text(
     :param menu_name: The name by which a specific menu can be identified.
     :type menu_name: str
     :param user: The user in a running USSD session.
-    :type user: User
+    :type user: Account
     :param ussd_session: A JSON serialized in-memory ussd session object
     :type ussd_session: dict
     :return: A string value corresponding the ussd menu's text value.
@@ -420,9 +462,13 @@ def custom_display_text(
         return process_start_menu(display_key=display_key, user=user)
     elif 'pin_authorization' in menu_name:
         return process_pin_authorization(display_key=display_key, user=user)
+    elif 'enter_current_pin' in menu_name:
+        return process_pin_authorization(display_key=display_key, user=user)
     elif menu_name == 'account_balances':
         return process_account_balances(display_key=display_key, user=user, ussd_session=ussd_session)
     elif 'transaction_set' in menu_name:
         return process_account_statement(display_key=display_key, user=user, ussd_session=ussd_session)
+    elif menu_name == 'display_user_metadata':
+        return process_display_user_metadata(display_key=display_key, user=user)
     else:
         return translation_for(key=display_key, preferred_language=user.preferred_language)

apps/cic-ussd/cic_ussd/requests.py

@@ -10,7 +10,7 @@ from urllib.parse import urlparse, parse_qs
 from sqlalchemy import desc
 
 # local imports
-from cic_ussd.db.models.user import AccountStatus, User
+from cic_ussd.db.models.account import AccountStatus, Account
 from cic_ussd.operations import get_account_status, reset_pin
 from cic_ussd.validator import check_known_user
 
@@ -123,9 +123,9 @@ def process_locked_accounts_requests(env: dict) -> tuple:
             else:
                 limit = r[1]
 
-        locked_accounts = User.session.query(User.blockchain_address).filter(
-            User.account_status == AccountStatus.LOCKED.value,
-            User.failed_pin_attempts >= 3).order_by(desc(User.updated)).offset(offset).limit(limit).all()
+        locked_accounts = Account.session.query(Account.blockchain_address).filter(
+            Account.account_status == AccountStatus.LOCKED.value,
+            Account.failed_pin_attempts >= 3).order_by(desc(Account.updated)).offset(offset).limit(limit).all()
 
         # convert lists to scalar blockchain addresses
         locked_accounts = [blockchain_address for (blockchain_address, ) in locked_accounts]

apps/cic-ussd/cic_ussd/runnable/server.py

@@ -23,10 +23,11 @@ from cic_ussd.encoder import PasswordEncoder
 from cic_ussd.files.local_files import create_local_file_data_stores, json_file_parser
 from cic_ussd.menu.ussd_menu import UssdMenu
 from cic_ussd.metadata.signer import Signer
-from cic_ussd.metadata.user import UserMetadata
+from cic_ussd.metadata.base import Metadata
 from cic_ussd.operations import (define_response_with_content,
                                  process_menu_interaction_requests,
                                  define_multilingual_responses)
+from cic_ussd.phone_number import process_phone_number
 from cic_ussd.redis import InMemoryStore
 from cic_ussd.requests import (get_request_endpoint,
                                get_request_method,
@@ -35,7 +36,8 @@ from cic_ussd.requests import (get_request_endpoint,
                                process_pin_reset_requests)
 from cic_ussd.session.ussd_session import UssdSession as InMemoryUssdSession
 from cic_ussd.state_machine import UssdStateMachine
-from cic_ussd.validator import check_ip, check_request_content_length, check_service_code, validate_phone_number
+from cic_ussd.validator import check_ip, check_request_content_length, check_service_code, validate_phone_number, \
+    validate_presence
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
@@ -55,20 +57,17 @@ arg_parser.add_argument('--env-prefix',
                         help='environment prefix for variables to overwrite configuration')
 args = arg_parser.parse_args()
 
+# define log levels
+if args.vv:
+    logging.getLogger().setLevel(logging.DEBUG)
+elif args.v:
+    logging.getLogger().setLevel(logging.INFO)
+
 # parse config
 config = Config(config_dir=args.c, env_prefix=args.env_prefix)
 config.process()
 config.censor('PASSWORD', 'DATABASE')
-
-# define log levels
-if args.vv:
-    logging.getLogger().setLevel(logging.DEBUG)
-    logging.getLogger('sqlalchemy.engine').setLevel(logging.DEBUG)
-elif args.v:
-    logging.getLogger().setLevel(logging.INFO)
-
-# log config vars
-logg.debug(config)
+logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 
 # initialize elements
 # set up translations
@@ -85,7 +84,7 @@ UssdMenu.ussd_menu_db = ussd_menu_db
 
 # set up db
 data_source_name = dsn_from_config(config)
-SessionBase.connect(data_source_name=data_source_name)
+SessionBase.connect(data_source_name, pool_size=int(config.get('DATABASE_POOL_SIZE')), debug=config.true('DATABASE_DEBUG'))
 # create session for the life time of http request
 SessionBase.session = SessionBase.create_session()
 
@@ -98,12 +97,18 @@ InMemoryStore.cache = redis.StrictRedis(host=config.get('REDIS_HOSTNAME'),
 InMemoryUssdSession.redis_cache = InMemoryStore.cache
 
 # define metadata URL
-UserMetadata.base_url = config.get('CIC_META_URL')
+Metadata.base_url = config.get('CIC_META_URL')
 
 # define signer values
-Signer.gpg_path = config.get('PGP_EXPORT_DIR')
+export_dir = config.get('PGP_EXPORT_DIR')
+if export_dir:
+    validate_presence(path=export_dir)
+Signer.gpg_path = export_dir
 Signer.gpg_passphrase = config.get('PGP_PASSPHRASE')
-Signer.key_file_path = f"{config.get('PGP_KEYS_PATH')}{config.get('PGP_PRIVATE_KEYS')}"
+key_file_path = f"{config.get('PGP_KEYS_PATH')}{config.get('PGP_PRIVATE_KEYS')}"
+if key_file_path:
+    validate_presence(path=key_file_path)
+Signer.key_file_path = key_file_path
 
 # initialize celery app
 celery.Celery(backend=config.get('CELERY_RESULT_URL'), broker=config.get('CELERY_BROKER_URL'))
@@ -144,6 +149,10 @@ def application(env, start_response):
         external_session_id = post_data.get('sessionId')
         user_input = post_data.get('text')
 
+        # add validation for phone number
+        if phone_number:
+            phone_number = process_phone_number(phone_number=phone_number, region=config.get('PHONE_NUMBER_REGION'))
+
         # validate ip address
         if not check_ip(config=config, env=env):
             start_response('403 Sneaky, sneaky', errors_headers)
@@ -167,8 +176,10 @@ def application(env, start_response):
 
         # validate phone number
         if not validate_phone_number(phone_number):
+            logg.error('invalid phone number {}'.format(phone_number))
             start_response('400 Invalid phone number format', errors_headers)
             return []
+        logg.debug('session {} started for {}'.format(external_session_id, phone_number))
 
         # handle menu interaction requests
         chain_str = chain_spec.__str__()

apps/cic-ussd/cic_ussd/runnable/tasker.py

@@ -6,6 +6,7 @@ import tempfile
 
 # third party imports
 import celery
+import i18n
 import redis
 from confini import Config
 
@@ -13,12 +14,14 @@ from confini import Config
 from cic_ussd.db import dsn_from_config
 from cic_ussd.db.models.base import SessionBase
 from cic_ussd.metadata.signer import Signer
-from cic_ussd.metadata.user import UserMetadata
+from cic_ussd.metadata.base import Metadata
 from cic_ussd.redis import InMemoryStore
 from cic_ussd.session.ussd_session import UssdSession as InMemoryUssdSession
+from cic_ussd.validator import validate_presence
 
 logging.basicConfig(level=logging.WARNING)
 logg = logging.getLogger()
+logging.getLogger('gnupg').setLevel(logging.WARNING)
 
 config_directory = '/usr/local/etc/cic-ussd/'
 
@@ -31,22 +34,22 @@ arg_parser.add_argument('-vv', action='store_true', help='be more verbose')
 arg_parser.add_argument('--env-prefix', default=os.environ.get('CONFINI_ENV_PREFIX'), dest='env_prefix', type=str, help='environment prefix for variables to overwrite configuration')
 args = arg_parser.parse_args()
 
-# parse config
-config = Config(config_dir=args.c, env_prefix=args.env_prefix)
-config.process()
-config.censor('PASSWORD', 'DATABASE')
-
 # define log levels
 if args.vv:
     logging.getLogger().setLevel(logging.DEBUG)
 elif args.v:
     logging.getLogger().setLevel(logging.INFO)
 
-logg.debug(config)
+# parse config
+config = Config(args.c, args.env_prefix)
+config.process()
+config.add(args.q, '_CELERY_QUEUE', True)
+config.censor('PASSWORD', 'DATABASE')
+logg.debug('config loaded from {}:\n{}'.format(args.c, config))
 
 # connect to database
 data_source_name = dsn_from_config(config)
-SessionBase.connect(data_source_name=data_source_name)
+SessionBase.connect(data_source_name, pool_size=int(config.get('DATABASE_POOL_SIZE')), debug=config.true('DATABASE_DEBUG'))
 
 # verify database connection with minimal sanity query
 session = SessionBase.create_session()
@@ -62,12 +65,22 @@ InMemoryStore.cache = redis.StrictRedis(host=config.get('REDIS_HOSTNAME'),
 InMemoryUssdSession.redis_cache = InMemoryStore.cache
 
 # define metadata URL
-UserMetadata.base_url = config.get('CIC_META_URL')
+Metadata.base_url = config.get('CIC_META_URL')
 
 # define signer values
-Signer.gpg_path = config.get('PGP_EXPORT_DIR')
+export_dir = config.get('PGP_EXPORT_DIR')
+if export_dir:
+    validate_presence(path=export_dir)
+Signer.gpg_path = export_dir
 Signer.gpg_passphrase = config.get('PGP_PASSPHRASE')
-Signer.key_file_path = f"{config.get('PGP_KEYS_PATH')}{config.get('PGP_PRIVATE_KEYS')}"
+key_file_path = f"{config.get('PGP_KEYS_PATH')}{config.get('PGP_PRIVATE_KEYS')}"
+if key_file_path:
+    validate_presence(path=key_file_path)
+Signer.key_file_path = key_file_path
+
+# set up translations
+i18n.load_path.append(config.get('APP_LOCALE_PATH'))
+i18n.set('fallback', config.get('APP_LOCALE_FALLBACK'))
 
 # set up celery
 current_app = celery.Celery(__name__)

apps/cic-ussd/cic_ussd/state_machine/logic/balance.py

@@ -5,12 +5,12 @@ from typing import Tuple
 # third-party imports
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 
 logg = logging.getLogger(__file__)
 
 
-def process_mini_statement_request(state_machine_data: Tuple[str, dict, User]):
+def process_mini_statement_request(state_machine_data: Tuple[str, dict, Account]):
     """This function compiles a brief statement of a user's last three inbound and outbound transactions and send the
     same as a message on their selected avenue for notification.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.

apps/cic-ussd/cic_ussd/state_machine/logic/menu.py

@@ -6,10 +6,10 @@ ussd menu facilitating the return of appropriate menu responses based on said us
 from typing import Tuple
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 
 
-def menu_one_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_one_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks that user input matches a string with value '1'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str
@@ -20,7 +20,7 @@ def menu_one_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user_input == '1'
 
 
-def menu_two_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_two_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks that user input matches a string with value '2'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -31,7 +31,7 @@ def menu_two_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user_input == '2'
 
 
-def menu_three_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_three_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks that user input matches a string with value '3'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -42,7 +42,7 @@ def menu_three_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user_input == '3'
 
 
-def menu_four_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_four_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """
     This function checks that user input matches a string with value '4'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -54,7 +54,7 @@ def menu_four_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user_input == '4'
 
 
-def menu_five_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_five_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """
     This function checks that user input matches a string with value '5'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -66,7 +66,7 @@ def menu_five_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user_input == '5'
 
 
-def menu_zero_zero_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_zero_zero_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """
     This function checks that user input matches a string with value '00'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -78,7 +78,7 @@ def menu_zero_zero_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user_input == '00'
 
 
-def menu_ninety_nine_selected(state_machine_data: Tuple[str, dict, User]) -> bool:
+def menu_ninety_nine_selected(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """
     This function checks that user input matches a string with value '99'
     :param state_machine_data: A tuple containing user input, a ussd session and user object.

apps/cic-ussd/cic_ussd/state_machine/logic/pin.py

@@ -12,7 +12,7 @@ from typing import Tuple
 import bcrypt
 
 # local imports
-from cic_ussd.db.models.user import AccountStatus, User
+from cic_ussd.db.models.account import AccountStatus, Account
 from cic_ussd.encoder import PasswordEncoder, create_password_hash
 from cic_ussd.operations import persist_session_to_db_task, create_or_update_session
 from cic_ussd.redis import InMemoryStore
@@ -21,7 +21,7 @@ from cic_ussd.redis import InMemoryStore
 logg = logging.getLogger(__file__)
 
 
-def is_valid_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_valid_pin(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks a pin's validity by ensuring it has a length of for characters and the characters are
     numeric.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -37,7 +37,7 @@ def is_valid_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
     return pin_is_valid
 
 
-def is_authorized_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_authorized_pin(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks whether the user input confirming a specific pin matches the initial pin entered.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -48,7 +48,7 @@ def is_authorized_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user.verify_password(password=user_input)
 
 
-def is_locked_account(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_locked_account(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks whether a user's account is locked due to too many failed attempts.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -59,7 +59,7 @@ def is_locked_account(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user.get_account_status() == AccountStatus.LOCKED.name
 
 
-def save_initial_pin_to_session_data(state_machine_data: Tuple[str, dict, User]):
+def save_initial_pin_to_session_data(state_machine_data: Tuple[str, dict, Account]):
     """This function hashes a pin and stores it in session data.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -94,7 +94,7 @@ def save_initial_pin_to_session_data(state_machine_data: Tuple[str, dict, User])
     persist_session_to_db_task(external_session_id=external_session_id, queue='cic-ussd')
 
 
-def pins_match(state_machine_data: Tuple[str, dict, User]) -> bool:
+def pins_match(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks whether the user input confirming a specific pin matches the initial pin entered.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -108,7 +108,7 @@ def pins_match(state_machine_data: Tuple[str, dict, User]) -> bool:
     return bcrypt.checkpw(user_input.encode(), initial_pin)
 
 
-def complete_pin_change(state_machine_data: Tuple[str, dict, User]):
+def complete_pin_change(state_machine_data: Tuple[str, dict, Account]):
     """This function persists the user's pin to the database
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -116,11 +116,11 @@ def complete_pin_change(state_machine_data: Tuple[str, dict, User]):
     user_input, ussd_session, user = state_machine_data
     password_hash = ussd_session.get('session_data').get('initial_pin')
     user.password_hash = password_hash
-    User.session.add(user)
-    User.session.commit()
+    Account.session.add(user)
+    Account.session.commit()
 
 
-def is_blocked_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_blocked_pin(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks whether the user input confirming a specific pin matches the initial pin entered.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -131,7 +131,7 @@ def is_blocked_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
     return user.get_account_status() == AccountStatus.LOCKED.name
 
 
-def is_valid_new_pin(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_valid_new_pin(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks whether the user's new pin is a valid pin and that it isn't the same as the old one.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple

apps/cic-ussd/cic_ussd/state_machine/logic/sms.py

@@ -3,21 +3,21 @@ import logging
 from typing import Tuple
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 
 logg = logging.getLogger()
 
 
-def send_terms_to_user_if_required(state_machine_data: Tuple[str, dict, User]):
+def send_terms_to_user_if_required(state_machine_data: Tuple[str, dict, Account]):
     user_input, ussd_session, user = state_machine_data
     logg.debug('Requires integration to cic-notify.')
 
 
-def process_mini_statement_request(state_machine_data: Tuple[str, dict, User]):
+def process_mini_statement_request(state_machine_data: Tuple[str, dict, Account]):
     user_input, ussd_session, user = state_machine_data
     logg.debug('Requires integration to cic-notify.')
 
 
-def upsell_unregistered_recipient(state_machine_data: Tuple[str, dict, User]):
+def upsell_unregistered_recipient(state_machine_data: Tuple[str, dict, Account]):
     user_input, ussd_session, user = state_machine_data
     logg.debug('Requires integration to cic-notify.')

apps/cic-ussd/cic_ussd/state_machine/logic/transaction.py

@@ -9,7 +9,7 @@ import celery
 # local imports
 from cic_ussd.balance import BalanceManager, compute_operational_balance
 from cic_ussd.chain import Chain
-from cic_ussd.db.models.user import AccountStatus, User
+from cic_ussd.db.models.account import AccountStatus, Account
 from cic_ussd.operations import save_to_in_memory_ussd_session_data
 from cic_ussd.phone_number import get_user_by_phone_number
 from cic_ussd.redis import create_cached_data_key, get_cached_data
@@ -19,7 +19,7 @@ from cic_ussd.transactions import OutgoingTransactionProcessor
 logg = logging.getLogger(__file__)
 
 
-def is_valid_recipient(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_valid_recipient(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks that a user exists, is not the initiator of the transaction, has an active account status
     and is authorized to perform  standard transactions.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -34,7 +34,7 @@ def is_valid_recipient(state_machine_data: Tuple[str, dict, User]) -> bool:
     return is_not_initiator and has_active_account_status and recipient is not None
 
 
-def is_valid_transaction_amount(state_machine_data: Tuple[str, dict, User]) -> bool:
+def is_valid_transaction_amount(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks that the transaction amount provided is valid as per the criteria for the transaction
     being attempted.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -49,7 +49,7 @@ def is_valid_transaction_amount(state_machine_data: Tuple[str, dict, User]) -> b
         return False
 
 
-def has_sufficient_balance(state_machine_data: Tuple[str, dict, User]) -> bool:
+def has_sufficient_balance(state_machine_data: Tuple[str, dict, Account]) -> bool:
     """This function checks that the transaction amount provided is valid as per the criteria for the transaction
     being attempted.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
@@ -72,19 +72,20 @@ def has_sufficient_balance(state_machine_data: Tuple[str, dict, User]) -> bool:
     return int(user_input) <= operational_balance
 
 
-def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, User]):
+def save_recipient_phone_to_session_data(state_machine_data: Tuple[str, dict, Account]):
     """This function saves the phone number corresponding the intended recipients blockchain account.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str
     """
     user_input, ussd_session, user = state_machine_data
-    session_data = {
-        'recipient_phone_number': user_input
-    }
+
+    session_data = ussd_session.get('session_data') or {}
+    session_data['recipient_phone_number'] = user_input
+
     save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
 
 
-def retrieve_recipient_metadata(state_machine_data: Tuple[str, dict, User]):
+def retrieve_recipient_metadata(state_machine_data: Tuple[str, dict, Account]):
     """
     :param state_machine_data:
     :type state_machine_data:
@@ -96,26 +97,27 @@ def retrieve_recipient_metadata(state_machine_data: Tuple[str, dict, User]):
     recipient = get_user_by_phone_number(phone_number=user_input)
     blockchain_address = recipient.blockchain_address
     # retrieve and cache account's metadata
-    s_query_user_metadata = celery.signature(
-        'cic_ussd.tasks.metadata.query_user_metadata',
+    s_query_person_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.query_person_metadata',
         [blockchain_address]
     )
-    s_query_user_metadata.apply_async(queue='cic-ussd')
+    s_query_person_metadata.apply_async(queue='cic-ussd')
 
 
-def save_transaction_amount_to_session_data(state_machine_data: Tuple[str, dict, User]):
+def save_transaction_amount_to_session_data(state_machine_data: Tuple[str, dict, Account]):
     """This function saves the phone number corresponding the intended recipients blockchain account.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str
     """
     user_input, ussd_session, user = state_machine_data
-    session_data = {
-        'transaction_amount': user_input
-    }
+
+    session_data = ussd_session.get('session_data') or {}
+    session_data['transaction_amount'] = user_input
+
     save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
 
 
-def process_transaction_request(state_machine_data: Tuple[str, dict, User]):
+def process_transaction_request(state_machine_data: Tuple[str, dict, Account]):
     """This function saves the phone number corresponding the intended recipients blockchain account.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str

apps/cic-ussd/cic_ussd/state_machine/logic/user.py

@@ -10,8 +10,8 @@ from cic_types.models.person import generate_vcard_from_contact_data, manage_ide
 
 # local imports
 from cic_ussd.chain import Chain
-from cic_ussd.db.models.user import User
-from cic_ussd.error import UserMetadataNotFoundError
+from cic_ussd.db.models.account import Account
+from cic_ussd.error import MetadataNotFoundError
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.operations import save_to_in_memory_ussd_session_data
 from cic_ussd.redis import get_cached_data
@@ -19,40 +19,40 @@ from cic_ussd.redis import get_cached_data
 logg = logging.getLogger(__file__)
 
 
-def change_preferred_language_to_en(state_machine_data: Tuple[str, dict, User]):
+def change_preferred_language_to_en(state_machine_data: Tuple[str, dict, Account]):
     """This function changes the user's preferred language to english.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
     """
     user_input, ussd_session, user = state_machine_data
     user.preferred_language = 'en'
-    User.session.add(user)
-    User.session.commit()
+    Account.session.add(user)
+    Account.session.commit()
 
 
-def change_preferred_language_to_sw(state_machine_data: Tuple[str, dict, User]):
+def change_preferred_language_to_sw(state_machine_data: Tuple[str, dict, Account]):
     """This function changes the user's preferred language to swahili.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
     """
     user_input, ussd_session, user = state_machine_data
     user.preferred_language = 'sw'
-    User.session.add(user)
-    User.session.commit()
+    Account.session.add(user)
+    Account.session.commit()
 
 
-def update_account_status_to_active(state_machine_data: Tuple[str, dict, User]):
+def update_account_status_to_active(state_machine_data: Tuple[str, dict, Account]):
     """This function sets user's account to active.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
     """
     user_input, ussd_session, user = state_machine_data
     user.activate_account()
-    User.session.add(user)
-    User.session.commit()
+    Account.session.add(user)
+    Account.session.commit()
 
 
-def process_gender_user_input(user: User, user_input: str):
+def process_gender_user_input(user: Account, user_input: str):
     """
     :param user:
     :type user:
@@ -74,7 +74,7 @@ def process_gender_user_input(user: User, user_input: str):
     return gender
 
 
-def save_metadata_attribute_to_session_data(state_machine_data: Tuple[str, dict, User]):
+def save_metadata_attribute_to_session_data(state_machine_data: Tuple[str, dict, Account]):
     """This function saves first name data to the ussd session in the redis cache.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -109,7 +109,7 @@ def save_metadata_attribute_to_session_data(state_machine_data: Tuple[str, dict,
     save_to_in_memory_ussd_session_data(queue='cic-ussd', session_data=session_data, ussd_session=ussd_session)
 
 
-def format_user_metadata(metadata: dict, user: User):
+def format_user_metadata(metadata: dict, user: Account):
     """
     :param metadata:
     :type metadata:
@@ -150,7 +150,7 @@ def format_user_metadata(metadata: dict, user: User):
     }
 
 
-def save_complete_user_metadata(state_machine_data: Tuple[str, dict, User]):
+def save_complete_user_metadata(state_machine_data: Tuple[str, dict, Account]):
     """This function persists elements of the user metadata stored in session data
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: tuple
@@ -164,14 +164,14 @@ def save_complete_user_metadata(state_machine_data: Tuple[str, dict, User]):
     user_metadata = format_user_metadata(metadata=metadata, user=user)
 
     blockchain_address = user.blockchain_address
-    s_create_user_metadata = celery.signature(
-        'cic_ussd.tasks.metadata.create_user_metadata',
+    s_create_person_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.create_person_metadata',
         [blockchain_address, user_metadata]
     )
-    s_create_user_metadata.apply_async(queue='cic-ussd')
+    s_create_person_metadata.apply_async(queue='cic-ussd')
 
 
-def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, User]):
+def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, Account]):
     user_input, ussd_session, user = state_machine_data
     blockchain_address = user.blockchain_address
     key = generate_metadata_pointer(
@@ -181,7 +181,7 @@ def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, User]):
     user_metadata = get_cached_data(key=key)
 
     if not user_metadata:
-        raise UserMetadataNotFoundError(f'Expected user metadata but found none in cache for key: {blockchain_address}')
+        raise MetadataNotFoundError(f'Expected user metadata but found none in cache for key: {blockchain_address}')
 
     given_name = ussd_session.get('session_data').get('given_name')
     family_name = ussd_session.get('session_data').get('family_name')
@@ -192,7 +192,7 @@ def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, User]):
     # validate user metadata
     person = Person()
     user_metadata = json.loads(user_metadata)
-    deserialized_person = person.deserialize(metadata=user_metadata)
+    deserialized_person = person.deserialize(person_data=user_metadata)
 
     # edit specific metadata attribute
     if given_name:
@@ -211,18 +211,18 @@ def edit_user_metadata_attribute(state_machine_data: Tuple[str, dict, User]):
 
     edited_metadata = deserialized_person.serialize()
 
-    s_edit_user_metadata = celery.signature(
-        'cic_ussd.tasks.metadata.edit_user_metadata',
-        [blockchain_address, edited_metadata, 'pgp']
+    s_edit_person_metadata = celery.signature(
+        'cic_ussd.tasks.metadata.edit_person_metadata',
+        [blockchain_address, edited_metadata]
     )
-    s_edit_user_metadata.apply_async(queue='cic-ussd')
+    s_edit_person_metadata.apply_async(queue='cic-ussd')
 
 
-def get_user_metadata(state_machine_data: Tuple[str, dict, User]):
+def get_user_metadata(state_machine_data: Tuple[str, dict, Account]):
     user_input, ussd_session, user = state_machine_data
     blockchain_address = user.blockchain_address
     s_get_user_metadata = celery.signature(
-        'cic_ussd.tasks.metadata.query_user_metadata',
+        'cic_ussd.tasks.metadata.query_person_metadata',
         [blockchain_address]
     )
     s_get_user_metadata.apply_async(queue='cic-ussd')

apps/cic-ussd/cic_ussd/state_machine/logic/validator.py

@@ -7,14 +7,14 @@ from typing import Tuple
 from cic_types.models.person import generate_metadata_pointer
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.redis import get_cached_data
 
 logg = logging.getLogger()
 
 
-def has_cached_user_metadata(state_machine_data: Tuple[str, dict, User]):
+def has_cached_user_metadata(state_machine_data: Tuple[str, dict, Account]):
     """This function checks whether the attributes of the user's metadata constituting a profile are filled out.
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str
@@ -29,7 +29,7 @@ def has_cached_user_metadata(state_machine_data: Tuple[str, dict, User]):
     return user_metadata is not None
 
 
-def is_valid_name(state_machine_data: Tuple[str, dict, User]):
+def is_valid_name(state_machine_data: Tuple[str, dict, Account]):
     """This function checks that a user provided name is valid
     :param state_machine_data: A tuple containing user input, a ussd session and user object.
     :type state_machine_data: str
@@ -43,7 +43,7 @@ def is_valid_name(state_machine_data: Tuple[str, dict, User]):
         return False
 
 
-def is_valid_gender_selection(state_machine_data: Tuple[str, dict, User]):
+def is_valid_gender_selection(state_machine_data: Tuple[str, dict, Account]):
     """
     :param state_machine_data:
     :type state_machine_data:

apps/cic-ussd/cic_ussd/state_machine/state_machine.py

@@ -13,7 +13,7 @@ class UssdStateMachine(Machine):
     """This class describes a finite state machine responsible for maintaining all the states that describe the ussd
     menu  as well as providing a means for navigating through these states based on different user inputs.
     It defines different helper functions that co-ordinate with the stakeholder components of the ussd menu: i.e  the
-    User, UssdSession, UssdMenu to facilitate user interaction with ussd menu.
+    Account, UssdSession, UssdMenu to facilitate user interaction with ussd menu.
     :cvar states: A list of pre-defined states.
     :type states: list
     :cvar transitions: A list of pre-defined transitions.

apps/cic-ussd/cic_ussd/tasks/base.py

@@ -5,9 +5,24 @@ import celery
 import sqlalchemy
 
 # local imports
+from cic_ussd.error import MetadataStoreError
+from cic_ussd.db.models.base import SessionBase
 
 
-class CriticalTask(celery.Task):
+class BaseTask(celery.Task):
+
+    session_func = SessionBase.create_session
+
+    def create_session(self):
+        return BaseTask.session_func()
+
+
+    def log_banner(self):
+        logg.debug('task {} root uuid {}'.format(self.__class__.__name__, self.request.root_id))
+        return
+
+    
+class CriticalTask(BaseTask):
     retry_jitter = True
     retry_backoff = True
     retry_backoff_max = 8
@@ -17,4 +32,11 @@ class CriticalSQLAlchemyTask(CriticalTask):
     autoretry_for = (
         sqlalchemy.exc.DatabaseError,
         sqlalchemy.exc.TimeoutError,
+        sqlalchemy.exc.ResourceClosedError,
+        ) 
+
+
+class CriticalMetadataTask(CriticalTask):
+    autoretry_for = (
+        MetadataStoreError,
     )

apps/cic-ussd/cic_ussd/tasks/callback_handler.py

@@ -9,7 +9,7 @@ import celery
 # local imports
 from cic_ussd.conversions import from_wei
 from cic_ussd.db.models.base import SessionBase
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 from cic_ussd.account import define_account_tx_metadata
 from cic_ussd.error import ActionDataNotFoundError
 from cic_ussd.redis import InMemoryStore, cache_data, create_cached_data_key
@@ -49,10 +49,17 @@ def process_account_creation_callback(self, result: str, url: str, status_code:
             phone_number = account_creation_data.get('phone_number')
 
             # create user
-            user = User(blockchain_address=result, phone_number=phone_number)
+            user = Account(blockchain_address=result, phone_number=phone_number)
             session.add(user)
             session.commit()
             session.close()
+    
+            queue = self.request.delivery_info.get('routing_key')
+            s = celery.signature(
+                    'cic_ussd.tasks.metadata.add_phone_pointer',
+                    [result, phone_number]
+                    )
+            s.apply_async(queue=queue)
 
             # expire cache
             cache.expire(task_id, timedelta(seconds=180))
@@ -65,6 +72,8 @@ def process_account_creation_callback(self, result: str, url: str, status_code:
         session.close()
         raise ActionDataNotFoundError(f'Account creation task: {task_id}, returned unexpected response: {status_code}')
 
+    session.close()
+
 
 @celery_app.task
 def process_incoming_transfer_callback(result: dict, param: str, status_code: int):
@@ -74,12 +83,12 @@ def process_incoming_transfer_callback(result: dict, param: str, status_code: in
         # collect result data
         recipient_blockchain_address = result.get('recipient')
         sender_blockchain_address = result.get('sender')
-        token_symbol = result.get('token_symbol')
-        value = result.get('destination_value')
+        token_symbol = result.get('destination_token_symbol')
+        value = result.get('destination_token_value')
 
         # try to find users in system
-        recipient_user = session.query(User).filter_by(blockchain_address=recipient_blockchain_address).first()
-        sender_user = session.query(User).filter_by(blockchain_address=sender_blockchain_address).first()
+        recipient_user = session.query(Account).filter_by(blockchain_address=recipient_blockchain_address).first()
+        sender_user = session.query(Account).filter_by(blockchain_address=sender_blockchain_address).first()
 
         # check whether recipient is in the system
         if not recipient_user:
@@ -118,6 +127,7 @@ def process_incoming_transfer_callback(result: dict, param: str, status_code: in
         session.close()
         raise ValueError(f'Unexpected status code: {status_code}.')
 
+    session.close()
 
 @celery_app.task
 def process_balances_callback(result: list, param: str, status_code: int):
@@ -143,21 +153,24 @@ def define_transaction_action_tag(
         # check preferred language
         if preferred_language == 'en':
             action_tag = 'SENT'
+            direction = 'TO'
         else:
             action_tag = 'ULITUMA'
+            direction = 'KWA'
     else:
         if preferred_language == 'en':
             action_tag = 'RECEIVED'
+            direction = 'FROM'
         else:
             action_tag = 'ULIPOKEA'
-    return action_tag
+            direction = 'KUTOKA'
+    return action_tag, direction
 
 
 @celery_app.task
 def process_statement_callback(result, param: str, status_code: int):
     if status_code == 0:
         # create session
-        session = SessionBase.create_session()
         processed_transactions = []
 
         # process transaction data to cache
@@ -170,35 +183,40 @@ def process_statement_callback(result, param: str, status_code: int):
             if '0x0000000000000000000000000000000000000000' in source_token:
                 pass
             else:
+                session = SessionBase.create_session()
                 # describe a processed transaction
                 processed_transaction = {}
 
                 # check if sender is in the system
-                sender: User = session.query(User).filter_by(blockchain_address=sender_blockchain_address).first()
+                sender: Account = session.query(Account).filter_by(blockchain_address=sender_blockchain_address).first()
+                owner: Account = session.query(Account).filter_by(blockchain_address=param).first()
                 if sender:
                     processed_transaction['sender_phone_number'] = sender.phone_number
 
-                    action_tag = define_transaction_action_tag(
-                        preferred_language=sender.preferred_language,
+                    action_tag, direction = define_transaction_action_tag(
+                        preferred_language=owner.preferred_language,
                         sender_blockchain_address=sender_blockchain_address,
                         param=param
                     )
                     processed_transaction['action_tag'] = action_tag
+                    processed_transaction['direction'] = direction
 
                 else:
                     processed_transaction['sender_phone_number'] = 'GRASSROOTS ECONOMICS'
 
                 # check if recipient is in the system
-                recipient: User = session.query(User).filter_by(blockchain_address=recipient_address).first()
+                recipient: Account = session.query(Account).filter_by(blockchain_address=recipient_address).first()
                 if recipient:
                     processed_transaction['recipient_phone_number'] = recipient.phone_number
 
                 else:
                     logg.warning(f'Tx with recipient not found in cic-ussd')
 
+                session.close()
+
                 # add transaction values
-                processed_transaction['to_value'] = from_wei(value=transaction.get('to_value'))
-                processed_transaction['from_value'] = from_wei(value=transaction.get('from_value'))
+                processed_transaction['to_value'] = from_wei(value=transaction.get('to_value')).__str__()
+                processed_transaction['from_value'] = from_wei(value=transaction.get('from_value')).__str__()
 
                 raw_timestamp = transaction.get('timestamp')
                 timestamp = datetime.utcfromtimestamp(raw_timestamp).strftime('%d/%m/%y, %H:%M')

apps/cic-ussd/cic_ussd/tasks/metadata.py

@@ -1,20 +1,22 @@
 # standard imports
-import json
 import logging
 
 # third-party imports
 import celery
+from hexathon import strip_0x
 
 # local imports
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
-from cic_ussd.metadata.user import UserMetadata
+from cic_ussd.metadata.person import PersonMetadata
+from cic_ussd.metadata.phone import PhonePointerMetadata
+from cic_ussd.tasks.base import CriticalMetadataTask
 
 celery_app = celery.current_app
-logg = logging.getLogger()
+logg = logging.getLogger().getChild(__name__)
 
 
 @celery_app.task
-def query_user_metadata(blockchain_address: str):
+def query_person_metadata(blockchain_address: str):
     """
     :param blockchain_address:
     :type blockchain_address:
@@ -22,12 +24,12 @@ def query_user_metadata(blockchain_address: str):
     :rtype:
     """
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
-    user_metadata_client.query()
+    person_metadata_client = PersonMetadata(identifier=identifier)
+    person_metadata_client.query()
 
 
 @celery_app.task
-def create_user_metadata(blockchain_address: str, data: dict):
+def create_person_metadata(blockchain_address: str, data: dict):
     """
     :param blockchain_address:
     :type blockchain_address:
@@ -37,12 +39,20 @@ def create_user_metadata(blockchain_address: str, data: dict):
     :rtype:
     """
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
-    user_metadata_client.create(data=data)
+    person_metadata_client = PersonMetadata(identifier=identifier)
+    person_metadata_client.create(data=data)
 
 
 @celery_app.task
-def edit_user_metadata(blockchain_address: str, data: bytes, engine: str):
+def edit_person_metadata(blockchain_address: str, data: bytes):
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
-    user_metadata_client.edit(data=data, engine=engine)
+    person_metadata_client = PersonMetadata(identifier=identifier)
+    person_metadata_client.edit(data=data)
+
+
+@celery_app.task(bind=True, base=CriticalMetadataTask)
+def add_phone_pointer(self, blockchain_address: str, phone_number: str):
+    identifier = phone_number.encode('utf-8')
+    stripped_address = strip_0x(blockchain_address)
+    phone_metadata_client = PhonePointerMetadata(identifier=identifier)
+    phone_metadata_client.create(data=stripped_address)

apps/cic-ussd/cic_ussd/tasks/ussd_session.py

@@ -70,3 +70,4 @@ def persist_session_to_db(external_session_id: str):
         session.close()
         raise SessionNotFoundError('Session does not exist!')
 
+    session.close()

apps/cic-ussd/cic_ussd/transactions.py

@@ -47,7 +47,7 @@ def to_wei(value: int) -> int:
     :return: Wei equivalent of value in SRF
     :rtype: int
     """
-    return int(value * 1e+18)
+    return int(value * 1e+6)
 
 
 class IncomingTransactionProcessor:

apps/cic-ussd/cic_ussd/translation.py

@@ -8,7 +8,7 @@ from typing import Optional
 def translation_for(key: str, preferred_language: Optional[str] = None, **kwargs) -> str:
     """
     Translates text mapped to a specific YAML key into the user's set preferred language.
-    :param preferred_language: User's preferred language in which to view the ussd menu.
+    :param preferred_language: Account's preferred language in which to view the ussd menu.
     :type preferred_language str
     :param key: Key to a specific YAML test entry
     :type key: str

apps/cic-ussd/cic_ussd/validator.py

@@ -1,12 +1,14 @@
 # standard imports
 import logging
+import os
 import re
+import ipaddress
 
 # third-party imports
 from confini import Config
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 
 logg = logging.getLogger(__file__)
 
@@ -20,7 +22,14 @@ def check_ip(config: Config, env: dict):
     :return: Request IP validity
     :rtype: boolean
     """
-    return env.get('REMOTE_ADDR') == config.get('APP_ALLOWED_IP')
+    # TODO: do once at boot time
+    actual_ip = ipaddress.ip_network(env.get('REMOTE_ADDR') + '/32')
+    for allowed_net_src in config.get('APP_ALLOWED_IP').split(','):
+        allowed_net = ipaddress.ip_network(allowed_net_src)
+        if actual_ip.subnet_of(allowed_net):
+            return True
+
+    return False
 
 
 def check_request_content_length(config: Config, env: dict):
@@ -59,7 +68,7 @@ def check_known_user(phone: str):
     :return: Is known phone number
     :rtype: boolean
     """
-    user = User.session.query(User).filter_by(phone_number=phone).first()
+    user = Account.session.query(Account).filter_by(phone_number=phone).first()
     return user is not None
 
 
@@ -110,7 +119,7 @@ def validate_phone_number(phone: str):
 
 
 def validate_response_type(processor_response: str) -> bool:
-    """1*3443*3443*Philip*Wanga*1*Juja*Software Developer*2*3
+    """
     This function checks the prefix for a corresponding menu's text from the response offered by the Ussd Processor and
     determines whether the response should prompt the end of a ussd session or the
     :param processor_response: A ussd menu's text value.
@@ -126,3 +135,14 @@ def validate_response_type(processor_response: str) -> bool:
         return True
     return False
 
+
+def validate_presence(path: str):
+    """
+
+    """
+    is_present = os.path.exists(path=path)
+
+    if not is_present:
+        raise ValueError(f'Directory/File in path: {path} not found.')
+    else:
+        logg.debug(f'Loading data from: {path}')

apps/cic-ussd/cic_ussd/version.py

@@ -1,7 +1,7 @@
 # standard imports
 import semver
 
-version = (0, 3, 0, 'alpha.8')
+version = (0, 3, 0, 'alpha.9')
 
 version_object = semver.VersionInfo(
         major=version[0],

apps/cic-ussd/docker/Dockerfile

@@ -51,4 +51,4 @@ RUN cd cic-ussd && \
 COPY cic-ussd/.config/ /usr/local/etc/cic-ussd/
 COPY cic-ussd/cic_ussd/db/migrations/ /usr/local/share/cic-ussd/alembic
 
-WORKDIR /root
+WORKDIR /root

apps/cic-ussd/docker/start_tasker.sh

@@ -2,4 +2,4 @@
 
 . /root/db.sh
 
-/usr/local/bin/cic-ussd-tasker -vv "$@"
+/usr/local/bin/cic-ussd-tasker $@

apps/cic-ussd/docker/start_uwsgi.sh

@@ -2,4 +2,6 @@
 
 . /root/db.sh
 
-/usr/local/bin/uwsgi --wsgi-file /usr/local/lib/python3.8/site-packages/cic_ussd/runnable/server.py --http :9000 --pyargv "-vv"
+server_port=${SERVER_PORT:-9000}
+
+/usr/local/bin/uwsgi --wsgi-file /usr/local/lib/python3.8/site-packages/cic_ussd/runnable/server.py --http :$server_port --pyargv "$@"

apps/cic-ussd/requirements.txt

@@ -1,4 +1,4 @@
-cic_base[full_graph]~=0.1.2a58
-cic-eth~=0.11.0a4
-cic-notify~=0.4.0a3
+cic_base[full_graph]~=0.1.2a79
+cic-eth~=0.11.0b7
+cic-notify~=0.4.0a4
 cic-types~=0.1.0a10

apps/cic-ussd/states/user_metadata_states.json

@@ -4,5 +4,6 @@
   "enter_gender",
   "enter_age",
   "enter_location",
-  "enter_products"
+  "enter_products",
+  "display_metadata_pin_authorization"
 ]

apps/cic-ussd/tests/cic_ussd/db/models/test_user.py

@@ -4,19 +4,19 @@
 import pytest
 
 # platform imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 
 
 def test_user(init_database, set_fernet_key):
-    user = User(blockchain_address='0x417f5962fc52dc33ff0689659b25848680dec6dcedc6785b03d1df60fc6d5c51',
-                phone_number='+254700000000')
+    user = Account(blockchain_address='0x417f5962fc52dc33ff0689659b25848680dec6dcedc6785b03d1df60fc6d5c51',
+                   phone_number='+254700000000')
     user.create_password('0000')
 
-    session = User.session
+    session = Account.session
     session.add(user)
     session.commit()
 
-    queried_user = session.query(User).get(1)
+    queried_user = session.query(Account).get(1)
     assert queried_user.blockchain_address == '0x417f5962fc52dc33ff0689659b25848680dec6dcedc6785b03d1df60fc6d5c51'
     assert queried_user.phone_number == '+254700000000'
     assert queried_user.failed_pin_attempts == 0
@@ -25,7 +25,7 @@ def test_user(init_database, set_fernet_key):
 
 def test_user_state_transition(create_pending_user):
     user = create_pending_user
-    session = User.session
+    session = Account.session
 
     assert user.get_account_status() == 'PENDING'
     user.activate_account()

apps/cic-ussd/tests/cic_ussd/metadata/test_user_metadata.py

@@ -9,26 +9,26 @@ from cic_types.models.person import generate_metadata_pointer
 # local imports
 from cic_ussd.metadata import blockchain_address_to_metadata_pointer
 from cic_ussd.metadata.signer import Signer
-from cic_ussd.metadata.user import UserMetadata
+from cic_ussd.metadata.person import PersonMetadata
 from cic_ussd.redis import get_cached_data
 
 
 def test_user_metadata(create_activated_user, define_metadata_pointer_url, load_config):
-    UserMetadata.base_url = load_config.get('CIC_META_URL')
+    PersonMetadata.base_url = load_config.get('CIC_META_URL')
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
+    person_metadata_client = PersonMetadata(identifier=identifier)
 
-    assert user_metadata_client.url == define_metadata_pointer_url
+    assert person_metadata_client.url == define_metadata_pointer_url
 
 
-def test_create_user_metadata(caplog,
-                              create_activated_user,
-                              define_metadata_pointer_url,
-                              load_config,
-                              mock_meta_post_response,
-                              person_metadata):
+def test_create_person_metadata(caplog,
+                                create_activated_user,
+                                define_metadata_pointer_url,
+                                load_config,
+                                mock_meta_post_response,
+                                person_metadata):
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
+    person_metadata_client = PersonMetadata(identifier=identifier)
 
     with requests_mock.Mocker(real_http=False) as request_mocker:
         request_mocker.register_uri(
@@ -38,7 +38,7 @@ def test_create_user_metadata(caplog,
             reason='CREATED',
             content=json.dumps(mock_meta_post_response).encode('utf-8')
         )
-        user_metadata_client.create(data=person_metadata)
+        person_metadata_client.create(data=person_metadata)
         assert 'Get signed material response status: 201' in caplog.text
 
     with pytest.raises(RuntimeError) as error:
@@ -49,19 +49,19 @@ def test_create_user_metadata(caplog,
                 status_code=400,
                 reason='BAD REQUEST'
             )
-            user_metadata_client.create(data=person_metadata)
+            person_metadata_client.create(data=person_metadata)
         assert str(error.value) == f'400 Client Error: BAD REQUEST for url: {define_metadata_pointer_url}'
 
 
-def test_edit_user_metadata(caplog,
-                            create_activated_user,
-                            define_metadata_pointer_url,
-                            load_config,
-                            person_metadata,
-                            setup_metadata_signer):
+def test_edit_person_metadata(caplog,
+                              create_activated_user,
+                              define_metadata_pointer_url,
+                              load_config,
+                              person_metadata,
+                              setup_metadata_signer):
     Signer.gpg_passphrase = load_config.get('KEYS_PASSPHRASE')
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
+    person_metadata_client = PersonMetadata(identifier=identifier)
     with requests_mock.Mocker(real_http=False) as request_mocker:
         request_mocker.register_uri(
             'PUT',
@@ -69,7 +69,7 @@ def test_edit_user_metadata(caplog,
             status_code=200,
             reason='OK'
         )
-        user_metadata_client.edit(data=person_metadata, engine='pgp')
+        person_metadata_client.edit(data=person_metadata)
         assert 'Signed content submission status: 200' in caplog.text
 
     with pytest.raises(RuntimeError) as error:
@@ -80,7 +80,7 @@ def test_edit_user_metadata(caplog,
                 status_code=400,
                 reason='BAD REQUEST'
             )
-            user_metadata_client.edit(data=person_metadata, engine='pgp')
+            person_metadata_client.edit(data=person_metadata)
         assert str(error.value) == f'400 Client Error: BAD REQUEST for url: {define_metadata_pointer_url}'
 
 
@@ -92,7 +92,7 @@ def test_get_user_metadata(caplog,
                            person_metadata,
                            setup_metadata_signer):
     identifier = blockchain_address_to_metadata_pointer(blockchain_address=create_activated_user.blockchain_address)
-    user_metadata_client = UserMetadata(identifier=identifier)
+    person_metadata_client = PersonMetadata(identifier=identifier)
     with requests_mock.Mocker(real_http=False) as request_mocker:
         request_mocker.register_uri(
             'GET',
@@ -101,7 +101,7 @@ def test_get_user_metadata(caplog,
             content=json.dumps(person_metadata).encode('utf-8'),
             reason='OK'
         )
-        user_metadata_client.query()
+        person_metadata_client.query()
         assert 'Get latest data status: 200' in caplog.text
     key = generate_metadata_pointer(
         identifier=identifier,
@@ -118,6 +118,6 @@ def test_get_user_metadata(caplog,
                 status_code=404,
                 reason='NOT FOUND'
             )
-            user_metadata_client.query()
+            person_metadata_client.query()
             assert 'The data is not available and might need to be added.' in caplog.text
         assert str(error.value) == f'400 Client Error: NOT FOUND for url: {define_metadata_pointer_url}'

apps/cic-ussd/tests/cic_ussd/state_machine/logic/test_user_logic.py

@@ -15,7 +15,7 @@ from cic_ussd.state_machine.logic.user import (
     get_user_metadata,
     save_complete_user_metadata,
     process_gender_user_input,
-    save_profile_attribute_to_session_data,
+    save_metadata_attribute_to_session_data,
     update_account_status_to_active)
 
 
@@ -41,14 +41,14 @@ def test_update_account_status_to_active(create_pending_user, create_in_db_ussd_
     ("enter_location", "location", "Kangemi", "Kangemi"),
     ("enter_products", "products", "Mandazi", "Mandazi"),
 ])
-def test_save_save_profile_attribute_to_session_data(current_state,
-                                                     expected_key,
-                                                     expected_result,
-                                                     user_input,
-                                                     celery_session_worker,
-                                                     create_activated_user,
-                                                     create_in_db_ussd_session,
-                                                     create_in_redis_ussd_session):
+def test_save_metadata_attribute_to_session_data(current_state,
+                                                 expected_key,
+                                                 expected_result,
+                                                 user_input,
+                                                 celery_session_worker,
+                                                 create_activated_user,
+                                                 create_in_db_ussd_session,
+                                                 create_in_redis_ussd_session):
     create_in_db_ussd_session.state = current_state
     serialized_in_db_ussd_session = create_in_db_ussd_session.to_json()
     state_machine_data = (user_input, serialized_in_db_ussd_session, create_activated_user)
@@ -56,7 +56,7 @@ def test_save_save_profile_attribute_to_session_data(current_state,
     in_memory_ussd_session = json.loads(in_memory_ussd_session)
     assert in_memory_ussd_session.get('session_data') == {}
     serialized_in_db_ussd_session['state'] = current_state
-    save_profile_attribute_to_session_data(state_machine_data=state_machine_data)
+    save_metadata_attribute_to_session_data(state_machine_data=state_machine_data)
 
     in_memory_ussd_session = InMemoryStore.cache.get('AT974186')
     in_memory_ussd_session = json.loads(in_memory_ussd_session)
@@ -82,23 +82,23 @@ def test_format_user_metadata(create_activated_user,
     from cic_types.models.person import Person
     formatted_user_metadata = format_user_metadata(metadata=complete_user_metadata, user=create_activated_user)
     person = Person()
-    user_metadata = person.deserialize(metadata=formatted_user_metadata)
+    user_metadata = person.deserialize(person_data=formatted_user_metadata)
     assert formatted_user_metadata == user_metadata.serialize()
 
 
 def test_save_complete_user_metadata(celery_session_worker,
-                                        complete_user_metadata,
-                                        create_activated_user,
-                                        create_in_redis_ussd_session,
-                                        mocker,
-                                        setup_chain_spec,
-                                        ussd_session_data):
+                                     complete_user_metadata,
+                                     create_activated_user,
+                                     create_in_redis_ussd_session,
+                                     mocker,
+                                     setup_chain_spec,
+                                     ussd_session_data):
     ussd_session = create_in_redis_ussd_session.get(ussd_session_data.get('external_session_id'))
     ussd_session = json.loads(ussd_session)
     ussd_session['session_data'] = complete_user_metadata
     user_metadata = format_user_metadata(metadata=ussd_session.get('session_data'), user=create_activated_user)
     state_machine_data = ('', ussd_session, create_activated_user)
-    mocked_create_metadata_task = mocker.patch('cic_ussd.tasks.metadata.create_user_metadata.apply_async')
+    mocked_create_metadata_task = mocker.patch('cic_ussd.tasks.metadata.create_person_metadata.apply_async')
     save_complete_user_metadata(state_machine_data=state_machine_data)
     mocked_create_metadata_task.assert_called_with(
         (user_metadata, create_activated_user.blockchain_address),
@@ -127,7 +127,7 @@ def test_edit_user_metadata_attribute(celery_session_worker,
     }
     state_machine_data = ('', ussd_session, create_activated_user)
 
-    mocked_edit_metadata = mocker.patch('cic_ussd.tasks.metadata.edit_user_metadata.apply_async')
+    mocked_edit_metadata = mocker.patch('cic_ussd.tasks.metadata.edit_person_metadata.apply_async')
     edit_user_metadata_attribute(state_machine_data=state_machine_data)
     person_metadata['location']['area_name'] = 'nairobi'
     mocked_edit_metadata.assert_called_with(
@@ -146,7 +146,7 @@ def test_get_user_metadata_attribute(celery_session_worker,
     ussd_session = json.loads(ussd_session)
     state_machine_data = ('', ussd_session, create_activated_user)
 
-    mocked_get_metadata = mocker.patch('cic_ussd.tasks.metadata.query_user_metadata.apply_async')
+    mocked_get_metadata = mocker.patch('cic_ussd.tasks.metadata.query_person_metadata.apply_async')
     get_user_metadata(state_machine_data=state_machine_data)
     mocked_get_metadata.assert_called_with(
         (create_activated_user.blockchain_address,),

apps/cic-ussd/tests/cic_ussd/tasks/test_callback_handler_tasks.py

@@ -8,7 +8,7 @@ import celery
 import pytest
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 from cic_ussd.error import ActionDataNotFoundError
 from cic_ussd.conversions import from_wei
 
@@ -29,7 +29,7 @@ def test_successful_process_account_creation_callback_task(account_creation_acti
     # WARNING: [THE SETTING OF THE ROOT ID IS A HACK AND SHOULD BE REVIEWED OR IMPROVED]
     mocked_task_request.root_id = task_id
 
-    user = init_database.query(User).filter_by(phone_number=phone_number).first()
+    user = init_database.query(Account).filter_by(phone_number=phone_number).first()
     assert user is None
 
     redis_cache = init_redis_cache
@@ -48,7 +48,7 @@ def test_successful_process_account_creation_callback_task(account_creation_acti
     )
     s_process_callback_request.apply_async().get()
 
-    user = init_database.query(User).filter_by(phone_number=phone_number).first()
+    user = init_database.query(Account).filter_by(phone_number=phone_number).first()
     assert user.blockchain_address == result
 
     action_data = redis_cache.get(task_id)

apps/cic-ussd/tests/cic_ussd/test_requests.py

@@ -2,7 +2,7 @@
 import json
 
 # local imports
-from cic_ussd.db.models.user import User
+from cic_ussd.db.models.account import Account
 from cic_ussd.requests import (get_query_parameters,
                                get_request_endpoint,
                                get_request_method,
@@ -58,8 +58,8 @@ def test_process_locked_accounts_requests(create_locked_accounts, valid_locked_a
     assert len(locked_account_addresses) == 10
 
     # check that blockchain addresses are ordered by most recently accessed
-    user_1 = User.session.query(User).filter_by(blockchain_address=locked_account_addresses[2]).first()
-    user_2 = User.session.query(User).filter_by(blockchain_address=locked_account_addresses[7]).first()
+    user_1 = Account.session.query(Account).filter_by(blockchain_address=locked_account_addresses[2]).first()
+    user_2 = Account.session.query(Account).filter_by(blockchain_address=locked_account_addresses[7]).first()
 
     assert user_1.updated > user_2.updated