forked from ton-blockchain/mytonctrl
-
Notifications
You must be signed in to change notification settings - Fork 0
/
custom_overlays.py
191 lines (162 loc) · 6.92 KB
/
custom_overlays.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
import base64
import json
import requests
from mypylib.mypylib import color_print
def hex2base64(h):
b = bytes.fromhex(h)
b64 = base64.b64encode(b)
s = b64.decode("utf-8")
return s
def parse_config(name: str, config: dict, vset: list = None):
"""
Converts config to validator-console friendly format
:param name: custom overlay name
:param config: config
:param vset: list of validators adnl addresses, can be None if `@validators` not in config
:return:
"""
result = {
"name": name,
"nodes": []
}
for k, v in config.items():
if k == '@validators' and v:
if vset is None:
raise Exception("Validators set is not defined but @validators is in config")
for v_adnl in vset:
result["nodes"].append({
"adnl_id": hex2base64(v_adnl),
"msg_sender": False,
})
else:
result["nodes"].append({
"adnl_id": hex2base64(k),
"msg_sender": v["msg_sender"],
})
if v["msg_sender"]:
result["nodes"][-1]["msg_sender_priority"] = v["msg_sender_priority"]
return result
def add_custom_overlay(args):
from mytonctrl import ton, local
if len(args) != 2:
color_print("{red}Bad args. Usage:{endc} add_custom_overlay <name> <path_to_config>")
return
path = args[1]
with open(path, 'r') as f:
config = json.load(f)
ton.set_custom_overlay(args[0], config)
if '@validators' in config:
print('Dynamic overlay will be added within 1 minute')
else:
result = add_custom_overlay_to_vc(local, ton, parse_config(args[0], config))
if not result:
print('Failed to add overlay to validator console')
color_print("add_custom_overlay - {red}ERROR{endc}")
return
color_print("add_custom_overlay - {green}OK{endc}")
def list_custom_overlays(args):
from mytonctrl import ton
if not ton.get_custom_overlays():
color_print("{red}No custom overlays{endc}")
return
for k, v in ton.get_custom_overlays().items():
color_print(f"Custom overlay {{bold}}{k}{{endc}}:")
print(json.dumps(v, indent=4))
def delete_custom_overlay(args):
from mytonctrl import ton
if len(args) != 1:
color_print("{red}Bad args. Usage:{endc} delete_custom_overlay <name>")
return
if '@validators' in ton.get_custom_overlays().get(args[0], {}):
ton.delete_custom_overlay(args[0])
print('Dynamic overlay will be deleted within 1 minute')
else:
ton.delete_custom_overlay(args[0])
result = delete_custom_overlay_from_vc(ton, args[0])
if not result:
print('Failed to delete overlay from validator console')
color_print("delete_custom_overlay - {red}ERROR{endc}")
return
color_print("delete_custom_overlay - {green}OK{endc}")
def check_node_eligible_for_custom_overlay(ton, config: dict):
vconfig = ton.GetValidatorConfig()
my_adnls = vconfig.adnl
node_adnls = [i["adnl_id"] for i in config["nodes"]]
for adnl in my_adnls:
if adnl.id in node_adnls:
return True
return False
def delete_custom_overlay_from_vc(ton, name: str):
result = ton.validatorConsole.Run(f"delcustomoverlay {name}")
return 'success' in result
def add_custom_overlay_to_vc(local, ton, config: dict):
if not check_node_eligible_for_custom_overlay(ton, config):
local.add_log(f"Node has no adnl address required for custom overlay {config.get('name')}", "debug")
return False
local.add_log(f"Adding custom overlay {config.get('name')}", "debug")
path = ton.tempDir + f'/custom_overlay_{config["name"]}.json'
with open(path, 'w') as f:
json.dump(config, f)
result = ton.validatorConsole.Run(f"addcustomoverlay {path}")
return 'success' in result
def custom_overlays(local, ton):
config = get_default_custom_overlay(local, ton)
if config is not None:
ton.set_custom_overlay('default', config)
deploy_custom_overlays(local, ton)
def deploy_custom_overlays(local, ton):
result = ton.validatorConsole.Run("showcustomoverlays")
if 'unknown command' in result:
return # node old version
names = []
for line in result.split('\n'):
if line.startswith('Overlay'):
names.append(line.split(' ')[1].replace('"', '').replace(':', ''))
config34 = ton.GetConfig34()
current_el_id = config34['startWorkTime']
current_vset = [i["adnlAddr"] for i in config34['validators']]
config36 = ton.GetConfig36()
next_el_id = config36['startWorkTime'] if config36['validators'] else 0
next_vset = [i["adnlAddr"] for i in config36['validators']]
for name in names:
# check that overlay still exists in mtc db
pure_name = name
suffix = name.split('_')[-1]
if suffix.startswith('elid') and suffix.split('elid')[-1].isdigit(): # probably election id
pure_name = '_'.join(name.split('_')[:-1])
el_id = int(suffix.split('elid')[-1])
if el_id not in (current_el_id, next_el_id):
local.add_log(f"Overlay {name} is not in current or next election, deleting", "debug")
delete_custom_overlay_from_vc(ton, name) # delete overlay if election id is not in current or next election
continue
if pure_name not in ton.get_custom_overlays():
local.add_log(f"Overlay {name} ({pure_name}) is not in mtc db, deleting", "debug")
delete_custom_overlay_from_vc(ton, name) # delete overlay if it's not in mtc db
for name, config in ton.get_custom_overlays().items():
if name in names:
continue
if '@validators' in config:
new_name = name + '_elid' + str(current_el_id)
if new_name not in names:
node_config = parse_config(new_name, config, current_vset)
add_custom_overlay_to_vc(local, ton, node_config)
if next_el_id != 0:
new_name = name + '_elid' + str(next_el_id)
if new_name not in names:
node_config = parse_config(new_name, config, next_vset)
add_custom_overlay_to_vc(local, ton, node_config)
else:
node_config = parse_config(name, config)
add_custom_overlay_to_vc(local, ton, node_config)
def get_default_custom_overlay(local, ton):
if not local.db.get('useDefaultCustomOverlays', True):
return None
network = ton.GetNetworkName()
default_url = 'https://ton-blockchain.github.io/fallback_custom_overlays.json'
url = local.db.get('defaultCustomOverlaysUrl', default_url)
resp = requests.get(url)
if resp.status_code != 200:
local.add_log(f"Failed to get default custom overlays from {url}", "error")
return None
config = resp.json()
return config.get(network)