kamikazechaser/cic-internal-integration
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

Add separate generic task for proof verification

Browse Source
This commit is contained in:
nolash 2021-10-08 20:19:15 +02:00
parent df21db958b
commit 7e2857d1e7
Signed by untrusted user who does not match committer: lash
GPG Key ID: 21D2E7BB88C2A746
4 changed files with 126 additions and 39 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

19
apps/cic-eth/cic_eth/api/api_task.py
View File

@ -35,11 +35,13 @@ class Api(ApiBase):
return s_token.apply_async()
def token(self, token_symbol):
return self.tokens([token_symbol])
def token(self, token_symbol, proof=None):
return self.tokens([token_symbol], proof=proof)
def tokens(self, token_symbols):
def tokens(self, token_symbols, proof=None):
if isinstance(proof, str):
proof = [[proof]]
chain_spec_dict = self.chain_spec.asdict()
s_token_resolve = celery.signature(
'cic_eth.eth.erc20.resolve_tokens_by_symbol',
@ -54,13 +56,22 @@ class Api(ApiBase):
'cic_eth.eth.erc20.token_info',
[
chain_spec_dict,
proof,
],
queue=self.queue,
)
s_token_verify = celery.signature(
'cic_eth.eth.erc20.verify_token_info',
[
chain_spec_dict,
],
queue=self.queue,
)
s_token_resolve.link(s_token)
s_token.link(s_token_verify)
if self.callback_param != None:
s_token.link(self.callback_success)
s_token_verify.link(self.callback_success)
return s_token_resolve.apply_async()

71
apps/cic-eth/cic_eth/eth/erc20.py
View File

