Remove cic-contracts references

Correct default token env vars in deployment step 2

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

.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):
+    #if SQLBackend.first(chain_spec):
-    #    backend = SyncerBackend.initial(chain_spec, block_offset)
+    #    backend = SQLBackend.initial(chain_spec, block_offset)
-    syncer_backends = SyncerBackend.resume(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.2a62
+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.4a12
+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.2a5
+chainsyncer[sql]~=0.0.2a2
-chainsyncer~=0.0.1a21

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

@@ -2,7 +2,7 @@
 import datetime
 import logging
 
-# third-party imports
+# external imports
 import celery
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.chain import ChainSpec
@@ -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:
@@ -139,3 +145,9 @@ def check_lock(chained_input, chain_spec_dict, lock_flags, address=None):
     session.flush()
     session.close()
     return chained_input
+
+
+@celery_app.task()
+def shutdown(message):
+    logg.critical('shutdown called: {}'.format(message))
+    celery_app.control.shutdown() #broadcast('shutdown')

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

@@ -0,0 +1,19 @@
+# standard imports
+import logging
+
+# external imports
+import celery
+
+# local imports
+from cic_eth.task import BaseTask
+
+celery_app = celery.current_app
+logg = logging.getLogger()
+
+
+@celery_app.task(bind=True, base=BaseTask)
+def default_token(self):
+    return {
+            'symbol': self.default_token_symbol,
+            'address': self.default_token_address,
+            }

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

@@ -60,6 +60,29 @@ class AdminApi:
         self.call_address = call_address
 
 
+    def proxy_do(self, chain_spec, o):
+        s_proxy = celery.signature(
+                'cic_eth.task.rpc_proxy',
+                [
+                    chain_spec.asdict(),
+                    o,
+                    'default',
+                    ],
+                queue=self.queue
+                )
+        return s_proxy.apply_async()
+
+
+    
+    def registry(self):
+        s_registry = celery.signature(
+                'cic_eth.task.registry',
+                [],
+                queue=self.queue
+                )
+        return s_registry.apply_async()
+
+
     def unlock(self, chain_spec, address, flags=None):
         s_unlock = celery.signature(
             'cic_eth.admin.ctrl.unlock',
@@ -146,7 +169,6 @@ class AdminApi:
 
         # TODO: This check should most likely be in resend task itself
         tx_dict = s_get_tx_cache.apply_async().get()
-        #if tx_dict['status'] in [StatusEnum.REVERTED, StatusEnum.SUCCESS, StatusEnum.CANCELLED, StatusEnum.OBSOLETED]: 
         if not is_alive(getattr(StatusEnum, tx_dict['status']).value):
             raise TxStateChangeError('Cannot resend mined or obsoleted transaction'.format(txold_hash_hex))
         
@@ -226,9 +248,6 @@ class AdminApi:
                 break
             last_nonce = nonce_otx
 
-        #nonce_cache = Nonce.get(address)
-        #nonce_w3 = self.w3.eth.getTransactionCount(address, 'pending') 
-        
         return {
             'nonce': {
                 #'network': nonce_cache,
@@ -272,20 +291,6 @@ class AdminApi:
         return s_nonce.apply_async()
 
 
-#    # TODO: this is a stub, complete all checks
-#    def ready(self):
-#        """Checks whether all required initializations have been performed.
-#
-#        :raises cic_eth.error.InitializationError: At least one setting pre-requisite has not been met.
-#        :raises KeyError: An address provided for initialization is not known by the keystore.
-#        """
-#        addr = AccountRole.get_address('ETH_GAS_PROVIDER_ADDRESS')
-#        if addr == ZERO_ADDRESS:
-#            raise InitializationError('missing account ETH_GAS_PROVIDER_ADDRESS')
-#
-#        self.w3.eth.sign(addr, text='666f6f')
-
-
     def account(self, chain_spec, address, include_sender=True, include_recipient=True, renderer=None, w=sys.stdout):
         """Lists locally originated transactions for the given Ethereum address.
 
@@ -348,6 +353,7 @@ class AdminApi:
 
 
     # TODO: Add exception upon non-existent tx aswell as invalid tx data to docstring 
+    # TODO: This method is WAY too long
     def tx(self, chain_spec, tx_hash=None, tx_raw=None, registry=None, renderer=None, w=sys.stdout):
         """Output local and network details about a given transaction with local origin.
 
@@ -370,7 +376,6 @@ class AdminApi:
 
         if tx_raw != None:
             tx_hash = add_0x(keccak256_hex_to_hex(tx_raw))
-            #tx_hash = self.w3.keccak(hexstr=tx_raw).hex()
 
         s = celery.signature(
             'cic_eth.queue.query.get_tx_cache',
@@ -386,38 +391,78 @@ class AdminApi:
   
         source_token = None
         if tx['source_token'] != ZERO_ADDRESS:
-            try:
+            if registry != None:
-                source_token = registry.by_address(tx['source_token'])
+                try:
-                #source_token = CICRegistry.get_address(chain_spec, tx['source_token']).contract
+                    source_token = registry.by_address(tx['source_token'])
-            except UnknownContractError:
+                except UnknownContractError:
-                #source_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
+                    logg.warning('unknown source token contract {} (direct)'.format(tx['source_token']))
-                #source_token = CICRegistry.add_token(chain_spec, source_token_contract)
+            else:
-                logg.warning('unknown source token contract {}'.format(tx['source_token']))
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['source_token'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                source_token = t.get()
+                if source_token == None:
+                    logg.warning('unknown source token contract {} (task pool)'.format(tx['source_token']))
+
 
         destination_token = None
-        if tx['source_token'] != ZERO_ADDRESS:
+        if tx['destination_token'] != ZERO_ADDRESS:
-            try:
+            if registry != None:
-                #destination_token = CICRegistry.get_address(chain_spec, tx['destination_token'])
+                try:
-                destination_token = registry.by_address(tx['destination_token'])
+                    destination_token = registry.by_address(tx['destination_token'])
-            except UnknownContractError:
+                except UnknownContractError:
-                #destination_token_contract = self.w3.eth.contract(abi=CICRegistry.abi('ERC20'), address=tx['source_token'])
+                    logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
-                #destination_token = CICRegistry.add_token(chain_spec, destination_token_contract)
+            else:
-                logg.warning('unknown destination token contract {}'.format(tx['destination_token']))
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['destination_token'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                destination_token = t.get()
+                if destination_token == None:
+                    logg.warning('unknown destination token contract {} (task pool)'.format(tx['destination_token']))
+
 
         tx['sender_description'] = 'Custodial account'
         tx['recipient_description'] = 'Custodial account'
 
         o = code(tx['sender'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         if len(strip_0x(r, allow_empty=True)) > 0:
-            try:
+            if registry != None:
-                #sender_contract = CICRegistry.get_address(chain_spec, tx['sender'])
+                try:
-                sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address)
+                    sender_contract = registry.by_address(tx['sender'], sender_address=self.call_address)
-                tx['sender_description'] = 'Contract at {}'.format(tx['sender']) #sender_contract)
+                    tx['sender_description'] = 'Contract at {}'.format(tx['sender'])
-            except UnknownContractError:
+                except UnknownContractError:
-                tx['sender_description'] = 'Unknown contract'
+                    tx['sender_description'] = 'Unknown contract'
-            except KeyError as e:
+                except KeyError as e:
-                tx['sender_description'] = 'Unknown contract'
+                    tx['sender_description'] = 'Unknown contract'
+            else:
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['sender'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                tx['sender_description'] = t.get()
+                if tx['sender_description'] == None:
+                    tx['sender_description'] = 'Unknown contract'
+
+
         else:
             s = celery.signature(
                     'cic_eth.eth.account.have',
@@ -446,16 +491,31 @@ class AdminApi:
                     tx['sender_description'] = role
 
         o = code(tx['recipient'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         if len(strip_0x(r, allow_empty=True)) > 0:
-            try:
+            if registry != None:
-                #recipient_contract = CICRegistry.by_address(tx['recipient'])
+                try:
-                recipient_contract = registry.by_address(tx['recipient'])
+                    recipient_contract = registry.by_address(tx['recipient'])
-                tx['recipient_description'] = 'Contract at {}'.format(tx['recipient']) #recipient_contract)
+                    tx['recipient_description'] = 'Contract at {}'.format(tx['recipient'])
-            except UnknownContractError as e:
+                except UnknownContractError as e:
-                tx['recipient_description'] = 'Unknown contract'
+                    tx['recipient_description'] = 'Unknown contract'
-            except KeyError as e:
+                except KeyError as e:
-                tx['recipient_description'] = 'Unknown contract'
+                    tx['recipient_description'] = 'Unknown contract'
+            else:
+                s = celery.signature(
+                        'cic_eth.task.registry_address_lookup',
+                        [
+                            chain_spec.asdict(),
+                            tx['recipient'],
+                            ],
+                        queue=self.queue
+                        )
+                t = s.apply_async()
+                tx['recipient_description'] = t.get()
+                if tx['recipient_description'] == None:
+                    tx['recipient_description'] = 'Unknown contract'
+
         else:
             s = celery.signature(
                     'cic_eth.eth.account.have',
@@ -497,7 +557,8 @@ class AdminApi:
         r = None
         try:
             o = transaction(tx_hash)
-            r = self.rpc.do(o)
+            t = self.proxy_do(chain_spec, o)
+            r = t.get()
             if r != None:
                 tx['network_status'] = 'Mempool'
         except Exception as e:
@@ -506,7 +567,8 @@ class AdminApi:
         if r != None:
             try:
                 o = receipt(tx_hash)
-                r = self.rpc.do(o)
+                t = self.proxy_do(chain_spec, o)
+                r = t.get()
                 logg.debug('h {} o {}'.format(tx_hash, o))
                 if int(strip_0x(r['status'])) == 1:
                     tx['network_status'] = 'Confirmed'
@@ -521,11 +583,13 @@ class AdminApi:
                 pass
 
         o = balance(tx['sender'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         tx['sender_gas_balance'] = r
 
         o = balance(tx['recipient'])
-        r = self.rpc.do(o)
+        t = self.proxy_do(chain_spec, o)
+        r = t.get()
         tx['recipient_gas_balance'] = r
 
         tx_unpacked = unpack(bytes.fromhex(strip_0x(tx['signed_tx'])), chain_spec)

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/redis.py

@@ -0,0 +1,18 @@
+# external imports
+import redis
+import os
+
+
+def health(*args, **kwargs):
+    r = redis.Redis(
+            host=kwargs['config'].get('REDIS_HOST'),
+            port=kwargs['config'].get('REDIS_PORT'),
+            db=kwargs['config'].get('REDIS_DB'),
+            )
+    try:
+        r.set(kwargs['unit'], os.getpid())
+    except redis.connection.ConnectionError:
+        return False
+    except redis.connection.ResponseError:
+        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
+    INIT=2
-    SEND=4
+    CREATE=4
-    QUEUE=8
+    SEND=8
-    QUERY=16
+    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/error.py

@@ -48,6 +48,8 @@ class RoleMissingError(Exception):
     pass
 
 
+
+
 class IntegrityError(Exception):
     """Exception raised to signal irregularities with deduplication and ordering of tasks
 
@@ -62,15 +64,19 @@ class LockedError(Exception):
     pass
 
 
-class SignerError(Exception):
+class SeppukuError(Exception):
+    """Exception base class for all errors that should cause system shutdown
+
+    """
+
+
+class SignerError(SeppukuError):
     """Exception raised when signer is unavailable or generates an error
 
     """
     pass
 
 
-class EthError(Exception):
+class RoleAgencyError(SeppukuError):
-    """Exception raised when unspecified error from evm node is encountered
+    """Exception raise when a role cannot perform its function. This is a critical exception
-
     """
-    pass

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

@@ -4,10 +4,10 @@ import logging
 # external imports
 import celery
 from erc20_single_shot_faucet import SingleShotFaucet as Faucet
-from chainlib.eth.constant import ZERO_ADDRESS
 from hexathon import (
         strip_0x,
         )
+from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.connection import RPCConnection
 from chainlib.eth.sign import (
         new_account,
@@ -19,6 +19,7 @@ from chainlib.eth.tx import (
         unpack,
         )
 from chainlib.chain import ChainSpec
+from chainlib.error import JSONRPCException
 from eth_accounts_index import AccountRegistry
 from sarafu_faucet import MinterFaucet as Faucet
 from chainqueue.db.models.tx import TxCache
@@ -70,11 +71,18 @@ def create(self, password, chain_spec_dict):
     a = None
     conn = RPCConnection.connect(chain_spec, 'signer')
     o = new_account()
-    a = conn.do(o)
+    try:
+        a = conn.do(o)
+    except ConnectionError as e:
+        raise SignerError(e)
+    except FileNotFoundError as e:
+        raise SignerError(e)
     conn.disconnect()
 
+    # TODO: It seems infeasible that a can be None in any case, verify
     if a == None:
         raise SignerError('create account')
+        
     logg.debug('created account {}'.format(a))
 
     # Initialize nonce provider record for account
@@ -219,21 +227,22 @@ def have(self, account, chain_spec_dict):
     """
     chain_spec = ChainSpec.from_dict(chain_spec_dict)
     o = sign_message(account, '0x2a')
-    try:
+    conn = RPCConnection.connect(chain_spec, 'signer')
-        conn = RPCConnection.connect(chain_spec, 'signer')
-    except Exception as e:
-        logg.debug('cannot sign with {}: {}'.format(account, e))
-        return None
 
     try:
         conn.do(o)
-        conn.disconnect()
+    except ConnectionError as e:
-        return account
+        raise SignerError(e)
-    except Exception as e:
+    except FileNotFoundError as e:
+        raise SignerError(e)
+    except JSONRPCException as e:
         logg.debug('cannot sign with {}: {}'.format(account, e))
         conn.disconnect()
         return None
    
+    conn.disconnect()
+    return account
+
 
 @celery_app.task(bind=True, base=CriticalSQLAlchemyTask)
 def set_role(self, tag, address, chain_spec_dict):

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

@@ -108,7 +108,13 @@ def transfer(self, tokens, holder_address, receiver_address, value, chain_spec_d
     nonce_oracle = CustodialTaskNonceOracle(holder_address, self.request.root_id, session=session)
     gas_oracle = self.create_gas_oracle(rpc, MaxGasOracle.gas)
     c = ERC20(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
-    (tx_hash_hex, tx_signed_raw_hex) = c.transfer(t['address'], holder_address, receiver_address, value, tx_format=TxFormat.RLP_SIGNED)
+    try:
+        (tx_hash_hex, tx_signed_raw_hex) = c.transfer(t['address'], holder_address, receiver_address, value, tx_format=TxFormat.RLP_SIGNED)
+    except FileNotFoundError as e:
+        raise SignerError(e)
+    except ConnectionError as e:
+        raise SignerError(e)
+    
 
     rpc_signer.disconnect()
     rpc.disconnect()
@@ -171,7 +177,12 @@ def approve(self, tokens, holder_address, spender_address, value, chain_spec_dic
     nonce_oracle = CustodialTaskNonceOracle(holder_address, self.request.root_id, session=session)
     gas_oracle = self.create_gas_oracle(rpc, MaxGasOracle.gas)
     c = ERC20(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle, nonce_oracle=nonce_oracle)
-    (tx_hash_hex, tx_signed_raw_hex) = c.approve(t['address'], holder_address, spender_address, value, tx_format=TxFormat.RLP_SIGNED)
+    try:
+        (tx_hash_hex, tx_signed_raw_hex) = c.approve(t['address'], holder_address, spender_address, value, tx_format=TxFormat.RLP_SIGNED)
+    except FileNotFoundError as e:
+        raise SignerError(e)
+    except ConnectionError as e:
+        raise SignerError(e)
 
     rpc_signer.disconnect()
     rpc.disconnect()

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

@@ -328,7 +328,12 @@ def refill_gas(self, recipient_address, chain_spec_dict):
 
     # build and add transaction
     logg.debug('tx send gas amount {} from provider {} to {}'.format(refill_amount, gas_provider, recipient_address))
-    (tx_hash_hex, tx_signed_raw_hex) = c.create(gas_provider, recipient_address, refill_amount, tx_format=TxFormat.RLP_SIGNED)
+    try:
+        (tx_hash_hex, tx_signed_raw_hex) = c.create(gas_provider, recipient_address, refill_amount, tx_format=TxFormat.RLP_SIGNED)
+    except ConnectionError as e:
+        raise SignerError(e)
+    except FileNotFoundError as e:
+        raise SignerError(e)
     logg.debug('adding queue refill gas tx {}'.format(tx_hash_hex))
     cache_task = 'cic_eth.eth.gas.cache_gas_data'
     register_tx(tx_hash_hex, tx_signed_raw_hex, chain_spec, queue, cache_task=cache_task, session=session)
@@ -404,7 +409,12 @@ def resend_with_higher_gas(self, txold_hash_hex, chain_spec_dict, gas=None, defa
     c = TxFactory(chain_spec, signer=rpc_signer, gas_oracle=gas_oracle)
     logg.debug('change gas price from old {} to new {} for tx {}'.format(tx['gasPrice'], new_gas_price, tx))
     tx['gasPrice'] = new_gas_price
-    (tx_hash_hex, tx_signed_raw_hex) = c.build_raw(tx)
+    try:
+        (tx_hash_hex, tx_signed_raw_hex) = c.build_raw(tx)
+    except ConnectionError as e:
+        raise SignerError(e)
+    except FileNotFoundError as e:
+        raise SignerError(e)
     queue_create(
         chain_spec,
         tx['nonce'],

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.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)
+            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)
+        st = ERC20Token(self.chain_spec, self.rpc, source)
-        dt = ERC20Token(self.rpc, destination)
+        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/ext/tx.py

@@ -114,7 +114,7 @@ def list_tx_by_bloom(self, bloomspec, address, chain_spec_dict):
 
                         # TODO: pass through registry to validate declarator entry of token
                         #token = registry.by_address(tx['to'], sender_address=self.call_address)
-                        token = ERC20Token(rpc, tx['to'])
+                        token = ERC20Token(chain_spec, rpc, tx['to'])
                         token_symbol = token.symbol
                         token_decimals = token.decimals
                         times = tx_times(tx['hash'], chain_spec)

apps/cic-eth/cic_eth/queue/time.py

@@ -12,6 +12,7 @@ from chainqueue.error import NotLocalTxError
 
 # local imports
 from cic_eth.task import CriticalSQLAlchemyAndWeb3Task
+from cic_eth.db.models.base import SessionBase
 
 celery_app = celery.current_app
 

apps/cic-eth/cic_eth/registry.py

@@ -29,5 +29,5 @@ def connect(rpc, chain_spec, registry_address):
     CICRegistry.address = registry_address
     registry = CICRegistry(chain_spec, rpc)
     registry_address = registry.by_name('ContractRegistry')
-    
     return registry
+

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

@@ -23,7 +23,6 @@ default_config_dir = os.environ.get('CONFINI_DIR', '/usr/local/etc/cic')
 
 argparser = argparse.ArgumentParser()
 argparser.add_argument('-p', '--provider', dest='p', default='http://localhost:8545', type=str, help='Web3 provider url (http only)')
-argparser.add_argument('-r', '--registry-address', type=str, help='CIC registry address')
 argparser.add_argument('-f', '--format', dest='f', default=default_format, type=str, help='Output format')
 argparser.add_argument('-c', type=str, default=default_config_dir, help='config root to use')
 argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')
@@ -59,6 +58,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 +67,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 +84,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 +96,7 @@ def main():
             'cic_eth.admin.ctrl.unlock',
             [
                 None,
-                chain_spec.asdict(),
+                chain_spec_dict,
                 args.address,
                 flags,
                 ],
@@ -110,7 +115,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')))
-        syncer.loop(conn, float(config.get('DISPATCHER_LOOP_INTERVAL')))
-    except SyncDone as e:
-        sys.stderr.write("dispatcher done at block {}\n".format(e))
 
     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,
+                self.parse_transfer,
-                parse_transferfrom,
+                self.parse_transferfrom,
-                parse_giftto,
+                self.parse_giftto,
                 ]:
             try:
-                (transfer_type, transfer_data) = parser(tx)
+                if tx:
-                break
+                    (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())
+                result = tokentx.asdict()
-                logg.info('callback success task id {} tx {}'.format(t, tx.hash))
+                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/tasker.py

@@ -11,10 +11,19 @@ import websocket
 # external imports
 import celery
 import confini
-from chainlib.connection import RPCConnection
+from chainlib.connection import (
-from chainlib.eth.connection import EthUnixSignerConnection
+        RPCConnection,
+        ConnType,
+        )
+from chainlib.eth.connection import (
+        EthUnixSignerConnection,
+        EthHTTPSignerConnection,
+        )
 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 (
@@ -39,6 +48,7 @@ from cic_eth.queue import (
 from cic_eth.callbacks import (
         Callback,
         http,
+        noop,
         #tcp,
         redis,
         )
@@ -50,6 +60,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 +73,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 +93,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 +103,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__)
@@ -133,11 +148,18 @@ else:
         })
 
 chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
+RPCConnection.register_constructor(ConnType.UNIX, EthUnixSignerConnection, 'signer')
+RPCConnection.register_constructor(ConnType.HTTP, EthHTTPSignerConnection, 'signer')
+RPCConnection.register_constructor(ConnType.HTTP_SSL, EthHTTPSignerConnection, 'signer')
 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')
 
 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, unit='cic-eth-tasker')
 
 def main():
     argv = ['worker']
@@ -161,7 +183,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 +196,18 @@ 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)
+    BaseTask.run_dir = config.get('CIC_RUN_DIR')
+    logg.info('default token set to {} {}'.format(BaseTask.default_token_symbol, BaseTask.default_token_address))
+   
+    liveness.linux.set(rundir=config.get('CIC_RUN_DIR'))
     current_app.worker_main(argv)
+    liveness.linux.reset(rundir=config.get('CIC_RUN_DIR'))
 
 
 @celery.signals.eventlet_pool_postshutdown.connect

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

@@ -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,
@@ -43,6 +42,12 @@ from cic_eth.runnable.daemons.filters import (
         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__))
 
@@ -88,18 +93,18 @@ def main():
 
     syncers = []
 
-    #if SyncerBackend.first(chain_spec):
+    #if SQLBackend.first(chain_spec):
-    #    backend = SyncerBackend.initial(chain_spec, block_offset)
+    #    backend = SQLBackend.initial(chain_spec, block_offset)
-    syncer_backends = SyncerBackend.resume(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:
@@ -109,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')
@@ -116,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 = []

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

@@ -0,0 +1,65 @@
+#!python3
+
+# SPDX-License-Identifier: GPL-3.0-or-later
+
+# standard imports
+import logging
+import argparse
+import os
+
+# external imports
+import confini
+import celery
+
+# local imports
+from cic_eth.api import Api
+
+logging.basicConfig(level=logging.WARNING)
+logg = logging.getLogger()
+
+default_format = 'terminal'
+default_config_dir = os.environ.get('CONFINI_DIR', '/usr/local/etc/cic')
+
+
+argparser = argparse.ArgumentParser()
+argparser.add_argument('-i', '--chain-spec', dest='i', type=str, help='chain spec')
+argparser.add_argument('-c', type=str, default=default_config_dir, help='config root to use')
+argparser.add_argument('-q', type=str, default='cic-eth', help='celery queue to submit transaction tasks to')
+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', help='be more verbose', action='store_true')
+args = argparser.parse_args()
+
+if args.v == True:
+    logging.getLogger().setLevel(logging.INFO)
+elif args.vv == True:
+    logging.getLogger().setLevel(logging.DEBUG)
+
+config_dir = os.path.join(args.c)
+os.makedirs(config_dir, 0o777, True)
+config = confini.Config(config_dir, args.env_prefix)
+config.process()
+args_override = {
+        'CIC_CHAIN_SPEC': getattr(args, 'i'),
+        }
+config.dict_override(args_override, 'cli args')
+config.censor('PASSWORD', 'DATABASE')
+config.censor('PASSWORD', 'SSL')
+logg.debug('config loaded from {}:\n{}'.format(config_dir, config))
+
+
+celery_app = celery.Celery(broker=config.get('CELERY_BROKER_URL'), backend=config.get('CELERY_RESULT_URL'))
+
+queue = args.q
+
+api = Api(config.get('CIC_CHAIN_SPEC'), queue=queue)
+
+def main():
+    t = api.default_token()
+    token_info = t.get()
+    print('Default token symbol: {}'.format(token_info['symbol']))
+    print('Default token address: {}'.format(token_info['address']))
+
+
+if __name__ == '__main__':
+    main()

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

@@ -85,9 +85,6 @@ def main():
             callback_queue=args.q,
             )
 
-    #register = not args.no_register
-    #logg.debug('register {}'.format(register))
-    #t = api.create_account(register=register)
     t = api.transfer(config.get('_SENDER'), config.get('_RECIPIENT'), config.get('_VALUE'), config.get('_SYMBOL'))
 
     ps.get_message()

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

@@ -81,10 +81,14 @@ chain_spec = ChainSpec.from_chain_str(config.get('CIC_CHAIN_SPEC'))
 
 rpc = EthHTTPConnection(args.p)
 
-registry_address = config.get('CIC_REGISTRY_ADDRESS')
+#registry_address = config.get('CIC_REGISTRY_ADDRESS')
 
 admin_api = AdminApi(rpc)
 
+t = admin_api.registry()
+registry_address = t.get()
+logg.info('got registry address from task pool: {}'.format(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')
@@ -151,14 +155,16 @@ def main():
     txs  = []
     renderer = render_tx
     if len(config.get('_QUERY')) > 66:
-        registry = connect_registry(rpc, chain_spec, registry_address)
+        #registry = connect_registry(rpc, chain_spec, registry_address)
-        admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), registry=registry, renderer=renderer)
+        #admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), registry=registry, renderer=renderer)
+        admin_api.tx(chain_spec, tx_raw=config.get('_QUERY'), renderer=renderer)
     elif len(config.get('_QUERY')) > 42:
-        registry = connect_registry(rpc, chain_spec, registry_address)
+        #registry = connect_registry(rpc, chain_spec, registry_address)
-        admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), registry=registry, renderer=renderer)
+        #admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), registry=registry, renderer=renderer)
+        admin_api.tx(chain_spec, tx_hash=config.get('_QUERY'), renderer=renderer)
 
     elif len(config.get('_QUERY')) == 42:
-        registry = connect_registry(rpc, chain_spec, registry_address)
+        #registry = connect_registry(rpc, chain_spec, registry_address)
         txs = admin_api.account(chain_spec, config.get('_QUERY'), include_recipient=False, renderer=render_account)
         renderer = render_account
     elif len(config.get('_QUERY')) >= 4 and config.get('_QUERY')[:4] == 'lock':

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

@@ -7,18 +7,20 @@ import uuid
 # external imports
 import celery
 import sqlalchemy
+from chainlib.chain import ChainSpec
+from chainlib.connection import RPCConnection
 from chainlib.eth.constant import ZERO_ADDRESS
 from chainlib.eth.nonce import RPCNonceOracle
 from chainlib.eth.gas import RPCGasOracle
+from cic_eth_registry import CICRegistry
+from cic_eth_registry.error import UnknownContractError
+import liveness.linux
 
 # local imports
-from cic_eth.error import (
+from cic_eth.error import SeppukuError
-        SignerError,
-        EthError,
-        )
 from cic_eth.db.models.base import SessionBase
 
-logg = logging.getLogger(__name__)
+logg = logging.getLogger().getChild(__name__)
 
 celery_app = celery.current_app
 
@@ -29,6 +31,9 @@ class BaseTask(celery.Task):
     call_address = ZERO_ADDRESS
     create_nonce_oracle = RPCNonceOracle
     create_gas_oracle = RPCGasOracle
+    default_token_address = None
+    default_token_symbol = None
+    run_dir = '/run'
 
     def create_session(self):
         return BaseTask.session_func()
@@ -38,6 +43,19 @@ class BaseTask(celery.Task):
         logg.debug('task {} root uuid {}'.format(self.__class__.__name__, self.request.root_id))
         return
 
+
+    def on_failure(self, exc, task_id, args, kwargs, einfo):
+        if isinstance(exc, SeppukuError):
+            liveness.linux.reset(rundir=self.run_dir)
+            logg.critical(einfo)
+            msg = 'received critical exception {}, calling shutdown'.format(str(exc))
+            s = celery.signature(
+                'cic_eth.admin.ctrl.shutdown',
+                [msg],
+                queue=self.request.delivery_info.get('routing_key'),
+                    )
+            s.apply_async()
+
     
 class CriticalTask(BaseTask):
     retry_jitter = True
@@ -67,7 +85,6 @@ class CriticalSQLAlchemyAndWeb3Task(CriticalTask):
         sqlalchemy.exc.TimeoutError,
         requests.exceptions.ConnectionError,
         sqlalchemy.exc.ResourceClosedError,
-        EthError,
         )
     safe_gas_threshold_amount = 2000000000 * 60000 * 3
     safe_gas_refill_amount = safe_gas_threshold_amount * 5 
@@ -78,19 +95,45 @@ class CriticalSQLAlchemyAndSignerTask(CriticalTask):
         sqlalchemy.exc.DatabaseError,
         sqlalchemy.exc.TimeoutError,
         sqlalchemy.exc.ResourceClosedError,
-        SignerError,
         ) 
 
 class CriticalWeb3AndSignerTask(CriticalTask):
     autoretry_for = (
         requests.exceptions.ConnectionError,
-        SignerError,
         )
     safe_gas_threshold_amount = 2000000000 * 60000 * 3
     safe_gas_refill_amount = safe_gas_threshold_amount * 5 
 
 
-@celery_app.task(bind=True, base=BaseTask)
+@celery_app.task()
-def hello(self):
+def check_health(self):
-    time.sleep(0.1)
+    pass
-    return id(SessionBase.create_session)
+
+
+# TODO: registry / rpc methods should perhaps be moved to better named module
+@celery_app.task()
+def registry():
+    return CICRegistry.address
+
+
+@celery_app.task()
+def registry_address_lookup(chain_spec_dict, address, connection_tag='default'):
+    chain_spec = ChainSpec.from_dict(chain_spec_dict)
+    conn = RPCConnection.connect(chain_spec, tag=connection_tag)
+    registry = CICRegistry(chain_spec, conn)
+    return registry.by_address(address)
+
+
+@celery_app.task(throws=(UnknownContractError,))
+def registry_name_lookup(chain_spec_dict, name, connection_tag='default'):
+    chain_spec = ChainSpec.from_dict(chain_spec_dict)
+    conn = RPCConnection.connect(chain_spec, tag=connection_tag)
+    registry = CICRegistry(chain_spec, conn)
+    return registry.by_name(name)
+
+
+@celery_app.task()
+def rpc_proxy(chain_spec_dict, o, connection_tag='default'):
+    chain_spec = ChainSpec.from_dict(chain_spec_dict)
+    conn = RPCConnection.connect(chain_spec, tag=connection_tag)
+    return conn.do(o)

apps/cic-eth/cic_eth/version.py

@@ -10,7 +10,7 @@ version = (
         0,
         11,
         0,
-        'beta.1',
+        'beta.11',
         )
 
 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.redis,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

@@ -3,3 +3,6 @@ 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.redis,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,3 @@
 [eth]
-#ws_provider = ws://localhost:8546
-#ttp_provider = http://localhost:8545
 provider = http://localhost:63545
-gas_provider_address = 
+gas_gifter_minimum_balance = 10000000000000000000000
-#chain_id = 
-abi_dir = /home/lash/src/ext/cic/grassrootseconomics/cic-contracts/abis
-account_accounts_index_writer = 

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/eth.ini

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

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.2a62
+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 ./
@@ -50,6 +50,4 @@ COPY cic-eth/config/ /usr/local/etc/cic-eth/
 COPY cic-eth/cic_eth/db/migrations/ /usr/local/share/cic-eth/alembic/
 COPY cic-eth/crypto_dev_signer_config/ /usr/local/etc/crypto-dev-signer/
 
-RUN git clone https://gitlab.com/grassrootseconomics/cic-contracts.git && \
+COPY util/liveness/health.sh /usr/local/bin/health.sh
-	mkdir -p /usr/local/share/cic/solidity && \
-	cp -R cic-contracts/abis /usr/local/share/cic/solidity/abi

apps/cic-eth/requirements.txt

@@ -1,25 +1,25 @@
-cic-base~=0.1.2a62
+cic-base==0.1.2b5
 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
+eth_accounts_index~=0.0.11a9
-erc20-transfer-authorization~=0.3.1a3
+erc20-transfer-authorization~=0.3.1a5
-#simple-rlp==0.1.2
 uWSGI==2.0.19.1
 semver==2.13.0
 websocket-client==0.57.0
 moolb~=0.1.1b2
-eth-address-index~=0.1.1a7
+eth-address-index~=0.1.1a9
-chainlib~=0.0.2a5
+chainlib~=0.0.2a20
 hexathon~=0.0.1a7
-chainsyncer~=0.0.1a21
+chainsyncer[sql]~=0.0.2a2
-chainqueue~=0.0.1a7
+chainqueue~=0.0.2a2
 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/conftest.py

@@ -3,8 +3,12 @@ import os
 import sys
 import logging
 
+# external imports
+from chainlib.eth.erc20 import ERC20
+
 # local imports
 from cic_eth.api import Api
+from cic_eth.task import BaseTask
 
 script_dir = os.path.dirname(os.path.realpath(__file__))
 root_dir = os.path.dirname(script_dir)
@@ -28,3 +32,26 @@ def api(
     ):
     chain_str = str(default_chain_spec)
     return Api(chain_str, queue=None, callback_param='foo')
+
+
+@pytest.fixture(scope='function')
+def foo_token_symbol(
+    default_chain_spec,
+    foo_token,
+    eth_rpc,
+    contract_roles,
+    ):
+    
+    c = ERC20(default_chain_spec)
+    o = c.symbol(foo_token, sender_address=contract_roles['CONTRACT_DEPLOYER'])
+    r = eth_rpc.do(o)
+    return c.parse_symbol(r)
+
+
+@pytest.fixture(scope='function')
+def default_token(
+        foo_token,
+        foo_token_symbol,
+        ):
+    BaseTask.default_token_symbol = foo_token_symbol
+    BaseTask.default_token_address = foo_token   

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

@@ -0,0 +1,225 @@
+# 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 = {}
+            self.queue = 'test'
+
+        def call_back(self, transfer_type, result):
+            self.results[transfer_type] = result
+            return self
+
+    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-eth/tests/filters/test_tx_filter.py

@@ -65,6 +65,7 @@ def test_tx(
     tx_hash_hex_orig = tx_hash_hex
 
     gas_oracle = OverrideGasOracle(price=1100000000, limit=21000)
+    c = Gas(default_chain_spec, signer=eth_signer, nonce_oracle=nonce_oracle, gas_oracle=gas_oracle)
     (tx_hash_hex, tx_signed_raw_hex) = c.create(agent_roles['ALICE'], agent_roles['BOB'], 100 * (10 ** 6), tx_format=TxFormat.RLP_SIGNED)
     queue_create(
             default_chain_spec,

apps/cic-eth/tests/fixtures_celery.py

@@ -34,6 +34,7 @@ def celery_includes():
         'cic_eth.admin.ctrl',
         'cic_eth.admin.nonce',
         'cic_eth.admin.debug',
+        'cic_eth.admin.token',
         'cic_eth.eth.account',
         'cic_eth.callbacks.noop',
         'cic_eth.callbacks.http',

apps/cic-eth/tests/fixtures_database.py

@@ -53,6 +53,9 @@ def init_database(
     alembic.command.downgrade(ac, 'base')
     alembic.command.upgrade(ac, 'head')
 
+    session.execute('DELETE FROM lock')
+    session.commit()
+
     yield session
     session.commit()
     session.close()

apps/cic-eth/tests/task/api/test_admin.py

@@ -273,7 +273,7 @@ def test_tx(
     eth_signer,
     agent_roles,
     contract_roles,
-    celery_worker,
+    celery_session_worker,
     ):
 
     nonce_oracle = RPCNonceOracle(agent_roles['ALICE'], eth_rpc)

apps/cic-eth/tests/task/api/test_list.py

@@ -35,7 +35,7 @@ def test_list_tx(
         foo_token,
         register_tokens,
         init_eth_tester,
-        celery_worker,
+        celery_session_worker,
         ):
 
     tx_hashes = []

apps/cic-eth/tests/unit/admin/test_default_token.py

@@ -0,0 +1,21 @@
+# external imports
+import celery
+
+
+def test_default_token(
+        default_token,
+        celery_session_worker,
+        foo_token,
+        foo_token_symbol,
+        ):
+      
+    s = celery.signature(
+            'cic_eth.admin.token.default_token',
+            [],
+            queue=None,
+            )
+    t = s.apply_async()
+    r = t.get()
+
+    assert r['address'] == foo_token
+    assert r['symbol'] == foo_token_symbol

apps/cic-meta/.gitignore

@@ -3,4 +3,3 @@ dist
 dist-web
 dist-server
 scratch
-tests

apps/cic-meta/package-lock.json

@@ -1,6 +1,6 @@
 {
 	"name": "cic-client-meta",
-	"version": "0.0.7-alpha.2",
+	"version": "0.0.7-alpha.7",
 	"lockfileVersion": 1,
 	"requires": true,
 	"dependencies": {
@@ -852,6 +852,75 @@
 				"printj": "~1.1.0"
 			}
 		},
+		"crdt-meta": {
+			"version": "0.0.8",
+			"resolved": "https://registry.npmjs.org/crdt-meta/-/crdt-meta-0.0.8.tgz",
+			"integrity": "sha512-CS0sS0L2QWthz7vmu6vzl3p4kcpJ+IKILBJ4tbgN4A3iNG8wnBeuDIv/z3KFFQjcfuP4QAh6E9LywKUTxtDc3g==",
+			"requires": {
+				"automerge": "^0.14.2",
+				"ini": "^1.3.8",
+				"openpgp": "^4.10.8",
+				"pg": "^8.5.1",
+				"sqlite3": "^5.0.2"
+			},
+			"dependencies": {
+				"automerge": {
+					"version": "0.14.2",
+					"resolved": "https://registry.npmjs.org/automerge/-/automerge-0.14.2.tgz",
+					"integrity": "sha512-shiwuJHCbNRI23WZyIECLV4Ovf3WiAFJ7P9BH4l5gON1In/UUbjcSJKRygtIirObw2UQumeYxp3F2XBdSvQHnA==",
+					"requires": {
+						"immutable": "^3.8.2",
+						"transit-immutable-js": "^0.7.0",
+						"transit-js": "^0.8.861",
+						"uuid": "^3.4.0"
+					}
+				},
+				"node-addon-api": {
+					"version": "3.1.0",
+					"resolved": "https://registry.npmjs.org/node-addon-api/-/node-addon-api-3.1.0.tgz",
+					"integrity": "sha512-flmrDNB06LIl5lywUz7YlNGZH/5p0M7W28k8hzd9Lshtdh1wshD2Y+U4h9LD6KObOy1f+fEVdgprPrEymjM5uw=="
+				},
+				"pg": {
+					"version": "8.6.0",
+					"resolved": "https://registry.npmjs.org/pg/-/pg-8.6.0.tgz",
+					"integrity": "sha512-qNS9u61lqljTDFvmk/N66EeGq3n6Ujzj0FFyNMGQr6XuEv4tgNTXvJQTfJdcvGit5p5/DWPu+wj920hAJFI+QQ==",
+					"requires": {
+						"buffer-writer": "2.0.0",
+						"packet-reader": "1.0.0",
+						"pg-connection-string": "^2.5.0",
+						"pg-pool": "^3.3.0",
+						"pg-protocol": "^1.5.0",
+						"pg-types": "^2.1.0",
+						"pgpass": "1.x"
+					}
+				},
+				"pg-connection-string": {
+					"version": "2.5.0",
+					"resolved": "https://registry.npmjs.org/pg-connection-string/-/pg-connection-string-2.5.0.tgz",
+					"integrity": "sha512-r5o/V/ORTA6TmUnyWZR9nCj1klXCO2CEKNRlVuJptZe85QuhFayC7WeMic7ndayT5IRIR0S0xFxFi2ousartlQ=="
+				},
+				"pg-pool": {
+					"version": "3.3.0",
+					"resolved": "https://registry.npmjs.org/pg-pool/-/pg-pool-3.3.0.tgz",
+					"integrity": "sha512-0O5huCql8/D6PIRFAlmccjphLYWC+JIzvUhSzXSpGaf+tjTZc4nn+Lr7mLXBbFJfvwbP0ywDv73EiaBsxn7zdg=="
+				},
+				"pg-protocol": {
+					"version": "1.5.0",
+					"resolved": "https://registry.npmjs.org/pg-protocol/-/pg-protocol-1.5.0.tgz",
+					"integrity": "sha512-muRttij7H8TqRNu/DxrAJQITO4Ac7RmX3Klyr/9mJEOBeIpgnF8f9jAfRz5d3XwQZl5qBjF9gLsUtMPJE0vezQ=="
+				},
+				"sqlite3": {
+					"version": "5.0.2",
+					"resolved": "https://registry.npmjs.org/sqlite3/-/sqlite3-5.0.2.tgz",
+					"integrity": "sha512-1SdTNo+BVU211Xj1csWa8lV6KM0CtucDwRyA0VHl91wEH1Mgh7RxUpI4rVvG7OhHrzCSGaVyW5g8vKvlrk9DJA==",
+					"requires": {
+						"node-addon-api": "^3.0.0",
+						"node-gyp": "3.x",
+						"node-pre-gyp": "^0.11.0"
+					}
+				}
+			}
+		},
 		"create-hash": {
 			"version": "1.2.0",
 			"resolved": "https://registry.npmjs.org/create-hash/-/create-hash-1.2.0.tgz",
@@ -966,17 +1035,17 @@
 			"dev": true
 		},
 		"elliptic": {
-			"version": "6.5.3",
+			"version": "6.5.4",
-			"resolved": "https://registry.npmjs.org/elliptic/-/elliptic-6.5.3.tgz",
+			"resolved": "https://registry.npmjs.org/elliptic/-/elliptic-6.5.4.tgz",
-			"integrity": "sha512-IMqzv5wNQf+E6aHeIqATs0tOLeOTwj1QKbRcS3jBbYkl5oLAserA8yJTT7/VyHUYG91PRmPyeQDObKLPpeS4dw==",
+			"integrity": "sha512-iLhC6ULemrljPZb+QutR5TQGB+pdW6KGD5RSegS+8sorOZT+rdQFbsQFJgvN3eRqNALqJer4oQ16YvJHlU8hzQ==",
 			"requires": {
-				"bn.js": "^4.4.0",
+				"bn.js": "^4.11.9",
-				"brorand": "^1.0.1",
+				"brorand": "^1.1.0",
 				"hash.js": "^1.0.0",
-				"hmac-drbg": "^1.0.0",
+				"hmac-drbg": "^1.0.1",
-				"inherits": "^2.0.1",
+				"inherits": "^2.0.4",
-				"minimalistic-assert": "^1.0.0",
+				"minimalistic-assert": "^1.0.1",
-				"minimalistic-crypto-utils": "^1.0.0"
+				"minimalistic-crypto-utils": "^1.0.1"
 			}
 		},
 		"emoji-regex": {
@@ -1489,9 +1558,9 @@
 			"integrity": "sha512-k/vGaX4/Yla3WzyMCvTQOXYeIHvqOKtnqBduzTHpzpQZzAskKMhZ2K+EnBiSM9zGSoIFeMpXKxa4dYeZIQqewQ=="
 		},
 		"ini": {
-			"version": "1.3.5",
+			"version": "1.3.8",
-			"resolved": "https://registry.npmjs.org/ini/-/ini-1.3.5.tgz",
+			"resolved": "https://registry.npmjs.org/ini/-/ini-1.3.8.tgz",
-			"integrity": "sha512-RZY5huIKCMRWDUqZlEi72f/lmXKMvuszcMBduliQ3nnWbx9X/ZBQO7DijMEYS9EhHBb2qacRUMtC7svLwe0lcw=="
+			"integrity": "sha512-JV/yugV2uzW5iMRSiZAyDtQd+nxtUnjeLt0acNdw98kKLrvuRVyB80tsREOE7yvGVgalhZ6RNXCmEHkUKBKxew=="
 		},
 		"interpret": {
 			"version": "2.2.0",
@@ -1957,9 +2026,9 @@
 					}
 				},
 				"y18n": {
-					"version": "4.0.0",
+					"version": "4.0.3",
-					"resolved": "https://registry.npmjs.org/y18n/-/y18n-4.0.0.tgz",
+					"resolved": "https://registry.npmjs.org/y18n/-/y18n-4.0.3.tgz",
-					"integrity": "sha512-r9S/ZyXu/Xu9q1tYlpsLIsa3EeLXXk0VwlxqTcFRfg9EhMW+17kbt9G0NrgCmhGb5vT2hyhJZLfDGx+7+5Uj/w==",
+					"integrity": "sha512-JKhqTOwSrqNA1NY5lSztJ1GrBiUodLMmIZuLiDaMRJ+itFd+ABVE8XBjOvIWL+rSqNDC74LCSFmlb/U4UZ4hJQ==",
 					"dev": true
 				},
 				"yargs": {

apps/cic-meta/package.json

@@ -1,6 +1,6 @@
 {
 	"name": "cic-client-meta",
-	"version": "0.0.7-alpha.3",
+	"version": "0.0.7-alpha.8",
 	"description": "Signed CRDT metadata graphs for the CIC network",
 	"main": "dist/index.js",
 	"types": "dist/index.d.ts",
@@ -15,8 +15,9 @@
 	"dependencies": {
 		"@ethereumjs/tx": "^3.0.0-beta.1",
 		"automerge": "^0.14.1",
+		"crdt-meta": "0.0.8",
 		"ethereumjs-wallet": "^1.0.1",
-		"ini": "^1.3.5",
+		"ini": "^1.3.8",
 		"openpgp": "^4.10.8",
 		"pg": "^8.4.2",
 		"sqlite3": "^5.0.0",
@@ -40,6 +41,6 @@
 	],
 	"license": "GPL-3.0-or-later",
 	"engines": {
-		"node": "~15.3.0"
+		"node": ">=14.16.1"
 	}
 }

apps/cic-meta/scripts/dumpconfig.js

@@ -1,4 +1,4 @@
-const config = require('./src/config');
+import { Config } from 'crdt-meta';
 const fs = require('fs');
 
 if (process.argv[2] === undefined) {
@@ -15,6 +15,6 @@ try {
 	process.exit(1);
 }
 
-const c = new config.Config(process.argv[2], process.env['CONFINI_ENV_PREFIX']);
+const c = new Config(process.argv[2], process.env['CONFINI_ENV_PREFIX']);
 c.process();
 process.stdout.write(c.toString());

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

@@ -1,8 +1,7 @@
 import * as Automerge from 'automerge';
 import * as pgp from 'openpgp';
-import * as pg from 'pg';
 
-import { Envelope, Syncable } from '../../src/sync';
+import { Envelope, Syncable } from 'crdt-meta';
 
 
 function handleNoMergeGet(db, digest, keystore) {

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

@@ -1,15 +1,11 @@
 import * as http from 'http';
 import * as fs from 'fs';
 import * as path from 'path';
-import * as pgp from 'openpgp';
 
 import * as handlers from './handlers';
-import { Envelope, Syncable } from '../../src/sync';
+import { PGPKeyStore, PGPSigner, Config, SqliteAdapter, PostgresAdapter } from 'crdt-meta';
-import { PGPKeyStore, PGPSigner } from '../../src/auth';
 
 import { standardArgs } from './args';
-import { Config } from '../../src/config';
-import { SqliteAdapter, PostgresAdapter } from '../../src/db';
 
 let configPath = '/usr/local/etc/cic-meta';
 
@@ -114,6 +110,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 +120,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/auth.ts

@@ -1,191 +0,0 @@
-import * as pgp from 'openpgp';
-import * as crypto from 'crypto';
-
-interface Signable {
-	digest():string;
-}
-
-type KeyGetter = () => any;
-
-type Signature = {
-	engine:string
-	algo:string
-	data:string
-	digest:string
-}
-
-interface Signer { 
-	prepare(Signable):boolean;
-	onsign(Signature):void;
-	onverify(boolean):void;
-	sign(digest:string):void
-	verify(digest:string, signature:Signature):void
-	fingerprint():string
-}
-
-interface Authoritative {
-}
-
-interface KeyStore {
-	getPrivateKey:		KeyGetter
-	getFingerprint:		() => string
-	getTrustedKeys: 	() => Array<any>
-	getTrustedActiveKeys: 	() => Array<any>
-	getEncryptKeys: 	() => Array<any>
-}
-
-class PGPKeyStore implements KeyStore {
-
-	fingerprint:	string
-	pk:		any
-
-	pubk = {
-		active: [],
-		trusted: [],
-		encrypt: [],
-	}
-	loads = 0x00;
-	loadsTarget = 0x0f;
-	onload: (k:KeyStore) => void;
-
-	constructor(passphrase:string, pkArmor:string, pubkActiveArmor:string, pubkTrustedArmor:string, pubkEncryptArmor:string, onload = (ks:KeyStore) => {}) {
-		this._readKey(pkArmor, undefined, 1, passphrase);
-		this._readKey(pubkActiveArmor, 'active', 2);
-		this._readKey(pubkTrustedArmor, 'trusted', 4);
-		this._readKey(pubkEncryptArmor, 'encrypt', 8);
-		this.onload = onload;
-	}
-
-	private _readKey(a:string, x:any, n:number, pass?:string) {
-		pgp.key.readArmored(a).then((k) => {
-			if (pass !== undefined) {
-				this.pk = k.keys[0];
-				this.pk.decrypt(pass).then(() => {
-					this.fingerprint = this.pk.getFingerprint();
-					console.log('private key (sign)', this.fingerprint);
-					this._registerLoad(n);
-				});
-			} else {
-				this.pubk[x] = k.keys;
-				k.keys.forEach((pubk) => {
-					console.log('public key (' + x + ')', pubk.getFingerprint());
-				});
-				this._registerLoad(n);
-			}
-		});
-	}
-
-	private _registerLoad(b:number) {
-		this.loads |= b;
-		if (this.loads == this.loadsTarget) {
-			this.onload(this);
-		}
-	}
-
-	public getTrustedKeys(): Array<any> {
-		return this.pubk['trusted'];
-	}
-
-	public getTrustedActiveKeys(): Array<any> {
-		return this.pubk['active'];
-
-	}
-
-	public getEncryptKeys(): Array<any> {
-		return this.pubk['encrypt'];
-
-	}
-
-	public getPrivateKey(): any {
-		return this.pk;
-	}
-
-	public getFingerprint(): string {
-		return this.fingerprint;
-	}
-}
-
-class PGPSigner implements Signer {
-
-	engine	= 'pgp'
-	algo	= 'sha256'
-	dgst:		string
-	signature:	Signature
-	keyStore:	KeyStore
-	onsign:		(Signature) => void
-	onverify:	(boolean) => void
-
-	constructor(keyStore:KeyStore) {
-		this.keyStore = keyStore
-		this.onsign = (string) => {};
-		this.onverify = (boolean) => {};
-	}
-
-	public fingerprint(): string {
-		return this.keyStore.getFingerprint();
-	}
-
-	public prepare(material:Signable):boolean {
-		this.dgst = material.digest();
-		return true;
-	}
-
-	public verify(digest:string, signature:Signature) {
-		pgp.signature.readArmored(signature.data).then((s) => {
-			const opts = {
-				message: pgp.cleartext.fromText(digest),
-				publicKeys: this.keyStore.getTrustedKeys(),
-				signature: s,
-			};
-			pgp.verify(opts).then((v) => {
-				let i = 0;
-				for (i = 0; i < v.signatures.length; i++) {
-					const s = v.signatures[i];
-					if (s.valid) {
-						this.onverify(s);
-						return;
-					}
-				}
-				console.error('checked ' + i + ' signature(s) but none valid');
-				this.onverify(false);
-			});
-		}).catch((e) => {
-			console.error(e);
-			this.onverify(false);
-		});
-	}
-
-	public sign(digest:string) {
-		const m = pgp.cleartext.fromText(digest);
-		const pk = this.keyStore.getPrivateKey();
-		const opts = {
-			message: m,
-			privateKeys: [pk],
-			detached: true,
-		}
-		pgp.sign(opts).then((s) => {
-			this.signature = {
-				engine: this.engine,
-				algo: this.algo,
-				data: s.signature,
-				// TODO: fix for browser later
-				digest: digest,
-			};
-			this.onsign(this.signature);
-		}).catch((e) => {
-			console.error(e);
-			this.onsign(undefined);
-		});
-	}
-}
-
-export {
-	Signature,
-	Authoritative,
-       	Signer,
-	KeyGetter,
-	Signable,
-	KeyStore,
-	PGPSigner,
-	PGPKeyStore,
-};

apps/cic-meta/src/config.ts

@@ -1,71 +0,0 @@
-import * as fs from 'fs';
-import * as ini from 'ini';
-import * as path from 'path';
-
-class Config {
-
-	filepath: 	string
-	store:		Object
-	censor:		Array<string>
-	require:	Array<string>
-	env_prefix:	string
-
-	constructor(filepath:string, env_prefix?:string) {
-		this.filepath = filepath;
-		this.store = {};
-		this.censor = [];
-		this.require = [];
-		this.env_prefix = '';
-		if (env_prefix !== undefined) {
-			this.env_prefix = env_prefix + "_";
-		}
-	}
-
-	public process() {
-		const d = fs.readdirSync(this.filepath);
-		
-		const r = /.*\.ini$/;
-		for (let i = 0; i < d.length; i++) {
-			const f = d[i];
-			if (!f.match(r)) {
-				return;
-			}
-
-			const fp = path.join(this.filepath, f);
-			const v = fs.readFileSync(fp, 'utf-8');
-			const inid = ini.decode(v);
-			const inik = Object.keys(inid);
-			for (let j = 0; j < inik.length; j++) {
-				const k_section = inik[j]
-				const k = k_section.toUpperCase();
-				Object.keys(inid[k_section]).forEach((k_directive) => {
-					const kk = k_directive.toUpperCase();
-					const kkk = k + '_' + kk;
-
-					let r = inid[k_section][k_directive];
-					const k_env = this.env_prefix + kkk
-					const env = process.env[k_env];
-					if (env !== undefined) {
-						console.debug('Environment variable ' + k_env + ' overrides ' + kkk);
-						r = env;
-					}
-					this.store[kkk] = r;
-				});
-			}
-		}
-	}
-
-	public get(s:string) {
-		return this.store[s];
-	}
-
-	public toString() {
-		let s = '';
-		Object.keys(this.store).forEach((k) => {
-			s += k + '=' + this.store[k] + '\n';
-		});
-		return s;
-	}	
-}
-
-export { Config };

apps/cic-meta/src/constants.ts

@@ -1,38 +0,0 @@
-import { JSONSerializable } from './format';
-
-const ENGINE_NAME = 'automerge';
-const ENGINE_VERSION = '0.14.1';
-
-const NETWORK_NAME = 'cic';
-const NETWORK_VERSION = '1';
-
-const CRYPTO_NAME = 'pgp';
-const CRYPTO_VERSION = '2';
-
-type VersionedSpec = {
-	name:		string
-	version: 	string
-	ext?:		Object
-}
-
-const engineSpec:VersionedSpec = {
-	name: ENGINE_NAME,
-	version: ENGINE_VERSION,			
-}
-
-const cryptoSpec:VersionedSpec = {
-	name: CRYPTO_NAME,
-	version: CRYPTO_VERSION,
-}
-
-const networkSpec:VersionedSpec = {
-	name: NETWORK_NAME,
-	version: NETWORK_VERSION,
-}
-
-export {
-	engineSpec,
-	cryptoSpec,
-	networkSpec,
-	VersionedSpec,
-};

apps/cic-meta/src/crypto.ts

@@ -1,27 +0,0 @@
-import * as crypto from 'crypto';
-
-const _algs = {
-	'SHA-256': 'sha256',
-}
-
-function cryptoWrapper() {
-}
-
-cryptoWrapper.prototype.digest = async function(s, d) {
-	const h = crypto.createHash(_algs[s]);
-	h.update(d);
-	return h.digest();
-}
-
-let subtle = undefined;
-if (typeof window !== 'undefined') {
-	subtle = window.crypto.subtle;
-} else {
-	subtle = new cryptoWrapper();
-}
-
-
-export {
-	subtle,
-}
-

apps/cic-meta/src/db.ts

@@ -1,90 +0,0 @@
-import * as pg from 'pg';
-import * as sqlite from 'sqlite3';
-
-type DbConfig = {
-	name:		string
-	host:		string
-	port:		number
-	user:		string
-	password:	string	
-}
-
-interface DbAdapter {
-	query:	(s:string, callback:(e:any, rs:any) => void) => void
-	close: 	() => void
-}
-
-const re_creatematch = /^(CREATE)/i
-const re_getmatch = /^(SELECT)/i;
-const re_setmatch = /^(INSERT|UPDATE)/i;
-
-class SqliteAdapter implements DbAdapter {
-
-	db:		any
-
-	constructor(dbConfig:DbConfig, callback?:(any) => void) {
-		this.db = new sqlite.Database(dbConfig.name); //, callback);
-	}
-
-	public query(s:string, callback:(e:any, rs?:any) => void): void {
-		const local_callback = (e, rs) => {
-			let r = undefined;
-			if (rs !== undefined) {
-				r = {
-					rowCount: rs.length,
-					rows: rs,
-				}
-			}
-			callback(e, r);
-		};
-		if (s.match(re_getmatch)) {
-			this.db.all(s, local_callback);
-		} else if (s.match(re_setmatch)) {
-			this.db.run(s, local_callback);
-		} else if (s.match(re_creatematch)) {
-			this.db.run(s, callback);
-		} else {
-			throw 'unhandled query';
-		}
-	}
-
-	public close() {
-		this.db.close();
-	}
-}
-
-class PostgresAdapter implements DbAdapter {
-
-	db:		any
-
-	constructor(dbConfig:DbConfig) {
-		let o = dbConfig;
-		o['database'] = o.name;
-		this.db = new pg.Pool(o);
-		return this.db;
-	}
-
-	public query(s:string, callback:(e:any, rs:any) => void): void {
-		this.db.query(s, (e, rs) => {
-			let r = {
-				length: rs.rowCount,
-			}
-			rs.length = rs.rowCount;
-			if (e === undefined) {
-				e = null;
-			}
-			console.debug(e, rs);
-			callback(e, rs);
-		});
-	}
-
-	public close() {
-		this.db.end();
-	}
-}
-
-export {
-	DbConfig,
-	SqliteAdapter,
-	PostgresAdapter,
-}

apps/cic-meta/src/digest.ts

@@ -1,67 +0,0 @@
-import * as crypto from './crypto';
-
-interface Addressable {
-	key(): string
-	digest(): string
-}
-
-function stringToBytes(s:string) {
-	const a = new Uint8Array(20);
-	let j = 2;
-	for (let i = 0; i < a.byteLength; i++) {
-		const n = parseInt(s.substring(j, j+2), 16);
-		a[i] = n;
-		j += 2;
-	}
-	return a;
-}
-
-function bytesToHex(a:Uint8Array) {
-	let s = '';
-	for (let i = 0; i < a.byteLength; i++) {
-		const h = '00' + a[i].toString(16);
-		s += h.slice(-2);
-	}
-	return s;
-}
-
-async function mergeKey(a:Uint8Array, s:Uint8Array) {
-	const y = new Uint8Array(a.byteLength + s.byteLength);
-	for (let i = 0; i < a.byteLength; i++) {
-		y[i] = a[i];
-	}
-	for (let i = 0; i < s.byteLength; i++) {
-		y[a.byteLength + i] = s[i];
-	}
-	const z = await crypto.subtle.digest('SHA-256', y);
-	return bytesToHex(new Uint8Array(z));
-}
-
-async function toKey(v:string, salt:string) {
-	const a = stringToBytes(v);
-	const s = new TextEncoder().encode(salt);
-	return await mergeKey(a, s);
-}
-
-
-async function toAddressKey(zeroExHex:string, salt:string) {
-	const a = addressToBytes(zeroExHex);
-	const s = new TextEncoder().encode(salt);
-	return await mergeKey(a, s);
-}
-
-const re_addrHex = /^0[xX][a-fA-F0-9]{40}$/;
-function addressToBytes(s:string) {
-	if (!s.match(re_addrHex)) {
-		throw 'invalid address hex';
-	}
-	return stringToBytes(s);
-}
-
-export {
-	toKey,
-	toAddressKey,
-	bytesToHex,
-	addressToBytes,
-	Addressable,
-}

apps/cic-meta/src/dispatch.ts

@@ -1,58 +0,0 @@
-import { v4 as uuidv4 } from 'uuid';
-import { Syncable } from './sync';
-import { Store } from './store';
-import { PubSub } from './transport';
-
-function toIndexKey(id:string):string {
-	const d = Date.now();
-	return d + '_' + id + '_' + uuidv4();
-}
-
-const _re_indexKey = /^\d+_(.+)_[-\d\w]+$/;
-function fromIndexKey(s:string):string {
-	const m = s.match(_re_indexKey);
-	if (m === null) {
-		throw 'Invalid index key';
-	}
-	return m[1];
-}
-
-class Dispatcher {
-
-	idx:		Array<string>
-	syncer:		PubSub
-	store:		Store
-
-	constructor(store:Store, syncer:PubSub) {
-		this.idx = new Array<string>()
-		this.syncer = syncer;
-		this.store = store;
-	}
-
-	public isDirty(): boolean {
-		return this.idx.length > 0;
-	}
-
-	public add(id:string, item:Syncable): string {
-		const v = item.toJSON();
-		const k = toIndexKey(id);
-		this.store.put(k, v, true);
-		localStorage.setItem(k, v);
-		this.idx.push(k);
-		return k;
-	}
-
-	public sync(offset:number): number {
-		let i = 0;
-		this.idx.forEach((k) => {
-			const v = localStorage.getItem(k);
-			const k_id = fromIndexKey(k);
-			this.syncer.pub(v); // this must block until guaranteed delivery
-			localStorage.removeItem(k);
-			i++;
-		});
-		return i;
-	}
-}
-
-export { Dispatcher, toIndexKey, fromIndexKey }

apps/cic-meta/src/format.ts

@@ -1,5 +0,0 @@
-interface JSONSerializable {
-	toJSON(): string
-}
-
-export { JSONSerializable };

apps/cic-meta/src/index.ts

@@ -1,5 +1,2 @@
-export { PGPSigner, PGPKeyStore, Signer, KeyStore } from './auth';
+export { User } from './user';
-export { ArgPair,  Envelope, Syncable } from './sync';
+export { Phone } from './phone';
-export { User } from './assets/user';
-export { Phone } from './assets/phone';
-export { Config } from './config';

apps/cic-meta/src/phone.ts

@@ -1,12 +1,11 @@
-import { ArgPair, Syncable } from '../sync';
+import { Syncable, Addressable, mergeKey } from 'crdt-meta';
-import { Addressable, addressToBytes, bytesToHex, toKey } 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 +16,8 @@ class Phone extends Syncable implements Addressable {
 		});
 	}
 
-	public static async toKey(msisdn:number) {
+	public static async toKey(msisdn:string) {
-		return await toKey(msisdn.toString(), ':cic.msisdn');
+		return await mergeKey(Buffer.from(msisdn), Buffer.from(':cic.phone'));
 	}
 
 	public key(): string {

apps/cic-meta/src/store.ts

@@ -1,9 +0,0 @@
-import { Syncable } from './sync';
-
-interface Store {
-	put(string, Syncable, boolean?)
-	get(string):Syncable
-	delete(string)
-}
-
-export { Store };

apps/cic-meta/src/sync.ts

@@ -1,266 +0,0 @@
-import * as Automerge from 'automerge';
-
-import { JSONSerializable } from './format';
-
-import { Authoritative, Signer, PGPSigner, Signable, Signature } from './auth';
-
-import { engineSpec, cryptoSpec, networkSpec, VersionedSpec  } from './constants';
-
-const fullSpec:VersionedSpec = {
-	name: 'cic',
-	version: '1',
-	ext: {
-		network: cryptoSpec,
-		engine: engineSpec,
-	},
-}
-
-class Envelope {
-
-	o = fullSpec
-
-	constructor(payload:Object) {
-		this.set(payload);
-	}
-
-	public set(payload:Object) {
-		this.o['payload'] = payload
-	}
-
-	public get():string {
-		return this.o['payload'];
-	}
-
-	public toJSON() {
-		return JSON.stringify(this.o);
-	}
-
-	public static fromJSON(s:string): Envelope {
-		const e = new Envelope(undefined);
-		e.o = JSON.parse(s);
-		return e;
-	}
-
-	public unwrap(): Syncable {
-		return Syncable.fromJSON(this.o['payload']);
-	}
-}
-
-class ArgPair {
-
-	k:string
-	v:any
-
-	constructor(k:string, v:any) {
-		this.k = k;
-		this.v = v;
-	}
-}
-
-class SignablePart implements Signable {
-
-	s: string
-
-	constructor(s:string) {
-		this.s = s;
-	}
-
-	public digest():string {
-		return this.s;
-	}
-}
-
-function orderDict(src) {
-	let dst;
-	if (Array.isArray(src)) {
-		dst = [];
-		src.forEach((v) => {
-			if (typeof(v) == 'object') {
-				v = orderDict(v);
-			}
-			dst.push(v);
-		});
-	} else {
-		dst = {}
-		Object.keys(src).sort().forEach((k) => {
-			let v = src[k];
-			if (typeof(v) == 'object') {
-				v = orderDict(v);
-			}
-			dst[k] = v;
-		});
-	} 
-	return dst;
-}
-
-class Syncable implements JSONSerializable, Authoritative, Signable {
-
-	id:		string
-	timestamp:	number
-	m:		any // automerge object
-	e:		Envelope
-	signer:		Signer
-	onwrap: 	(string) => void
-	onauthenticate:	(boolean) => void
-
-	// TODO: Move data to sub-object so timestamp, id, signature don't collide
-	constructor(id:string, v:Object) {
-		this.id = id;
-		const o = {
-			'id': id,
-			'timestamp': Math.floor(Date.now() / 1000),
-			'data': v,
-		}
-		//this.m = Automerge.from(v)
-		this.m = Automerge.from(o)
-	}
-
-	public setSigner(signer:Signer) {
-		this.signer = signer;
-		this.signer.onsign = (s) => {
-			this.wrap(s);
-		};
-	}
-
-	// TODO: To keep integrity, the non-link key/value pairs for each step also need to be hashed
-	public digest(): string {
-		const links = [];
-		Automerge.getAllChanges(this.m).forEach((ch:Object) => {
-			const op:Array<any> = ch['ops'];
-			ch['ops'].forEach((op:Array<Object>) => {
-				if (op['action'] == 'link') {
-					//console.log('op link', op);
-					links.push([op['obj'], op['value']]);
-				}
-			});
-		});
-		//return JSON.stringify(links);
-		const j = JSON.stringify(links);
-		return Buffer.from(j).toString('base64');
-	}
-
-	private wrap(s:any) {
-		this.m = Automerge.change(this.m, 'sign', (doc) => {
-			doc['signature'] = s;
-		});
-		this.e = new Envelope(this.toJSON());
-		console.log('wrappin s', s, typeof(s));
-		this.e.o['digest'] = s.digest;
-		if (this.onwrap !== undefined) {
-			this.onwrap(this.e);
-		}
-
-	}
-
-//	private _verifyLoop(i:number, history:Array<any>, signable:Signable, result:boolean) {
-//		if (!result) {
-//			this.onauthenticate(false);
-//			return;
-//		} else if (history.length == 0) {
-//			this.onauthenticate(true);
-//			return;
-//		}
-//		const h = history.shift()
-//		if (i % 2 == 0) {
-//			i++;
-//			signable = {
-//				digest: () => {
-//					return Automerge.save(h.snapshot)
-//				},
-//			};
-//			this._verifyLoop(i, history, signable, true);
-//		} else {
-//			i++;
-//			const signature = h.snapshot['signature'];
-//			console.debug('signature', signature, signable.digest());
-//			this.signer.onverify = (v) => {
-//				this._verifyLoop(i, history, signable, v)
-//			}
-//			this.signer.verify(signable, signature);
-//		}
-//	}
-//
-//	// TODO: This should replay the graph and check signatures on each step
-//	public _authenticate(full:boolean=false) {
-//		let h = Automerge.getHistory(this.m);
-//		h.forEach((m) => {
-//			//console.debug(m.snapshot);
-//		});
-//		const signable = {
-//			digest: () => { return '' },
-//		}
-//		if (!full) {
-//			h = h.slice(h.length-2);
-//		}
-//		this._verifyLoop(0, h, signable, true);
-//	}
-
-	public authenticate(full:boolean=false) {
-		if (full) {
-			console.warn('only doing shallow authentication for now, sorry');			
-		}
-		//console.log('authenticating', signable.digest());
-		//console.log('signature', this.m.signature);
-		this.signer.onverify = (v) => {
-			//this._verifyLoop(i, history, signable, v)
-			this.onauthenticate(v);
-		}
-		this.signer.verify(this.m.signature.digest, this.m.signature);
-	}
-
-
-	public sign() {
-		//this.signer.prepare(this);
-		this.signer.sign(this.digest());
-	}
-
-	public update(changes:Array<ArgPair>, changesDescription:string) {
-		this.m = Automerge.change(this.m, changesDescription, (m) => {
-			changes.forEach((c) => {
-				let path = c.k.split('.');
-				let target = m['data'];
-				while (path.length > 1) {
-					const part = path.shift();
-					target = target[part];
-				}
-				target[path[0]] = c.v;
-			});
-			m['timestamp'] = Math.floor(Date.now() / 1000);
-		});
-	}
-
-	public replace(o:Object, changesDescription:string) {
-		this.m = Automerge.change(this.m, changesDescription, (m) => {
-			Object.keys(o).forEach((k) => {
-				m['data'][k] = o[k];
-			});
-			Object.keys(m).forEach((k) => {
-				if (o[k] == undefined) {
-					delete m['data'][k];
-				}
-			});
-			m['timestamp'] = Math.floor(Date.now() / 1000);
-		});
-	}
-
-	public merge(s:Syncable) {
-		this.m = Automerge.merge(s.m, this.m);
-	}
-
-	public toJSON(): string {
-		const s = Automerge.save(this.m);
-		const o = JSON.parse(s);
-		const oo = orderDict(o)
-		return JSON.stringify(oo);
-
-	}
-
-	public static fromJSON(s:string): Syncable {
-		const doc = Automerge.load(s);
-		let y = new Syncable(doc['id'], {});
-		y.m = doc
-		return y
-	}
-}
-
-export { JSONSerializable, Syncable, ArgPair, Envelope };

apps/cic-meta/src/transport.ts

@@ -1,11 +0,0 @@
-interface SubConsumer {
-	post(string)
-}
-
-interface PubSub {
-	pub(v:string):boolean
-	close()
-}
-
-export { PubSub, SubConsumer };
-

apps/cic-meta/src/user.ts

@@ -1,5 +1,4 @@
-import { ArgPair, Syncable } from '../sync';
+import { Syncable, Addressable, toAddressKey } from 'crdt-meta';
-import { Addressable, addressToBytes, bytesToHex, toAddressKey } from '../digest';
 
 const keySalt = new TextEncoder().encode(':cic.person');
 class User extends Syncable implements Addressable {

apps/cic-meta/tests/1_basic.ts

@@ -1,50 +0,0 @@
-import * as Automerge from 'automerge';
-import assert = require('assert');
-
-import { Dispatcher, toIndexKey, fromIndexKey } from '../src/dispatch';
-import { User } from '../src/assets/user';
-import { Syncable, ArgPair } from '../src/sync';
-
-import { MockSigner, MockStore } from './mock';
-
-describe('basic', () => {
-
-	it('store', () => {
-		const store = new MockStore('s');
-		assert.equal(store.name, 's');
-
-		const mockSigner = new MockSigner();
-		const v = new Syncable('foo', {baz: 42});
-		v.setSigner(mockSigner);
-		store.put('foo', v);
-		const one = store.get('foo').toJSON();
-		const vv = new Syncable('bar', {baz: 666});
-		vv.setSigner(mockSigner);
-		assert.throws(() => {
-			store.put('foo', vv)
-		});
-		store.put('foo', vv, true);
-		const other = store.get('foo').toJSON();
-		assert.notEqual(one, other);
-		store.delete('foo');
-		assert.equal(store.get('foo'), undefined);
-	});
-
-	it('add_doc_to_dispatcher', () => {
-		const store = new MockStore('s');
-		//const syncer = new MockSyncer();
-		const dispatcher = new Dispatcher(store, undefined);
-		const user = new User('foo'); 
-		dispatcher.add(user.id, user);
-		assert(dispatcher.isDirty());
-	});
-
-	it('dispatch_keyindex', () => {
-		const s = 'foo';
-		const k = toIndexKey(s);
-		const v = fromIndexKey(k);
-		assert.equal(s, v);
-	});
-
-
-});

apps/cic-meta/tests/2_sync.ts

@@ -1,212 +0,0 @@
-import * as Automerge from 'automerge';
-import assert = require('assert');
-
-import * as pgp from 'openpgp';
-import * as fs from 'fs';
-
-import { PGPSigner } from '../src/auth';
-
-import { Syncable, ArgPair } from '../src/sync';
-
-import { MockKeyStore, MockSigner } from './mock';
-
-
-describe('sync', async () => {
-	it('sync_merge', () => {
-		const mockSigner = new MockSigner();
-		const s = new Syncable('foo', {
-			bar: 'baz',
-		});
-		s.setSigner(mockSigner);
-		const changePair = new ArgPair('xyzzy', 42);
-		s.update([changePair], 'ch-ch-cha-changes');
-		assert.equal(s.m.data['xyzzy'], 42)
-		assert.equal(s.m.data['bar'], 'baz')
-		assert.equal(s.m['id'], 'foo')
-		assert.equal(Automerge.getHistory(s.m).length, 2);
-	});
-
-	it('sync_serialize', () => {
-		const mockSigner = new MockSigner();
-		const s = new Syncable('foo', {
-			bar: 'baz',
-		});
-		s.setSigner(mockSigner);
-		const j = s.toJSON();
-		const ss = Syncable.fromJSON(j);
-		assert.equal(ss.m['id'], 'foo');
-		assert.equal(ss.m['data']['bar'], 'baz');
-		assert.equal(Automerge.getHistory(ss.m).length, 1);
-	});
-
-	it('sync_sign_and_wrap', () => {
-		const mockSigner = new MockSigner();
-		const s = new Syncable('foo', {
-			bar: 'baz',
-		});
-		s.setSigner(mockSigner);
-		s.onwrap = (e) => {
-			const j = e.toJSON();
-			const v = JSON.parse(j);
-			assert.deepEqual(v.payload, e.o.payload);
-
-		}
-		s.sign();
-	});
-	it('sync_verify_success', async () => {
-		const pksa = fs.readFileSync(__dirname + '/privatekeys.asc');
-		const pks = await pgp.key.readArmored(pksa);
-		await pks.keys[0].decrypt('merman');
-		await pks.keys[1].decrypt('beastman');
-
-		const pubksa = fs.readFileSync(__dirname + '/publickeys.asc');
-		const pubks = await pgp.key.readArmored(pubksa);
-
-		const oneStore = new MockKeyStore(pks.keys[0], pubks.keys);
-		const twoStore = new MockKeyStore(pks.keys[1], pubks.keys);
-		const threeStore = new MockKeyStore(pks.keys[2], [pubks.keys[0], pubks.keys[2]]);
-
-		const oneSigner = new PGPSigner(oneStore);
-		const twoSigner = new PGPSigner(twoStore);
-		const threeSigner = new PGPSigner(threeStore);
-
-		const x = new Syncable('foo', {
-			bar: 'baz',		  
-		});
-		x.setSigner(oneSigner);
-
-		// TODO: make this look better
-		x.onwrap = (e) => {
-			let updateData = new ArgPair('bar', 'xyzzy');
-			x.update([updateData], 'change one');
-
-			x.onwrap = (e) => {
-				x.setSigner(twoSigner);
-				updateData = new ArgPair('bar', 42);
-				x.update([updateData], 'change two');
-
-				x.onwrap = (e) => {
-					const p = e.unwrap();
-					p.setSigner(twoSigner);
-					p.onauthenticate = (v) => {
-						assert(v);
-					}
-					p.authenticate();
-				}
-
-				x.sign();
-			};
-
-			x.sign();
-		}
-
-		x.sign();
-
-	});
-
-	it('sync_verify_fail', async () => {
-		const pksa = fs.readFileSync(__dirname + '/privatekeys.asc');
-		const pks = await pgp.key.readArmored(pksa);
-		await pks.keys[0].decrypt('merman');
-		await pks.keys[1].decrypt('beastman');
-
-		const pubksa = fs.readFileSync(__dirname + '/publickeys.asc');
-		const pubks = await pgp.key.readArmored(pubksa);
-
-		const oneStore = new MockKeyStore(pks.keys[0], pubks.keys);
-		const twoStore = new MockKeyStore(pks.keys[1], pubks.keys);
-		const threeStore = new MockKeyStore(pks.keys[2], [pubks.keys[0], pubks.keys[2]]);
-
-		const oneSigner = new PGPSigner(oneStore);
-		const twoSigner = new PGPSigner(twoStore);
-		const threeSigner = new PGPSigner(threeStore);
-
-		const x = new Syncable('foo', {
-			bar: 'baz',		  
-		});
-		x.setSigner(oneSigner);
-
-		// TODO: make this look better
-		x.onwrap = (e) => {
-			let updateData = new ArgPair('bar', 'xyzzy');
-			x.update([updateData], 'change one');
-
-			x.onwrap = (e) => {
-				x.setSigner(twoSigner);
-				updateData = new ArgPair('bar', 42);
-				x.update([updateData], 'change two');
-
-				x.onwrap = (e) => {
-					const p = e.unwrap();
-					p.setSigner(threeSigner);
-					p.onauthenticate = (v) => {
-						assert(!v);
-					}
-					p.authenticate();
-				}
-
-				x.sign();
-			};
-
-			x.sign();
-		}
-
-		x.sign();
-
-	});
-
-	xit('sync_verify_shallow_tricked', async () => {
-		const pksa = fs.readFileSync(__dirname + '/privatekeys.asc');
-		const pks = await pgp.key.readArmored(pksa);
-		await pks.keys[0].decrypt('merman');
-		await pks.keys[1].decrypt('beastman');
-
-		const pubksa = fs.readFileSync(__dirname + '/publickeys.asc');
-		const pubks = await pgp.key.readArmored(pubksa);
-
-		const oneStore = new MockKeyStore(pks.keys[0], pubks.keys);
-		const twoStore = new MockKeyStore(pks.keys[1], pubks.keys);
-		const threeStore = new MockKeyStore(pks.keys[2], [pubks.keys[0], pubks.keys[2]]);
-
-		const oneSigner = new PGPSigner(oneStore);
-		const twoSigner = new PGPSigner(twoStore);
-		const threeSigner = new PGPSigner(threeStore);
-
-		const x = new Syncable('foo', {
-			bar: 'baz',		  
-		});
-		x.setSigner(twoSigner);
-
-		// TODO: make this look better
-		x.onwrap = (e) => {
-			let updateData = new ArgPair('bar', 'xyzzy');
-			x.update([updateData], 'change one');
-
-			x.onwrap = (e) => {
-				updateData = new ArgPair('bar', 42);
-				x.update([updateData], 'change two');
-				x.setSigner(oneSigner);
-
-				x.onwrap = (e) => {
-					const p = e.unwrap();
-					p.setSigner(threeSigner);
-					p.onauthenticate = (v) => {
-						assert(v);
-						p.onauthenticate = (v) => {
-							assert(!v);
-						}
-						p.authenticate(true);
-					}
-					p.authenticate();
-				}
-
-				x.sign();
-			};
-
-			x.sign();
-		}
-
-		x.sign();
-
-	});
-});

apps/cic-meta/tests/3_transport.ts

@@ -1,14 +0,0 @@
-import * as assert from 'assert';
-
-import { MockPubSub, MockConsumer } from './mock';
-
-describe('transport', () => {
-	it('pub_sub', () => {
-		const c = new MockConsumer();
-		const ps = new MockPubSub('foo', c);
-		ps.pub('foo');	
-		ps.pub('bar');
-		ps.flush();
-		assert.deepEqual(c.omnoms, ['foo', 'bar']);
-	});
-});

apps/cic-meta/tests/4_auth.ts

@@ -1,46 +0,0 @@
-import assert = require('assert');
-import pgp = require('openpgp');
-import crypto  = require('crypto');
-
-import { Syncable, ArgPair } from '../src/sync';
-
-import { MockKeyStore, MockSignable } from './mock';
-
-import { PGPSigner } from '../src/auth';
-
-
-describe('auth', async () => {
-	await it('digest', async () => {
-		const opts = {
-			userIds: [
-				{
-					name: 'John Marston',
-					email: 'red@dead.com',
-				},
-			],
-			numBits: 2048,
-			passphrase: 'foo',
-		};
-		const pkgen = await pgp.generateKey(opts);
-		const pka = pkgen.privateKeyArmored;
-		const pks = await pgp.key.readArmored(pka);
-		await pks.keys[0].decrypt('foo');
-		const pubka = pkgen.publicKeyArmored;
-		const pubks = await pgp.key.readArmored(pubka);
-		const keyStore = new MockKeyStore(pks.keys[0], pubks.keys);
-		const s = new PGPSigner(keyStore);
-
-		const message = await pgp.cleartext.fromText('foo');
-		s.onverify = (ok) => {
-			assert(ok);
-		}
-		s.onsign = (signature) => {
-			s.onverify((v) => {
-				console.log('bar', v);
-			});
-			s.verify('foo', signature);
-		}
-		
-		await s.sign('foo');
-	});
-});	

apps/cic-meta/tests/999_functional.ts

@@ -1,47 +0,0 @@
-import * as assert from 'assert';
-import * as pgp from 'openpgp';
-
-import { Dispatcher } from '../src/dispatch';
-import { User } from '../src/assets/user';
-import { PGPSigner, KeyStore } from '../src/auth';
-import { SubConsumer } from '../src/transport';
-
-import { MockStore, MockPubSub, MockConsumer, MockKeyStore } from './mock';
-
-async function createKeyStore() {
-	const opts = {
-		userIds: [
-			{
-				name: 'John Marston',
-				email: 'red@dead.com',
-			},
-		],
-		numBits: 2048,
-		passphrase: 'foo',
-	};
-	const pkgen = await pgp.generateKey(opts);
-	const pka = pkgen.privateKeyArmored;
-	const pks = await pgp.key.readArmored(pka);
-	await pks.keys[0].decrypt('foo');
-	return new MockKeyStore(pks.keys[0], []);
-}
-
-describe('fullchain', async () => {
-	it('dispatch_and_publish_user', async () => {
-		const g = await createKeyStore();
-		const n = new PGPSigner(g);
-		const u = new User('u1', {});
-		u.setSigner(n);
-		u.setName('Nico', 'Bellic');
-		const s = new MockStore('fooStore');
-		const c = new MockConsumer();
-		const p = new MockPubSub('fooPubSub', c);
-		const d = new Dispatcher(s, p);
-		u.onwrap = (e) => {
-			d.add(u.id, e);
-			d.sync(0);
-			assert.equal(p.pubs.length, 1);
-		};
-		u.sign();
-	});
-});

apps/cic-meta/tests/mock.ts

@@ -1,150 +0,0 @@
-import * as crypto from 'crypto';
-
-import { Signable, Signature, KeyStore } from '../src/auth';
-import { Store } from '../src/store';
-import { PubSub, SubConsumer } from '../src/transport';
-import { Syncable } from '../src/sync';
-
-class MockStore implements Store {
-	
-	contents:	Object
-	name:		string
-
-	constructor(name:string) {
-		this.name = name;
-		this.contents = {};
-	}
-
-	public put(k:string, v:Syncable, existsOk = false) {
-		if (!existsOk && this.contents[k] !== undefined) {
-			throw '"' + k + '" already exists in store ' + this.name;
-		} 
-		this.contents[k] = v;
-	}
-
-	public get(k:string): Syncable {
-		return this.contents[k];
-	}
-
-	public delete(k:string) {
-		delete this.contents[k];
-	}
-}
-
-class MockSigner {
-	onsign: (string) => void
-	onverify: (boolean) => void
-	public verify(src:string, signature:Signature) {
-		return true;
-	}
-
-	public sign(s:string):boolean {
-		this.onsign('there would be a signature here');
-		return true;
-	}
-
-	public prepare(m:Signable):boolean {
-		return true;
-	}
-
-	public fingerprint():string {
-		return '';
-	}
-}
-
-class MockConsumer implements SubConsumer {
-
-	omnoms:	Array<string>
-
-	constructor() {
-		this.omnoms = Array<string>();
-	}
-
-	public post(v:string) {
-		this.omnoms.push(v);
-	}
-}
-
-class MockPubSub implements PubSub {
-
-	pubs:		Array<string>
-	consumer:	SubConsumer
-
-	constructor(name:string, consumer:SubConsumer) {
-		this.pubs = Array<string>();	
-		this.consumer = consumer;
-	}
-	
-	public pub(v:string): boolean {
-		this.pubs.push(v);
-		return true;
-	}
-
-	public flush() {
-		while (this.pubs.length > 0) {
-			const s = this.pubs.shift();
-			this.consumer.post(s);
-		}
-	}
-
-	public close() {
-	}
-}
-
-class MockSignable implements Signable {
-
-	src:	string
-	dst:	string
-
-	constructor(src:string) {
-		this.src = src;
-	}
-
-	public digest():string {
-		const h = crypto.createHash('sha256');
-		h.update(this.src);
-		this.dst= h.digest('hex');
-		return this.dst;
-	}
-
-}
-
-class MockKeyStore implements KeyStore {
-
-	pk: any
-	pubks: Array<any>
-
-	constructor(pk:any, pubks:Array<any>) {
-		this.pk = pk;
-		this.pubks = pubks;	
-	}
-
-	public getPrivateKey(): any {
-		return this.pk;
-	}
-
-	public getTrustedKeys(): Array<any> {
-		return this.pubks;
-	}
-
-	public getTrustedActiveKeys(): Array<any> {
-		return [];
-	}
-
-	public getEncryptKeys(): Array<any> {
-		return [];
-	}
-
-	public getFingerprint(): string {
-		return '';
-	}
-}
-
-export {
-	MockStore,
-	MockPubSub,
-	MockConsumer,
-	MockSignable,
-	MockKeyStore,
-	MockSigner,
-};

apps/cic-meta/tests/server.ts

@@ -1,13 +1,10 @@
-import Automerge = require('automerge');
 import assert = require('assert');
 import fs = require('fs');
 import pgp = require('openpgp');
 import sqlite = require('sqlite3');
 
 import * as handlers from '../scripts/server/handlers';
-import { Envelope, Syncable, ArgPair } from '../src/sync';
+import { Envelope, Syncable, ArgPair, PGPKeyStore, PGPSigner, KeyStore, Signer, SqliteAdapter } from 'crdt-meta';
-import { PGPKeyStore, PGPSigner, KeyStore, Signer } from '../src/auth';
-import { SqliteAdapter } from '../src/db';
 
 function createKeystore() {
 	const pksa = fs.readFileSync(__dirname + '/privatekeys.asc', 'utf-8');

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

apps/cic-notify/setup.cfg

@@ -1,6 +1,5 @@
 [metadata]
 name = cic-notify
-version= 0.4.0a3
 description = CIC notifications service
 author = Louis Holbrook
 author_email = dev@holbrook.no

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

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_account_phone_number'), 'account', ['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_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_account_blockchain_address'), table_name='account')
-    op.drop_index(op.f('ix_user_phone_number'), table_name='user')
+    op.drop_index(op.f('ix_account_phone_number'), table_name='account')
-    op.drop_table('user')
+    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
+    """Database connection engine of the running aplication"""
-    query = None
+    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)
+        """Creates a new database session.
-        return 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():
+    def connect(dsn, pool_size=16, debug=False):
-        Model.metadata.create_all(bind=SessionBase.engine)
+        """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/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)