funga/crypto_dev_signer/eth/web3ext/middleware.py

117 lines
3.9 KiB
Python
Raw Normal View History

2020-08-08 10:45:37 +02:00
# standard imports
import logging
import re
import socket
import uuid
import json
2020-08-07 11:45:16 +02:00
logg = logging.getLogger(__file__)
def jsonrpc_request(method, params):
uu = uuid.uuid4()
return {
"jsonrpc": "2.0",
"id": str(uu),
"method": method,
"params": params,
}
class PlatformMiddleware:
# id for the request is not available, meaning we cannot easily short-circuit
# hack workaround
id_seq = -1
re_personal = re.compile('^personal_.*')
ipcaddr = None
def __init__(self, make_request, w3):
self.w3 = w3
self.make_request = make_request
2020-10-17 15:08:33 +02:00
if self.ipcaddr == None:
raise AttributeError('ipcaddr not set')
# TODO: understand what format input params come in
# single entry input gives a tuple on params, wtf...
# dict input comes as [{}] and fails if not passed on as an array
@staticmethod
def _translate_params(params):
2020-10-26 09:02:29 +01:00
#if params.__class__.__name__ == 'tuple':
# r = []
# for p in params:
# r.append(p)
# return r
if params.__class__.__name__ == 'list' and len(params) > 0:
return params[0]
return params
2020-12-19 08:47:21 +01:00
# TODO: DRY
def __call__(self, method, suspect_params):
self.id_seq += 1
logg.debug('in middleware method {} params {} ipcpath {}'.format(method, suspect_params, self.ipcaddr))
if self.re_personal.match(method) != None:
params = PlatformMiddleware._translate_params(suspect_params)
2020-08-08 14:14:48 +02:00
# multiple providers is removed in web3.py 5.12.0
# https://github.com/ethereum/web3.py/issues/1701
2020-08-08 14:14:48 +02:00
# thus we need a workaround to use the same web3 instance
s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM, proto=0)
ipc_provider_workaround = s.connect(self.ipcaddr)
logg.info('redirecting method {} params {} original params {}'.format(method, params, suspect_params))
2020-10-26 09:02:29 +01:00
o = jsonrpc_request(method, params[0])
j = json.dumps(o)
logg.debug('send {}'.format(j))
s.send(j.encode('utf-8'))
r = s.recv(4096)
s.close()
logg.debug('got recv {}'.format(str(r)))
jr = json.loads(r)
jr['id'] = self.id_seq
#return str(json.dumps(jr))
return jr
elif method == 'eth_signTransaction':
params = PlatformMiddleware._translate_params(suspect_params)
s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM, proto=0)
ipc_provider_workaround = s.connect(self.ipcaddr)
2020-12-19 08:47:21 +01:00
logg.info('redirecting method {} params {} original params {}'.format(method, params, suspect_params))
2020-10-26 09:02:29 +01:00
o = jsonrpc_request(method, params[0])
j = json.dumps(o)
logg.debug('send {}'.format(j))
s.send(j.encode('utf-8'))
r = s.recv(4096)
s.close()
logg.debug('got recv {}'.format(str(r)))
jr = json.loads(r)
jr['id'] = self.id_seq
#return str(json.dumps(jr))
return jr
2020-12-19 08:47:21 +01:00
elif method == 'eth_sign':
params = PlatformMiddleware._translate_params(suspect_params)
s = socket.socket(family=socket.AF_UNIX, type=socket.SOCK_STREAM, proto=0)
ipc_provider_workaround = s.connect(self.ipcaddr)
logg.info('redirecting method {} params {} original params {}'.format(method, params, suspect_params))
o = jsonrpc_request(method, params[0])
j = json.dumps(o)
logg.debug('send {}'.format(j))
s.send(j.encode('utf-8'))
r = s.recv(4096)
s.close()
logg.debug('got recv {}'.format(str(r)))
jr = json.loads(r)
jr['id'] = self.id_seq
return jr
r = self.make_request(method, suspect_params)
return r