Add consistency proof checking to testcase1

Fix consistency proof checking when first size is power of 2

3 files changed, 28 insertions, 3 deletions

diff --git a/tools/certtools.py b/tools/certtools.py
index fdff0e1..2fb1492 100644
--- a/tools/certtools.py
+++ b/tools/certtools.py
@@ -549,8 +549,11 @@ def nodes_for_index(pos, treesize):
         level += 1
     return nodes
 
-def verify_consistency_proof(consistency_proof, first, second):
+def verify_consistency_proof(consistency_proof, first, second, oldhash_input):
+    if 2 ** bits(first - 1) == first:
+        consistency_proof = [oldhash_input] + consistency_proof
     chain = zip(nodes_for_subtree(first, second), consistency_proof)
+    assert len(nodes_for_subtree(first, second)) == len(consistency_proof)
     (_, hash) = reduce(lambda e1, e2: combine_two_hashes(e1, e2, second), chain)
     (_, oldhash) = reduce(lambda e1, e2: combine_two_hashes(e1, e2, first), chain)
     return (oldhash, hash)
diff --git a/tools/fetchallcerts.py b/tools/fetchallcerts.py
index 39ffd64..398c563 100755
--- a/tools/fetchallcerts.py
+++ b/tools/fetchallcerts.py
@@ -170,7 +170,7 @@ print "calculated root hash", base64.b16encode(calculated_root_hash)
 
 if oldsth and oldsth["tree_size"] > 0 and oldsth["tree_size"] != tree_size:
     consistency_proof = [base64.decodestring(entry) for entry in get_consistency_proof(args.baseurl, oldsth["tree_size"], tree_size)]
-    (old_treehead, new_treehead) = verify_consistency_proof(consistency_proof, oldsth["tree_size"], tree_size)
+    (old_treehead, new_treehead) = verify_consistency_proof(consistency_proof, oldsth["tree_size"], tree_size, base64.b64decode(oldsth["sha256_root_hash"]))
     assert old_treehead == base64.b64decode(oldsth["sha256_root_hash"])
     assert new_treehead == base64.b64decode(sth["sha256_root_hash"])
 
diff --git a/tools/testcase1.py b/tools/testcase1.py
index ce322f1..a41a783 100755
--- a/tools/testcase1.py
+++ b/tools/testcase1.py
@@ -95,6 +95,15 @@ def get_and_validate_proof(timestamp, chain, leaf_index, nentries):
     assert_equal(root_hash, calc_root_hash, "verified root hash", nodata=True)
     get_and_check_entry(timestamp, chain, leaf_index)
 
+def get_and_validate_consistency_proof(sth1, sth2, size1, size2):
+    consistency_proof = [base64.decodestring(entry) for entry in get_consistency_proof(baseurl, size1, size2)]
+    (old_treehead, new_treehead) = verify_consistency_proof(consistency_proof, size1, size2, sth1)
+    #print repr(sth1), repr(old_treehead)
+    #print repr(sth2), repr(new_treehead)
+    assert_equal(old_treehead, sth1, "sth1", nodata=True)
+    assert_equal(new_treehead, sth2, "sth2", nodata=True)
+
+
 def get_and_check_entry(timestamp, chain, leaf_index):
     entries = get_entries(baseurl, leaf_index, leaf_index)
     assert_equal(len(entries), 1, "get_entries", quiet=True)
@@ -118,7 +127,6 @@ def get_and_check_entry(timestamp, chain, leaf_index):
             print_success("fetched chain has an appended root cert")
         else:
             print_error("fetched chain has an extra entry")
-            failures += 1
     elif len(certchain) == len(submittedcertchain):
         print_success("cert chains are the same length")
     else:
@@ -142,7 +150,10 @@ result1 = do_add_chain(cc1)
 mergeresult = merge()
 assert_equal(mergeresult, 0, "merge", quiet=True)
 
+size_sth = {}
+
 print_and_check_tree_size(1)
+size_sth[1] = base64.b64decode(get_sth(baseurl)["sha256_root_hash"])
 
 result2 = do_add_chain(cc1)
 
@@ -152,6 +163,9 @@ mergeresult = merge()
 assert_equal(mergeresult, 0, "merge", quiet=True)
 
 print_and_check_tree_size(1)
+size1_v2_sth = base64.b64decode(get_sth(baseurl)["sha256_root_hash"])
+
+assert_equal(size_sth[1], size1_v2_sth, "sth", nodata=True)
 
 # TODO: add invalid cert and check that it generates an error
 #       and that treesize still is 1
@@ -166,6 +180,7 @@ mergeresult = merge()
 assert_equal(mergeresult, 0, "merge", quiet=True)
 
 print_and_check_tree_size(2)
+size_sth[2] = base64.b64decode(get_sth(baseurl)["sha256_root_hash"])
 
 get_and_validate_proof(result1["timestamp"], cc1, 0, 1)
 get_and_validate_proof(result3["timestamp"], cc2, 1, 1)
@@ -178,6 +193,7 @@ mergeresult = merge()
 assert_equal(mergeresult, 0, "merge", quiet=True)
 
 print_and_check_tree_size(3)
+size_sth[3] = base64.b64decode(get_sth(baseurl)["sha256_root_hash"])
 
 get_and_validate_proof(result1["timestamp"], cc1, 0, 2)
 get_and_validate_proof(result3["timestamp"], cc2, 1, 2)
@@ -191,6 +207,7 @@ mergeresult = merge()
 assert_equal(mergeresult, 0, "merge", quiet=True)
 
 print_and_check_tree_size(4)
+size_sth[4] = base64.b64decode(get_sth(baseurl)["sha256_root_hash"])
 
 get_and_validate_proof(result1["timestamp"], cc1, 0, 2)
 get_and_validate_proof(result3["timestamp"], cc2, 1, 2)
@@ -205,6 +222,7 @@ mergeresult = merge()
 assert_equal(mergeresult, 0, "merge", quiet=True)
 
 print_and_check_tree_size(5)
+size_sth[5] = base64.b64decode(get_sth(baseurl)["sha256_root_hash"])
 
 get_and_validate_proof(result1["timestamp"], cc1, 0, 3)
 get_and_validate_proof(result3["timestamp"], cc2, 1, 3)
@@ -212,6 +230,10 @@ get_and_validate_proof(result4["timestamp"], cc3, 2, 3)
 get_and_validate_proof(result5["timestamp"], cc4, 3, 3)
 get_and_validate_proof(result6["timestamp"], cc5, 4, 1)
 
+for first_size in range(1, 5):
+    for second_size in range(first_size + 1, 6):
+        get_and_validate_consistency_proof(size_sth[first_size], size_sth[second_size], first_size, second_size)
+
 print "-------"
 if failures:
     print failures, "failed tests" if failures != 1 else "failed test"