Merge remote-tracking branch 'refs/remotes/map/testtools4'

1 files changed, 184 insertions, 0 deletions

diff --git a/tools/testcase1.py b/tools/testcase1.py
new file mode 100755
index 0000000..f78faa3
--- /dev/null
+++ b/tools/testcase1.py
@@ -0,0 +1,184 @@
+#!/usr/bin/env python
+import urllib2
+import urllib
+import json
+import base64
+import sys
+import struct
+import hashlib
+import itertools
+from certtools import *
+
+baseurl = "https://127.0.0.1:8080/"
+certfiles = ["testcerts/cert1.txt", "testcerts/cert2.txt",
+             "testcerts/cert3.txt", "testcerts/cert4.txt",
+             "testcerts/cert5.txt"]
+
+cc1 = get_certs_from_file(certfiles[0])
+cc2 = get_certs_from_file(certfiles[1])
+cc3 = get_certs_from_file(certfiles[2])
+cc4 = get_certs_from_file(certfiles[3])
+cc5 = get_certs_from_file(certfiles[4])
+
+failures = 0
+indentation = ""
+
+def testgroup(name):
+    global indentation
+    print name + ":"
+    indentation = "    "
+
+def print_error(message, *args):
+    global failures, indentation
+    print indentation + "ERROR:", message % args
+    failures += 1
+
+def print_success(message, *args):
+    print indentation + message % args
+
+def assert_equal(actual, expected, name, quiet=False, nodata=False):
+    global failures
+    if actual != expected:
+        if nodata:
+            print_error("%s differs", name)
+        else:
+            print_error("%s expected %s got %s", name, expected, actual)
+    elif not quiet:
+        print_success("%s was correct", name)
+
+def print_and_check_tree_size(expected):
+    global failures
+    sth = get_sth(baseurl)
+    try:
+        check_sth_signature(baseurl, sth)
+    except AssertionError, e:
+        print_error("%s", e)
+    except ecdsa.keys.BadSignatureError, e:
+        print_error("bad STH signature")
+    tree_size = sth["tree_size"]
+    assert_equal(tree_size, expected, "tree size")
+
+def do_add_chain(chain):
+    global failures
+    try:
+        result = add_chain(baseurl, {"chain":map(base64.b64encode, chain)})
+    except ValueError, e:
+        print_error("%s", e)
+    try:
+        check_sct_signature(baseurl, chain[0], result)
+    except AssertionError, e:
+        print_error("%s", e)
+    except ecdsa.keys.BadSignatureError, e:
+        print_error("bad SCT signature")
+    print_success("signature check succeeded")
+    return result
+
+def get_and_validate_proof(timestamp, chain, leaf_index, nentries):
+    cert = chain[0]
+    merkle_tree_leaf = pack_mtl(timestamp, cert)
+    leaf_hash = get_leaf_hash(merkle_tree_leaf)
+    sth = get_sth(baseurl)
+    proof = get_proof_by_hash(baseurl, leaf_hash, sth["tree_size"])
+    assert_equal(proof["leaf_index"], leaf_index, "leaf_index")
+    assert_equal(len(proof["audit_path"]), nentries, "audit_path length")
+    get_and_check_entry(timestamp, chain, leaf_index)
+
+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)
+    fetched_entry = entries["entries"][0]
+    merkle_tree_leaf = pack_mtl(timestamp, chain[0])
+    leaf_input =  base64.decodestring(fetched_entry["leaf_input"])
+    assert_equal(leaf_input, merkle_tree_leaf, "entry", nodata=True)
+    extra_data = base64.decodestring(fetched_entry["extra_data"])
+    certchain = decode_certificate_chain(extra_data)
+
+    submittedcertchain = chain[1:]
+
+    for (submittedcert, fetchedcert, i) in zip(submittedcertchain,
+                                               certchain, itertools.count(1)):
+        assert_equal(fetchedcert, submittedcert, "cert %d in chain" % (i,))
+
+    if len(certchain) == len(submittedcertchain) + 1:
+        last_issuer = get_cert_info(certs[-1])["issuer"]
+        root_subject = get_cert_info(certchain[-1])["subject"]
+        if last_issuer == root_subject:
+            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:
+        print_error("cert chain length %d expected %d or %d",
+                    len(certchain),
+                    len(submittedcertchain),
+                    len(submittedcertchain))
+
+print_and_check_tree_size(0)
+
+testgroup("cert1")
+
+result1 = do_add_chain(cc1)
+
+print_and_check_tree_size(1)
+
+result2 = do_add_chain(cc1)
+
+assert_equal(result2["timestamp"], result1["timestamp"], "timestamp")
+
+print_and_check_tree_size(1)
+
+# TODO: add invalid cert and check that it generates an error
+#       and that treesize still is 1
+
+get_and_validate_proof(result1["timestamp"], cc1, 0, 0)
+
+testgroup("cert2")
+
+result3 = do_add_chain(cc2)
+
+print_and_check_tree_size(2)
+
+get_and_validate_proof(result1["timestamp"], cc1, 0, 1)
+get_and_validate_proof(result3["timestamp"], cc2, 1, 1)
+
+testgroup("cert3")
+
+result4 = do_add_chain(cc3)
+
+print_and_check_tree_size(3)
+
+get_and_validate_proof(result1["timestamp"], cc1, 0, 2)
+get_and_validate_proof(result3["timestamp"], cc2, 1, 2)
+get_and_validate_proof(result4["timestamp"], cc3, 2, 1)
+
+testgroup("cert4")
+
+result5 = do_add_chain(cc4)
+
+print_and_check_tree_size(4)
+
+get_and_validate_proof(result1["timestamp"], cc1, 0, 2)
+get_and_validate_proof(result3["timestamp"], cc2, 1, 2)
+get_and_validate_proof(result4["timestamp"], cc3, 2, 2)
+get_and_validate_proof(result5["timestamp"], cc4, 3, 2)
+
+testgroup("cert5")
+
+result6 = do_add_chain(cc5)
+
+print_and_check_tree_size(5)
+
+get_and_validate_proof(result1["timestamp"], cc1, 0, 3)
+get_and_validate_proof(result3["timestamp"], cc2, 1, 3)
+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)
+
+print "-------"
+if failures:
+    print failures, "failed tests" if failures != 1 else "failed test"
+    sys.exit(1)
+else:
+    print "all tests succeeded"