@ -20,7 +20,6 @@ from chainqueue.error import NotLocalTxError
from eth_erc20 import ERC20
from chainqueue.sql.tx import cache_tx_dict
from okota.token_index import to_identifier
from eth_address_declarator import Declarator
# local imports
from cic_eth.db.models.base import SessionBase
@ -29,7 +28,6 @@ from cic_eth.error import (
TokenCountError,
PermanentTxError,
OutOfGasError,
TrustError,
YouAreBrokeError,
)
from cic_eth.queue.tx import register_tx
@ -46,6 +44,7 @@ from cic_eth.task import (
)
from cic_eth.eth.nonce import CustodialTaskNonceOracle
from cic_eth.encode import tx_normalize
from cic_eth.eth.trust import verify_proofs
celery_app = celery.current_app
logg = logging.getLogger()
@ -479,54 +478,60 @@ def cache_approve_data(
@celery_app.task(bind=True, base=BaseTask)
def token_info(self, tokens, chain_spec_dict):
def token_info(self, tokens, chain_spec_dict, proofs=[]):
chain_spec = ChainSpec.from_dict(chain_spec_dict)
rpc = RPCConnection.connect(chain_spec, 'default')
declarator = Declarator(chain_spec)
session = self.create_session()
sender_address = AccountRole.get_address('DEFAULT', session)
sender_address = AccountRole.get_address('DEFAULT', session)
registry = CICRegistry(chain_spec, rpc)
declarator_address = registry.by_name('AddressDeclarator', sender_address=sender_address)
have_proof = False
result_data = []
i = 0
for token in tokens:
token_chain_object = ERC20Token(chain_spec, rpc, add_0x(token['address']))
token_chain_object.load(rpc)
token_symbol_proof_hex = to_identifier(token_chain_object.symbol)
token_proofs = [token_symbol_proof_hex]
if len(proofs) > 0:
token_proofs += proofs[i]
token_data = {
'decimals': token_chain_object.decimals,
'name': token_chain_object.name,
'symbol': token_chain_object.symbol,
'address': token_chain_object.address,
'declaration': {},
}
token_proof_hex = to_identifier(token_chain_object.symbol)
logg.debug('token proof to match is {}'.format(token_proof_hex))
for trusted_address in self.trusted_addresses:
o = declarator.declaration(declarator_address, trusted_address, token_chain_object.address, sender_address=sender_address)
r = rpc.do(o)
declarations = declarator.parse_declaration(r)
token_data['declaration'][trusted_address] = declarations
logg.debug('declarations for {} by {}: {}'.format(token_chain_object.address, trusted_address, declarations))
for declaration in declarations:
if declaration == token_proof_hex:
logg.debug('have token proof {} match for trusted address {}'.format(declaration, trusted_address))
have_proof = True
if not have_proof:
raise TrustError('no proof found for token {}'.format(token_chain_object.symbol))
'proofs': token_proofs,
}
result_data.append(token_data)
i += 1
return result_data
@celery_app.task(bind=True, base=BaseTask)
def verify_token_info(self, tokens, chain_spec_dict):
subjects = []
proofs = []
for token in tokens:
subjects.append(token['address'])
proofs.append(token['proofs'])
queue = self.request.delivery_info.get('routing_key')
s = celery.signature(
'cic_eth.eth.trust.verify_proofs',
[
tokens,
chain_spec_dict,
subjects,
proofs,
],
queue=queue,
)
s.apply_async()
return tokens
@celery_app.task(bind=True, base=BaseTask)
def default_token(self):
return {

69
apps/cic-eth/cic_eth/eth/trust.py Normal file
View File

@ -0,0 +1,69 @@
# standard imports
import logging
# external imports
import celery
from eth_address_declarator import Declarator
from chainlib.connection import RPCConnection
from chainlib.chain import ChainSpec
from cic_eth.db.models.role import AccountRole
from cic_eth_registry import CICRegistry
# local imports
from cic_eth.task import BaseTask
from cic_eth.error import TrustError
celery_app = celery.current_app
logg = logging.getLogger()
@celery_app.task(bind=True, base=BaseTask)
def verify_proofs(self, chained_input, chain_spec_dict, subjects, proofs):
if not isinstance(subjects, list):
raise ValueError('subjects argument must be list')
if isinstance(proofs, str):
proofs = [[proofs]]
elif not isinstance(proofs, list):
raise ValueError('proofs argument must be string or list')
if len(proofs) != 1 and len(subjects) != len(proofs):
raise ValueError('proof argument must be single proof or one proof input per subject')
chain_spec = ChainSpec.from_dict(chain_spec_dict)
rpc = RPCConnection.connect(chain_spec, 'default')
declarator = Declarator(chain_spec)
session = self.create_session()
sender_address = AccountRole.get_address('DEFAULT', session)
registry = CICRegistry(chain_spec, rpc)
declarator_address = registry.by_name('AddressDeclarator', sender_address=sender_address)
i = 0
for proof in proofs:
if not isinstance(proof, list):
logg.debug('proof entry {} is not a list'.format(i))
i += 1
i = 0
for subject in subjects:
for trusted_address in self.trusted_addresses:
proof_count = {}
for proof in proofs[i]:
o = declarator.declaration(declarator_address, trusted_address, subject, sender_address=sender_address)
r = rpc.do(o)
declarations = declarator.parse_declaration(r)
logg.debug('comparing proof {} with declarations for {} by {}: {}'.format(proofs, subject, trusted_address, declarations))
for declaration in declarations:
if declaration == proof:
logg.debug('have token proof {} match for trusted address {}'.format(declaration, trusted_address))
if proof_count.get(proof) == None:
proof_count[proof] = 0
proof_count[proof] += 1
for k in proof_count.keys():
if proof_count[k] == 0:
raise TrustError('no proof found for token {}'.format(subject))
i += 1
return chained_input

6
apps/cic-eth/tests/task/api/test_app_noncritical.py
View File

@ -36,17 +36,19 @@ def test_tokens(
init_database,
init_celery_tasks,
custodial_roles,
foo_token_declaration,
bar_token_declaration,
celery_worker,
):
api = Api(str(default_chain_spec), queue=None)
t = api.token('FOO')
t = api.token('FOO', proof=foo_token_declaration)
r = t.get_leaf()
assert len(r) == 1
assert r[0]['address'] == foo_token
t = api.tokens(['BAR', 'FOO'])
t = api.tokens(['BAR', 'FOO'], proof=[[foo_token_declaration], [bar_token_declaration]])
r = t.get_leaf()
assert len(r) == 2
assert r[1]['address'] == foo_token