#!/usr/bin/env python # -*- coding: utf-8 -*- # Copyright (c) 2014, NORDUnet A/S. # See LICENSE for licensing information. 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) subprocess.call(["./merge.py"]) print_and_check_tree_size(1) result2 = do_add_chain(cc1) assert_equal(result2["timestamp"], result1["timestamp"], "timestamp") subprocess.call(["./merge.py"]) 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) subprocess.call(["./merge.py"]) 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) subprocess.call(["./merge.py"]) 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) subprocess.call(["./merge.py"]) 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) subprocess.call(["./merge.py"]) 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"