forked from blockparty-sh/Fulcrum
-
Notifications
You must be signed in to change notification settings - Fork 1
/
FulcrumAdmin
executable file
·460 lines (422 loc) · 19.5 KB
/
FulcrumAdmin
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
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
#!/usr/bin/env python3
import argparse
import json
import random
import socket
import sys
from collections import defaultdict
HOST = 'localhost'
PORT = None
ID_NEXT = random.randint(0, 262144)
JSON = False
EXITSTATUS = 0
class ErrorResponse(RuntimeError):
pass
def send_request(method, params=None):
verb = False#not args.q
with socket.create_connection((HOST, PORT), timeout=10.0) as sock:
def sndrecv(_id, method, params=None):
outj = { "id" : _id, "jsonrpc" : "2.0", "method" : method, "params": params or [] }
msg = json.dumps(outj, indent=None).encode("utf8") + b'\n'
if verb: print(f"Srv --> {msg[:2048]}")
sock.send(msg)
resp = bytearray()
while b'\n' not in resp:
resp += sock.recv(4096)
if verb: print(f"Srv <-- {resp[:2048]}")
j = json.loads(resp.decode("utf8").strip())
if j.get("error") or j.get("id") != _id:
raise ErrorResponse("Error response from Fulcrum:\n\n" + (json.dumps(j.get("error"), indent=4) or j))
return j.get("result")
global ID_NEXT
reqid = ID_NEXT; ID_NEXT += 1
return sndrecv(reqid, method, params)
def main():
global HOST, PORT, JSON, EXITSTATUS
parser = argparse.ArgumentParser(prog="FulcrumAdmin.py", description="Fulcrum CLI admin tool")
parser.add_argument('-p', type=int, metavar="port", nargs=1, required=True, help=f"Specify the port for the Fulcrum admin RPC service. This is a required argument.")
parser.add_argument('-j', action='store_true', dest="json", help=f"Print the response from the server as JSON, not as formatted text")
parser.add_argument('-H', type=str, metavar="host", nargs='?', default=HOST, help=f"Specify the host for the Fulcrum admin RPC service. Defaults to {HOST}.")
subparsers = parser.add_subparsers(title="command", description="Select from one of the following commands:", dest="command")
addpeer = subparsers.add_parser('addpeer', help="Add a peer to the server's list of peers")
addpeer.add_argument('hostname', metavar='hostname', nargs=1, help="Hostname of peer.")
addpeer.add_argument('-s', metavar='ssl_port', type=int, nargs='?', help="Peer's SSL port.")
addpeer.add_argument('-t', metavar='tcp_port', type=int, nargs='?', help="Peer's TCP port.")
ban = subparsers.add_parser('ban', help="Ban clients by ID and/or IP address")
ban.add_argument('id_or_ip', metavar='ipaddress_or_id', nargs='+', help="Client ID or IP address to ban.")
banpeer = subparsers.add_parser('banpeer', help="Ban peers by hostname suffix")
banpeer.add_argument('hostnames', metavar='hostname', nargs='+', help="A hostname or hostname suffix e.g. somehost.com or *some.host.com.")
bitcoind_throttle = subparsers.add_parser('bitcoind_throttle', help="Query or set server bitcoind_throttle setting")
bitcoind_throttle.add_argument('param', metavar='param', nargs='*', help='The new desired setting. Specify 3 arguments to set this properly for: high low decay. Omit arguments to query.')
clients = subparsers.add_parser('clients', help="Print information on all the currently connected clients", aliases=['sessions'])
getinfo = subparsers.add_parser('getinfo', help="Get server information")
kick = subparsers.add_parser('kick', help="Kick clients by ID and/or IP address")
kick.add_argument('id_or_ip', metavar='ipaddress_or_id', nargs='+', help="Client ID or IP addresses to kick.")
listbanned = subparsers.add_parser('listbanned', help="Print the list of banned IP addresses and peer hostnames", aliases=['banlist'])
loglevel = subparsers.add_parser('loglevel', help="Set the server's logging verbosity")
loglevel.add_argument('level', metavar='level', nargs=1, help="One of: 'normal', 'debug', or 'trace'")
maxbuffer = subparsers.add_parser('maxbuffer', help="Query or set server max_buffer setting")
maxbuffer.add_argument('bytes', metavar='bytes', type=int, nargs='?', help='The new desired max_buffer setting in bytes. Must be >= 64KB and <= 100MB. If omitted, then this script will just query the current value.')
peers = subparsers.add_parser('peers', help="Print peering information")
rmpeer = subparsers.add_parser('rmpeer', help="Remove peers by hostname suffix")
rmpeer.add_argument('hostnames', metavar='hostname', nargs='+', help="A hostname or hostname suffix e.g. somehost.com or *some.host.com.")
simdjson = subparsers.add_parser('simdjson', help="Get or set the server's 'simdjson' (JSON parser) setting")
simdjson.add_argument('enabled', type=int, nargs='?',
help='Flag used to enable or disable the simdjson JSON parser on the server (1=enabled,'
' 0=disabled). If this option is omitted, then the current setting is queried.')
stop = subparsers.add_parser('stop', help="Gracefully shut down the server", aliases=['shutdown'])
unban = subparsers.add_parser('unban', help="Unban IP addresses")
unban.add_argument('ips', metavar='ipaddress', nargs='+', help="Specify an existing banned IP address to unban.")
unbanpeer = subparsers.add_parser('unbanpeer', help="Unban peers by hostname suffix")
unbanpeer.add_argument('hostnames', metavar='hostname', nargs='+', help="Specify an existing peer ban to unban.")
args = parser.parse_args()
HOST = args.H
PORT, = args.p
JSON = args.json
if PORT > 65535:
sys.exit("Port argument must be < 65536")
command = args.command
if command == 'sessions': command = 'clients' # Is there a better way to do this by referring back to the subparser above?? TODO
if command == 'shutdown': command = 'stop'
if command == 'banlist' : command = 'listbanned'
command_params = tuple()
response_handler = lambda r: json.dumps(r, indent = 4) # default handler just pretty-prints the JSON
if command is None:
print("Please specify a command to run.\n")
parser.print_help()
sys.exit(1)
elif command == 'getinfo':
orig_handler = response_handler
def handler(r):
# mogrify the uptime field to be more useful to humans
if r.get('uptime'):
r['uptime'] = formatTimeField(r['uptime'])
return orig_handler(r)
response_handler = handler
elif command == 'stop' and not JSON:
def handler(r):
if isinstance(r, bool) and r:
return "Fulcrum server is shutting down"
else:
global EXITSTATUS
EXITSTATUS = 1
return f"Unexpected response from Fulcrum server: {r}"
response_handler = handler
elif command == 'clients' and not JSON:
response_handler = clients_handler
elif command == 'maxbuffer':
command_params = [args.bytes] if args.bytes else command_params
if not JSON:
def handler(x):
if command_params:
return f"Server max_buffer setting -> {x}"
else:
return f"Server max_buffer setting is: {x}"
response_handler = handler
elif command == 'simdjson':
command_params = [bool(args.enabled)] if args.enabled is not None else command_params
if not JSON:
def handler(x):
if command_params:
# set
assert isinstance(x, bool)
if x:
verb = "enabled" if command_params[0] else "disabled"
return f"Server simdjson parser {verb} successfully"
else:
verb = "enable" if command_params[0] else "disable"
return f"Failed to {verb} server simdjson parser"
else:
# query
if x is not None:
return json.dumps(x, indent=4)
else:
return "Server simdjson parser is disabled"
response_handler = handler
elif command == 'bitcoind_throttle':
command_params = args.param if len(args.param) else command_params
if not JSON:
def handler(x):
names = defaultdict(lambda: "???")
names.update({ 0: "high", 1: "low", 2: "decay" })
if command_params:
return f"Server bitcoind_throttle setting -> " + ', '.join(names[n] + ' = ' + str(i) for n,i in enumerate(x))
else:
return "Server bitcoind_throttle setting is: " + ', '.join(names[n] + ' = ' + str(i) for n,i in enumerate(x))
response_handler = handler
elif command in ('kick', 'ban', 'banpeer', 'unban', 'unbanpeer', 'rmpeer', 'loglevel'):
extratxt = ''
if command in ('kick', 'ban'):
command_params = args.id_or_ip
elif command in ('banpeer', 'unbanpeer', 'rmpeer'):
command_params = args.hostnames
if command == 'rmpeer':
extratxt = ('''
Note: Removal of 'Good' peers is not guaranteed to keep them from re-peering
with the server in the near future. If you want the specified peer(s) to never
possibly peer with this Fulcrum server, then please use the 'banpeer' command.
''')
elif command == 'unban':
command_params = args.ips
elif command == 'loglevel':
levels = { 'normal': 0, 'debug' : 1, 'trace' : 2}
if args.level[0] not in levels:
print("level argument must be one of:", ', '.join(levels.keys()))
sys.exit(1)
command_params = [levels[args.level[0]]]
extratxt = f' -> {args.level[0]}'
if not JSON:
def handler(r):
if isinstance(r, bool) and r:
return f"{command} command submitted for: " + ', '.join([str(x) for x in command_params]) + extratxt
else:
global EXITSTATUS
EXITSTATUS = 1
return f"Unexpected response from Fulcrum server: {r}"
response_handler = handler
elif command == 'listbanned' and not JSON:
response_handler = listbanned_handler
elif command == 'peers' and not JSON:
response_handler = peers_handler
elif command == 'addpeer':
if not args.s and not args.t:
print("addpeer requires at least one of the two port arguments (-s, -t), or both.")
sys.exit(1)
host = args.hostname[0]
command_params = {
'host' : host,
'ssl' : args.s or 0,
'tcp' : args.t or 0,
}
if not JSON:
def handler(r):
if isinstance(r, bool) and r:
return f"{command} command submitted for: {host}"
else:
global EXITSTATUS
EXITSTATUS = 1
return f"Unexpected response from Fulcrum server: {r}"
response_handler = handler
try:
print(response_handler(send_request(command, command_params)))
sys.exit(EXITSTATUS)
except OSError as e:
print(f"Error communicating with the admin RPC port at {HOST}:{PORT}\n\n {e}\n")
sys.exit(1)
except ErrorResponse as e:
print(f"{e}")
sys.exit(1)
def clients_handler(r):
def badResp():
global EXITSTATUS
EXITSTATUS = 1
return f"Unexpected resposne from Fulcrum server: {r!r}"
if not isinstance(r, (list, tuple)):
return badResp()
lines = []
line = ("ID","IP:PORT","Typ","UAgent","ProtocolVer","Subs","HdrSub?","ReqRcv","RespSent","RecvBytes","SentBytes","TxsSent","Notifs","ErrorCt","Elapsed")
maxfields = defaultdict(int)
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for serverdict in r:
if not isinstance(serverdict, dict):
return badResp()
for server_name, subdict in serverdict.items():
if not isinstance(subdict, dict):
# in case we add stuff to here that is not a server dict some day
continue;
is_wss = server_name.lower().startswith('wsss')
is_ws = not is_wss and server_name.lower().startswith('wss')
if not is_wss and not is_ws:
is_ssl = server_name.lower().startswith('ssl')
typnam = 'SSL' if is_ssl else 'TCP'
else:
typnam = 'WS' if is_ws else 'WSS'
for client in subdict.get('clients', []):
for cname, cdict in client.items():
line = (
cdict.get('id', -1),
cdict.get('remote', '?'),
typnam,
cdict.get('userAgent', 'Unk'),
cdict.get('version', ['?'])[-1],
cdict.get('nSubscriptions', -1),
'Y' if cdict.get('isSubscribedToHeaders') else 'N',
cdict.get('nRequestsRcv', -1),
cdict.get('nResultsSent', -1),
cdict.get('nBytesReceived', -1),
cdict.get('nBytesSent', -1),
cdict.get('nTxSent', -1),
cdict.get('nNotificationsSent', -1),
cdict.get('nErrorsSent', -1),
formatTimeField( cdict.get('connectedTime', '-') )
)
line = [str(x) for x in line]
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for i,line in enumerate(list(lines)):
line = list(line)
for j,c in enumerate(line):
line[j] = c.ljust(maxfields[j])
lines[i] = ' '.join(line)
return '\n'.join(lines) + '\n'
def listbanned_handler(r):
def badResp():
global EXITSTATUS
EXITSTATUS = 1
return f"Unexpected resposne from Fulcrum server: {r!r}"
if not isinstance(r, (dict,)):
return badResp()
lines = []
line = ("IP","AgeSecs","RejectedConnections",)
maxfields = defaultdict(int)
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for ipaddr, subdict in r.get('Banned_IPAddrs', {}).items():
if not isinstance(subdict, dict):
return badResp()
line = (
ipaddr,
subdict.get('age_secs', -1),
subdict.get('connections_rejected', 0),
)
line = [str(x) for x in line]
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for i,line in enumerate(list(lines)):
line = list(line)
for j,c in enumerate(line):
line[j] = c.ljust(maxfields[j])
lines[i] = ' '.join(line)
clientPart = '~Client Bans~\n' + '\n'.join(lines)
maxFields = defaultdict(int)
lines = []
line = ("HostName","AgeSecs",)
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for hostname, subdict in r.get('Banned_Peers', {}).items():
if not isinstance(subdict, dict):
return badResp()
line = (
hostname,
subdict.get('age_secs', -1),
)
line = [str(x) for x in line]
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for i,line in enumerate(list(lines)):
line = list(line)
for j,c in enumerate(line):
line[j] = c.ljust(maxfields[j])
lines[i] = ' '.join(line)
peerPart = '~Peer Bans~\n' + '\n'.join(lines)
return clientPart + '\n\n' + peerPart
def peers_handler(r):
def badResp():
global EXITSTATUS
EXITSTATUS = 1
return f"Unexpected resposne from Fulcrum server: {r!r}"
if not isinstance(r, (dict,)):
return badResp()
lines = []
line = ("Hostname","IP","Status","TCP","SSL","Version","ProtoMin","ProtoMax","Elapsed","Message","RetryPeriod")
maxfields = defaultdict(int)
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
retryTimes = {
"bad" : r.get('activeTimers', {}).get("badPeerRetry", 0),
"failed" : r.get('activeTimers', {}).get("failedPeerRetry", 0),
}
# peers (good, connected peers)
peerdict = r.get('peers')
if not isinstance(peerdict, dict):
return badResp()
for hostname, d in peerdict.items():
line = (
hostname,
d.get('addr','-'),
'Good' if d.get('verified') == True else 'Verifying',
d.get('tcp_port') if d.get('tcp_port') else '-',
d.get('ssl_port') if d.get('ssl_port') else '-',
d.get('server_version', 'Unk'),
d.get('protocol_min', '-'),
d.get('protocol_max', '-'),
formatTimeField( d.get('connectedTime', '-') ),
'-', '-'
)
line = [str(x) for x in line]
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
# next, do the failed, bad, and queued peers
for status in ('failed', 'bad', 'queued'):
d = r.get(status)
if not d:
continue # skip empties
if not isinstance(d, dict):
return badResp()
for hostname, l in d.items():
try:
ip, tcp, ssl, subver, pver, time, msg = l
except ValueError:
pass
line = (
hostname, ip, status.title(), tcp if tcp else '-', ssl if ssl else '-',
subver, pver, pver, formatTimeField(time), msg, '~' + str(retryTimes.get(status, 0)//1000) + 's'
)
line = [str(x) for x in line]
for i,c in enumerate(line):
maxfields[i] = max(maxfields[i], len(c))
lines.append(line)
for i,line in enumerate(list(lines)):
line = list(line)
for j,c in enumerate(line):
line[j] = c.ljust(maxfields[j])
lines[i] = ' '.join(line)
return '\n'.join(lines) + '\n'
def formatTimeField(s):
if s is None:
return '-' # transform None to '-' since it takes up fewer characters
secs_per_hour = 60.0 * 60.0
if isinstance(s, str):
to_secs_factor = 1.0
if s.endswith("hours"): # one of the "xx hours" fields
to_secs_factor = secs_per_hour
elif s.endswith("secs"):
pass
else:
return s # unknown field type, just return it verbatim
try:
s = float(s.split()[0].strip()) * to_secs_factor # transform hours to seconds
except (ValueError, TypeError, IndexError):
return s
del to_secs_factor
# at this point s should be in seconds.. if not, give up and just return it
if not isinstance(s, (int, float)):
return s
# now turn seconds to minutes, hours, days, months, years
secs_per_day = secs_per_hour * 24.0
secs_per_year = 365.0 * secs_per_day
secs_per_month = 30.0 * secs_per_day
secs_per_minute = 60.0
if s >= secs_per_year:
return f'{s/secs_per_year:0.3f} years'
elif s >= secs_per_month:
return f'{s/secs_per_month:0.3f} months'
elif s >= secs_per_day:
return f'{s/secs_per_day:0.2f} days'
elif s >= secs_per_hour:
return f'{s/secs_per_hour:0.2f} hours'
elif s >= secs_per_minute:
return f'{s/secs_per_minute:0.1f} mins'
else:
return f'{s:0.1f} secs'
if __name__ == '__main__':
main()