Page MenuHomePhabricator
Differential D13891 Diff 40299 contrib/linearize/linearize-hashes.py

Changeset View

Standalone View

View Options

contrib/linearize/linearize-hashes.py

Show All 16 Lines
import re import re
import sys import sys
from typing import Any, Dict from typing import Any, Dict
settings: Dict[str, Any] = {} settings: Dict[str, Any] = {}
def hex_switchEndian(s): def hex_switchEndian(s):
""" Switches the endianness of a hex string (in pairs of hex chars) """ """Switches the endianness of a hex string (in pairs of hex chars)"""
pairList = [s[i:i + 2].encode() for i in range(0, len(s), 2)] pairList = [s[i : i + 2].encode() for i in range(0, len(s), 2)]
return b''.join(pairList[::-1]).decode() return b"".join(pairList[::-1]).decode()
class BitcoinRPC: class BitcoinRPC:
def __init__(self, host, port, username, password): def __init__(self, host, port, username, password):
authpair = f"{username}:{password}" authpair = f"{username}:{password}"
authpair = authpair.encode('utf-8') authpair = authpair.encode("utf-8")
self.authhdr = b"Basic " + base64.b64encode(authpair) self.authhdr = b"Basic " + base64.b64encode(authpair)
self.conn = httplib.HTTPConnection(host, port=port, timeout=30) self.conn = httplib.HTTPConnection(host, port=port, timeout=30)
def execute(self, obj): def execute(self, obj):
try: try:
self.conn.request('POST', '/', json.dumps(obj), self.conn.request(
{'Authorization': self.authhdr, "POST",
'Content-type': 'application/json'}) "/",
json.dumps(obj),
{"Authorization": self.authhdr, "Content-type": "application/json"},
)
except ConnectionRefusedError: except ConnectionRefusedError:
print('RPC connection refused. Check RPC settings and the server status.', print(
file=sys.stderr) "RPC connection refused. Check RPC settings and the server status.",
file=sys.stderr,
)
return None return None
resp = self.conn.getresponse() resp = self.conn.getresponse()
if resp is None: if resp is None:
print("JSON-RPC: no response", file=sys.stderr) print("JSON-RPC: no response", file=sys.stderr)
return None return None
body = resp.read().decode('utf-8') body = resp.read().decode("utf-8")
resp_obj = json.loads(body) resp_obj = json.loads(body)
return resp_obj return resp_obj
@staticmethod @staticmethod
def build_request(idx, method, params): def build_request(idx, method, params):
obj = {'version': '1.1', obj = {"version": "1.1", "method": method, "id": idx}
'method': method,
'id': idx}
if params is None: if params is None:
obj['params'] = [] obj["params"] = []
else: else:
obj['params'] = params obj["params"] = params
return obj return obj
@staticmethod @staticmethod
def response_is_error(resp_obj): def response_is_error(resp_obj):
return 'error' in resp_obj and resp_obj['error'] is not None return "error" in resp_obj and resp_obj["error"] is not None
def get_block_hashes(settings, max_blocks_per_call=10000): def get_block_hashes(settings, max_blocks_per_call=10000):
rpc = BitcoinRPC(settings['host'], settings['port'], rpc = BitcoinRPC(
settings['rpcuser'], settings['rpcpassword']) settings["host"], settings["port"], settings["rpcuser"], settings["rpcpassword"]
)
height = settings['min_height']
while height < settings['max_height'] + 1: height = settings["min_height"]
num_blocks = min(settings['max_height'] + while height < settings["max_height"] + 1:
1 - height, max_blocks_per_call) num_blocks = min(settings["max_height"] + 1 - height, max_blocks_per_call)
batch = [] batch = []
for x in range(num_blocks): for x in range(num_blocks):
batch.append(rpc.build_request(x, 'getblockhash', [height + x])) batch.append(rpc.build_request(x, "getblockhash", [height + x]))
reply = rpc.execute(batch) reply = rpc.execute(batch)
if reply is None: if reply is None:
print('Cannot continue. Program will halt.') print("Cannot continue. Program will halt.")
return None return None
for x, resp_obj in enumerate(reply): for x, resp_obj in enumerate(reply):
if rpc.response_is_error(resp_obj): if rpc.response_is_error(resp_obj):
print('JSON-RPC: error at height', height + x, print(
': ', resp_obj['error'], file=sys.stderr) "JSON-RPC: error at height",
height + x,
": ",
resp_obj["error"],
file=sys.stderr,
)
sys.exit(1) sys.exit(1)
assert resp_obj['id'] == x # assume replies are in-sequence assert resp_obj["id"] == x # assume replies are in-sequence
if settings['rev_hash_bytes'] == 'true': if settings["rev_hash_bytes"] == "true":
resp_obj['result'] = hex_switchEndian(resp_obj['result']) resp_obj["result"] = hex_switchEndian(resp_obj["result"])
print(resp_obj['result']) print(resp_obj["result"])
height += num_blocks height += num_blocks
def get_rpc_cookie(): def get_rpc_cookie():
# Open the cookie file # Open the cookie file
with open(os.path.join(os.path.expanduser(settings['datadir']), '.cookie'), 'r', encoding="ascii") as f: with open(
os.path.join(os.path.expanduser(settings["datadir"]), ".cookie"),
"r",
encoding="ascii",
) as f:
combined = f.readline() combined = f.readline()
combined_split = combined.split(":") combined_split = combined.split(":")
settings['rpcuser'] = combined_split[0] settings["rpcuser"] = combined_split[0]
settings['rpcpassword'] = combined_split[1] settings["rpcpassword"] = combined_split[1]
if __name__ == '__main__': if __name__ == "__main__":
if len(sys.argv) != 2: if len(sys.argv) != 2:
print("Usage: linearize-hashes.py CONFIG-FILE") print("Usage: linearize-hashes.py CONFIG-FILE")
sys.exit(1) sys.exit(1)
f = open(sys.argv[1], encoding="utf8") f = open(sys.argv[1], encoding="utf8")
for line in f: for line in f:
# skip comment lines # skip comment lines
m = re.search(r'^\s*#', line) m = re.search(r"^\s*#", line)
if m: if m:
continue continue
# parse key=value lines # parse key=value lines
m = re.search(r'^(\w+)\s*=\s*(\S.*)$', line) m = re.search(r"^(\w+)\s*=\s*(\S.*)$", line)
if m is None: if m is None:
continue continue
settings[m.group(1)] = m.group(2) settings[m.group(1)] = m.group(2)
f.close() f.close()
if 'host' not in settings: if "host" not in settings:
settings['host'] = '127.0.0.1' settings["host"] = "127.0.0.1"
if 'port' not in settings: if "port" not in settings:
settings['port'] = 8332 settings["port"] = 8332
if 'min_height' not in settings: if "min_height" not in settings:
settings['min_height'] = 0 settings["min_height"] = 0
if 'max_height' not in settings: if "max_height" not in settings:
settings['max_height'] = 313000 settings["max_height"] = 313000
if 'rev_hash_bytes' not in settings: if "rev_hash_bytes" not in settings:
settings['rev_hash_bytes'] = 'false' settings["rev_hash_bytes"] = "false"
use_userpass = True use_userpass = True
use_datadir = False use_datadir = False
if 'rpcuser' not in settings or 'rpcpassword' not in settings: if "rpcuser" not in settings or "rpcpassword" not in settings:
use_userpass = False use_userpass = False
if 'datadir' in settings and not use_userpass: if "datadir" in settings and not use_userpass:
use_datadir = True use_datadir = True
if not use_userpass and not use_datadir: if not use_userpass and not use_datadir:
print("Missing datadir or username and/or password in cfg file", print(
file=sys.stderr) "Missing datadir or username and/or password in cfg file", file=sys.stderr
)
sys.exit(1) sys.exit(1)
settings['port'] = int(settings['port']) settings["port"] = int(settings["port"])
settings['min_height'] = int(settings['min_height']) settings["min_height"] = int(settings["min_height"])
settings['max_height'] = int(settings['max_height']) settings["max_height"] = int(settings["max_height"])
# Force hash byte format setting to be lowercase to make comparisons # Force hash byte format setting to be lowercase to make comparisons
# easier. # easier.
settings['rev_hash_bytes'] = settings['rev_hash_bytes'].lower() settings["rev_hash_bytes"] = settings["rev_hash_bytes"].lower()
# Get the rpc user and pass from the cookie if the datadir is set # Get the rpc user and pass from the cookie if the datadir is set
if use_datadir: if use_datadir:
get_rpc_cookie() get_rpc_cookie()
get_block_hashes(settings) get_block_hashes(settings)