-
Notifications
You must be signed in to change notification settings - Fork 37
/
util.py
136 lines (109 loc) · 4.5 KB
/
util.py
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
#
# Copyright 2023 Ocean Protocol Foundation
# SPDX-License-Identifier: Apache-2.0
#
import hashlib
import json
import logging
from typing import Tuple
import werkzeug
from eth_account.signers.local import LocalAccount
from eth_keys import KeyAPI
from eth_keys.backends import NativeECCBackend
from eth_typing.encoding import HexStr
from ocean_provider.utils.asset import Asset
from ocean_provider.utils.basics import get_network_name
from ocean_provider.utils.encryption import do_decrypt
from ocean_provider.utils.services import Service
from web3 import Web3
from web3.types import TxParams, TxReceipt
logger = logging.getLogger(__name__)
keys = KeyAPI(NativeECCBackend)
def get_request_data(request):
try:
return request.args if request.args else request.json
except werkzeug.exceptions.BadRequest:
return {}
def msg_hash(message: str):
return hashlib.sha256(message.encode("utf-8")).hexdigest()
def get_service_files_list(
service: Service, provider_wallet: LocalAccount, asset: Asset = None
) -> list:
version = asset.version if asset is not None and asset.version else "4.0.0"
if asset is None or version == "4.0.0":
return get_service_files_list_old_structure(service, provider_wallet)
network_name = get_network_name(asset.chain_id)
try:
files_str = do_decrypt(service.encrypted_files, provider_wallet)
if not files_str:
return None
files_json = json.loads(files_str)
for key in ["datatokenAddress", "nftAddress", "files"]:
if key not in files_json:
raise Exception(
f"Provider {network_name}: Key {key} not found in files."
)
if Web3.toChecksumAddress(
files_json["datatokenAddress"]
) != Web3.toChecksumAddress(service.datatoken_address):
raise Exception(
f"Provider {network_name}: Mismatch of datatoken. Got {files_json['datatokenAddress']} vs expected {service.datatoken_address}"
)
if Web3.toChecksumAddress(files_json["nftAddress"]) != Web3.toChecksumAddress(
asset.nftAddress
):
raise Exception(
f"Provider {network_name}: Mismatch of dataNft. Got {files_json['nftAddress']} vs expected {asset.nftAddress}"
)
files_list = files_json["files"]
if not isinstance(files_list, list):
raise TypeError(
f"Provider {network_name}: Expected a files list, got {type(files_list)}."
)
return files_list
except Exception as e:
logger.error(
f"Provider {network_name}: Error decrypting service files {Service}: {str(e)}"
)
return None
def get_service_files_list_old_structure(
service: Service, provider_wallet: LocalAccount
) -> list:
try:
files_str = do_decrypt(service.encrypted_files, provider_wallet)
if not files_str:
return None
logger.debug(f"Got decrypted files str {files_str}")
files_list = json.loads(files_str)
if not isinstance(files_list, list):
raise TypeError(f"Expected a files list, got {type(files_list)}.")
return files_list
except Exception as e:
logger.error(f"Error decrypting service files {Service}: {str(e)}")
return None
def sign_tx(web3, tx, private_key):
"""
:param web3: Web3 object instance
:param tx: transaction
:param private_key: Private key of the account
:return: rawTransaction (str)
"""
account = web3.eth.account.from_key(private_key)
nonce = web3.eth.get_transaction_count(account.address)
tx["nonce"] = nonce
signed_tx = web3.eth.account.sign_transaction(tx, private_key)
return signed_tx.rawTransaction
def sign_and_send(
web3: Web3, transaction: TxParams, from_account: LocalAccount
) -> Tuple[HexStr, TxReceipt]:
"""Returns the transaction id and transaction receipt."""
transaction_signed = sign_tx(web3, transaction, from_account.key)
transaction_hash = web3.eth.send_raw_transaction(transaction_signed)
transaction_id = Web3.toHex(transaction_hash)
return transaction_hash, transaction_id
def sign_send_and_wait_for_receipt(
web3: Web3, transaction: TxParams, from_account: LocalAccount
) -> Tuple[HexStr, TxReceipt]:
"""Returns the transaction id and transaction receipt."""
transaction_hash, transaction_id = sign_and_send(web3, transaction, from_account)
return (transaction_id, web3.eth.wait_for_transaction_receipt(transaction_hash))