Merge branch 'mergesecondary'

1 files changed, 29 insertions, 53 deletions

diff --git a/tools/merge.py b/tools/merge.py
index 76ffede..b426039 100755
--- a/tools/merge.py
+++ b/tools/merge.py
@@ -22,6 +22,8 @@ from certtools import build_merkle_tree, create_sth_signature, \
     check_sth_signature, get_eckey_from_file, timing_point, http_request, \
     get_public_key_from_file, get_leaf_hash, decode_certificate_chain, \
     create_ssl_context
+from mergetools import parselogrow, get_logorder, read_chain, unpack_entry, \
+    verify_entry
 
 parser = argparse.ArgumentParser(description="")
 parser.add_argument('--config', help="System configuration", required=True)
@@ -36,6 +38,7 @@ localconfig = yaml.load(open(args.localconfig))
 ctbaseurl = config["baseurl"]
 frontendnodes = config["frontendnodes"]
 storagenodes = config["storagenodes"]
+secondaries = localconfig.get("secondary", [])
 paths = localconfig["paths"]
 mergedb = paths["mergedb"]
 
@@ -51,13 +54,6 @@ logpublickey = get_public_key_from_file(paths["logpublickey"])
 
 hashed_dir = True
 
-def parselogrow(row):
-    return base64.b16decode(row)
-
-def get_logorder():
-    f = open(logorderfile, "r")
-    return [parselogrow(row.rstrip()) for row in f]
-
 def write_chain(key, value):
     filename = base64.b16encode(key)
     if hashed_dir:
@@ -72,17 +68,6 @@ def write_chain(key, value):
     f.write(value)
     f.close()
 
-def read_chain(key):
-    filename = base64.b16encode(key)
-    path = chainsdir + "/" + filename[0:2] + "/" + filename[2:4] + "/" + filename[4:6]
-    try:
-        f = open(path + "/" + filename, "r")
-    except IOError, e:
-        f = open(chainsdir + "/" + filename, "r")
-    value = f.read()
-    f.close()
-    return value
-
 def add_to_logorder(key):
     f = open(logorderfile, "a")
     f.write(base64.b16encode(key) + "\n")
@@ -194,7 +179,7 @@ def chunks(l, n):
 
 timing = timing_point()
 
-logorder = get_logorder()
+logorder = get_logorder(logorderfile)
 
 timing_point(timing, "get logorder")
 
@@ -210,41 +195,8 @@ for storagenode in storagenodes:
     new_entries.update(new_entries_per_node[storagenode["name"]])
     entries_to_fetch[storagenode["name"]] = []
 
-def unpack_entry(entry):
-    pieces = []
-    while len(entry):
-        (length,) = struct.unpack(">I", entry[0:4])
-        data = entry[4:4+length]
-        entry = entry[4+length:]
-        pieces.append(data)
-    return pieces
-
 import subprocess
 
-def verify_entry(verifycert, entry, hash):
-    unpacked = unpack_entry(entry)
-    mtl = unpacked[0]
-    assert hash == get_leaf_hash(mtl)
-    s = struct.pack(">I", len(entry)) + entry
-    try:
-        verifycert.stdin.write(s)
-    except IOError, e:
-        sys.stderr.write("merge: unable to write to verifycert process: ")
-        while 1:
-            line = verifycert.stdout.readline()
-            if line:
-                sys.stderr.write(line)
-            else:
-                sys.exit(1)
-    result_length_packed = verifycert.stdout.read(4)
-    (result_length,) = struct.unpack(">I", result_length_packed)
-    result = verifycert.stdout.read(result_length)
-    assert len(result) == result_length
-    (error_code,) = struct.unpack("B", result[0:1])
-    if error_code != 0:
-        print >>sys.stderr, result[1:]
-        sys.exit(1)
-
 timing_point(timing, "get new entries")
 
 new_entries -= certsinlog
@@ -286,6 +238,30 @@ tree_size = len(logorder)
 root_hash = tree[-1][0]
 timestamp = int(time.time() * 1000)
 
+for secondary in secondaries:
+    remotehost = secondary["host"]
+    remotedir = remotehost + ":" + secondary["mergedir"]
+    localdir = mergedb
+    if localdir[:-1] != '/':
+        localdir = localdir + "/"
+
+    print >>sys.stderr, "copying database to secondary:", remotehost
+    rsyncstatus = subprocess.call(["rsync", "-r", "--append", "--rsh=ssh", localdir, remotedir])
+    if rsyncstatus:
+        print >>sys.stderr, "rsync failed:", rsyncstatus
+        sys.exit(1)
+
+    print >>sys.stderr, "verifying database at secondary:", remotehost
+    verifysecondary = subprocess.Popen(["ssh", remotehost, secondary["verifycommand"], secondary["mergedir"]],
+                                    stdout=subprocess.PIPE)
+
+    (verifysecondaryresult, _) = verifysecondary.communicate()
+
+    if root_hash != base64.b16decode(verifysecondaryresult.strip()):
+        print >>sys.stderr, "secondary root hash was", verifysecondaryresult.strip()
+        print >>sys.stderr, "  expected", base64.b16encode(root_hash)
+        sys.exit(1)
+
 tree_head_signature = None
 for signingnode in signingnodes:
     try:
@@ -343,7 +319,7 @@ for frontendnode in frontendnodes:
     print "missing entries:", len(missingentries)
     for missingentry in missingentries:
         hash = base64.b64decode(missingentry)
-        sendentryresult = sendentry(nodename, nodeaddress, read_chain(hash), hash)
+        sendentryresult = sendentry(nodename, nodeaddress, read_chain(chainsdir, hash), hash)
         if sendentryresult["result"] != "ok":
             print "send sth:", sendentryresult
             sys.exit(1)