-
Notifications
You must be signed in to change notification settings - Fork 0
/
consensus.py
272 lines (209 loc) · 9.89 KB
/
consensus.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
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
# Github https://github.com/Northa
import sys
from urllib import request
import math
from sys import exit
from json import loads
GREEN_BLOCK = '🟩'
RED_BLOCK = '🟥'
ERR_MSG = f"\033[91m[ERR] API endpoint unreachable: api\n" \
f"[ERR] Be sure you have enabled your API " \
f"(you can enable this in your app.toml config file)\n" \
f"Bugreports Discord: https://github.com/stakingcabin/celestia-task.git\033[0m"
# default ports
REST = "http://127.0.0.1:1317"
RPC = "http://127.0.0.1:26657"
def handle_request(api: str, pattern: str):
try:
response = loads(request.urlopen(f"{api}/{pattern}").read())
return response if response is not None else exit(ERR_MSG.replace('api', api))
except Exception:
exit(ERR_MSG.replace('api', api))
def get_validator_votes():
validator_votes = []
votes = STATE['result']['round_state']['votes']
height = STATE['result']['round_state']['height']
step = STATE['result']['round_state']['step']
chain = get_chain_id()
for r_ound in votes:
if float(r_ound['precommits_bit_array'].split('=')[-1].strip()) <= 0.66 and float(r_ound['precommits_bit_array'].split('=')[-1].strip()) > 0:
print(F"\nChain-id: {chain}\n"
f"Height: {height} Round: {r_ound['round']} "
f"step: {step}\n"
f"precommits_bit_array: {r_ound['precommits_bit_array'].split('} ')[-1]}")
for precommit in r_ound['precommits']:
try:
validator_votes.append(precommit.split('@')[0].split(':')[1].split(' ')[0])
except IndexError:
validator_votes.append(precommit)
elif float(r_ound['prevotes_bit_array'].split('=')[-1].strip()) <= 0.66 and float(r_ound['prevotes_bit_array'].split('=')[-1].strip()) > 0:
print(F"\nChain-id: {chain}\nHeight: {height} Round: {r_ound['round']} step: {step}\nprevotes_bit_array: {r_ound['prevotes_bit_array'].split('} ')[-1]}")
for precommit in r_ound['prevotes']:
try:
validator_votes.append(precommit.split('@')[0].split(':')[1].split(' ')[0])
except IndexError:
validator_votes.append(precommit)
if len(validator_votes) == 0:
commit_votes = STATE['result']['round_state']['last_commit']
height = STATE['result']['round_state']['height']
if float(commit_votes['votes_bit_array'].split('=')[-1].strip()) > 0.66 and float(commit_votes['votes_bit_array'].split('=')[-1].strip()) > 0:
print(F"\nChain-id: {chain}\nHeight: {height} Round: {r_ound['round']} step: {step}\nvotes_bit_array: {commit_votes['votes_bit_array'].split('} ')[-1]}")
for commit_vote in commit_votes['votes']:
try:
validator_votes.append(commit_vote.split('@')[0].split(':')[1].split(' ')[0])
except IndexError:
validator_votes.append(commit_vote)
return validator_votes
def get_validators():
validators = []
state_validators = STATE['result']['round_state']['validators']['validators']
for val in state_validators:
res = val['address'], val['voting_power'], val['pub_key']['value']
validators.append(res)
return validators
def get_bonded():
result = handle_request(REST, '/cosmos/staking/v1beta1/pool')['pool']
return result
def strip_emoji_non_ascii(moniker):
# moniker = emoji.replace_emoji(moniker, replace='')
moniker = "".join([letter for letter in moniker if letter.isascii()])
return moniker[:15].strip().lstrip()
def get_validators_rest():
validator_dict = dict()
bonded_tokens = int(get_bonded()["bonded_tokens"])
validators = handle_request(REST, '/cosmos/staking/v1beta1/validators?status=BOND_STATUS_BONDED&pagination.limit=2000')
# print(type(validators))
# for i in validators:print(i)
for validator in validators['validators']:
# print(validator)
validator_vp = int(validator["tokens"])
vp_percentage = round((100 / bonded_tokens) * validator_vp, 3)
moniker = validator["description"]["moniker"][:15].strip()
moniker = strip_emoji_non_ascii(moniker)
validator_dict[validator["consensus_pubkey"]["key"]] = {
"moniker": moniker,
"address": validator["operator_address"],
"status": validator["status"],
"voting_power": validator_vp,
"voting_power_perc": f"{vp_percentage}%"}
return validator_dict, len(validators['validators'])
def merge_info():
votes = get_validator_votes()
validators = get_validators()
votes_and_vals = list(zip(votes, validators))
validator_rest, total_validators = get_validators_rest()
final_list = []
for k, v in votes_and_vals:
if v[2] in validator_rest:
validator_rest[v[2]]['voted'] = k
final_list.append(validator_rest[v[2]])
return final_list, total_validators
def list_columns(obj, cols=3, columnwise=True, gap=8):
# thnx to https://stackoverflow.com/a/36085705
sobj = [str(item) for item in obj]
if cols > len(sobj): cols = len(sobj)
max_len = max([len(item) for item in sobj])
if columnwise: cols = int(math.ceil(float(len(sobj)) / float(cols)))
plist = [sobj[i: i+cols] for i in range(0, len(sobj), cols)]
if columnwise:
if not len(plist[-1]) == cols:
plist[-1].extend(['']*(len(sobj) - len(plist[-1])))
plist = zip(*plist)
printer = '\n'.join([
''.join([c.ljust(max_len + gap) for c in p])
for p in plist])
return printer
def get_chain_id():
response = handle_request(REST, '/cosmos/base/tendermint/v1beta1/node_info')
# for i in response:print()
chain_id = response['default_node_info']['network'] if 'default_node_info' in response else response['node_info']['network']
return chain_id
def colorize_output(validators):
result = []
for num, val in enumerate(validators):
vp_perc = f"{val['voting_power_perc']:<7}"
moniker = val['moniker']
if val['voted'] != 'nil-Vote':
stat = f"\033[92m{num+1:<3} {'ONLINE':<8} \033[0m"
result.append(f"{stat} {moniker:<18} {vp_perc}")
else:
stat = f"\033[91m{num+1:<3} {'OFFLINE':<8} \033[0m"
result.append(f"{stat} {moniker:<18} {vp_perc}")
return result
def calculate_colums(result):
if len(result) <= 30:
return list_columns(result, cols=1)
elif 30 < len(result) <= 100:
return list_columns(result, cols=2)
elif 100 < len(result) <= 150:
return list_columns(result, cols=3)
else:
return list_columns(result, cols=4)
def get_pubkey_by_valcons(valcons, height):
response = handle_request(REST, f"/validatorsets/{height}")
for validator in response['result']['validators']:
if valcons in validator['address']:
return validator['pub_key']['value']
def get_moniker_by_pub_key(pub_key, height):
response = handle_request(REST, f"cosmos/staking/v1beta1/historical_info/{height}")
for validator in response['hist']['valset']:
if pub_key in validator['consensus_pubkey']['key']:
return strip_emoji_non_ascii(validator['description']['moniker'])
def get_evidence(height):
evidences = handle_request(REST, '/cosmos/evidence/v1beta1/evidence')
for evidence in evidences['evidence']:
if int(height) - int(evidence['height']) < 1000:
pub_key = get_pubkey_by_valcons(evidence['consensus_address'], evidence['height']).strip()
moniker = get_moniker_by_pub_key(pub_key, evidence['height'])
# print(colored(f"Evidence: {moniker}\nHeight: {evidence['height']} {evidence['consensus_address']} power: {evidence['power']}\n", 'yellow'))
print(f"\033[93mEvidence: {moniker}\nHeight: {evidence['height']} {evidence['consensus_address']} power: {evidence['power']}\033[0m\n")
def main(STATE):
validators, total_validators = merge_info()
online_vals = 0
for num, val in enumerate(validators):
if val['voted'] != 'nil-Vote':
online_vals += 1
print(f"Online: {online_vals}/{total_validators}\n")
# get_evidence(STATE['result']['round_state']['height'])
result = colorize_output(validators)
print(calculate_colums(result))
def block_check():
try:
r = handle_request(RPC, "/status")
local_height = r['result']['sync_info']['latest_block_height']
VALIDATOR_ADDRESS = r['result']['validator_info']['address']
signed = False
print("check height at {}-{}".format(int(local_height)-99, int(local_height)))
for offset in range(100):
check_height = int(local_height) - (99-offset)
response = handle_request(RPC, f'/block?height={check_height}')
block = response['result']['block']
# Check if the validator signed the block
signed = False
for vote in block['last_commit']['signatures']:
if vote['validator_address'] == VALIDATOR_ADDRESS:
signed = True
break
print(GREEN_BLOCK if signed else RED_BLOCK, end='')
if (offset+1) % 10 == 0:
print("")
print("")
except Exception as e:
print("block check exception occurred:", e)
Usage = '''
python3 consensus.py --help :
show usage
python3 consensus.py sign-check :
show local validator sign status for recent 100 blocks
python3 consensus.py :
show validators consensus staus
'''
if __name__ == '__main__':
if len(sys.argv) == 2 and sys.argv[1] == "--help":
print(Usage)
exit(0)
if len(sys.argv) == 2 and sys.argv[1] == "sign-check":
block_check()
exit(0)
STATE = handle_request(RPC, 'dump_consensus_state')
exit(main(STATE))