diff options
Diffstat (limited to 'tools')
-rw-r--r-- | tools/certtools.py | 13 | ||||
-rw-r--r-- | tools/precerttools.py | 2 | ||||
-rwxr-xr-x | tools/testcase1.py | 127 |
3 files changed, 44 insertions, 98 deletions
diff --git a/tools/certtools.py b/tools/certtools.py index 9296401..ec4177c 100644 --- a/tools/certtools.py +++ b/tools/certtools.py @@ -137,7 +137,7 @@ def pyopenssl_https_get(url): return response_lines[len(response_lines) - 1] def get_sth(baseurl): - result = urlopen(baseurl + "open/gaol/v1/get-sth").read() + result = urlopen(baseurl + "dt/v1/get-sth").read() return json.loads(result) def get_proof_by_hash(baseurl, hash, tree_size): @@ -145,7 +145,7 @@ def get_proof_by_hash(baseurl, hash, tree_size): params = urllib.urlencode({"hash":base64.b64encode(hash), "tree_size":tree_size}) result = \ - urlopen(baseurl + "open/gaol/v1/get-proof-by-hash?" + params).read() + urlopen(baseurl + "dt/v1/get-proof-by-hash?" + params).read() return json.loads(result) except urllib2.HTTPError, e: print "ERROR:", e.read() @@ -156,7 +156,7 @@ def get_consistency_proof(baseurl, tree_size1, tree_size2): params = urllib.urlencode({"first":tree_size1, "second":tree_size2}) result = \ - urlopen(baseurl + "open/gaol/v1/get-sth-consistency?" + params).read() + urlopen(baseurl + "dt/v1/get-sth-consistency?" + params).read() return json.loads(result)["consistency"] except urllib2.HTTPError, e: print "ERROR:", e.read() @@ -179,7 +179,8 @@ def unpack_tls_array(packed_data, length_len): def add_chain(baseurl, submission): try: - result = urlopen(baseurl + "open/gaol/v1/add-blob", json.dumps(submission)).read() + result = urlopen(baseurl + "dt/v1/add-rr-chain", + json.dumps(submission)).read() return json.loads(result) except urllib2.HTTPError, e: print "ERROR", e.code,":", e.read() @@ -196,7 +197,7 @@ def add_chain(baseurl, submission): def add_prechain(baseurl, submission): try: - result = urlopen(baseurl + "open/gaol/v1/add-pre-chain", + result = urlopen(baseurl + "dt/v1/add-pre-chain", json.dumps(submission)).read() return json.loads(result) except urllib2.HTTPError, e: @@ -215,7 +216,7 @@ def add_prechain(baseurl, submission): def get_entries(baseurl, start, end): params = urllib.urlencode({"start":start, "end":end}) try: - result = urlopen(baseurl + "open/gaol/v1/get-entries?" + params).read() + result = urlopen(baseurl + "dt/v1/get-entries?" + params).read() return json.loads(result) except urllib2.HTTPError, e: print "ERROR:", e.read() diff --git a/tools/precerttools.py b/tools/precerttools.py index 13ac572..9687e28 100644 --- a/tools/precerttools.py +++ b/tools/precerttools.py @@ -3,7 +3,7 @@ import sys import hashlib -import rfc2459 +import rfc2459 # debian package python-pyasn1-modules from pyasn1.type import univ, tag from pyasn1.codec.der import encoder, decoder diff --git a/tools/testcase1.py b/tools/testcase1.py index 1a294d9..ec85d85 100755 --- a/tools/testcase1.py +++ b/tools/testcase1.py @@ -13,23 +13,19 @@ import struct import hashlib import itertools from certtools import * +from dnstools import c14n_dsrr, unpack_rrset baseurls = [sys.argv[1]] logpublickeyfile = sys.argv[2] cacertfile = sys.argv[3] -certfiles = ["../tools/testcerts/cert1.txt", "../tools/testcerts/cert2.txt", - "../tools/testcerts/cert3.txt", "../tools/testcerts/cert4.txt", - "../tools/testcerts/cert5.txt"] +RRfiles = ["../test/testdata/dnssec/testrrsets/req-basic"] -def get_blob_from_file(filename): - return [open(filename, 'r').read()] +def get_rrset_from_file(filename): + return open(filename, 'r').read() -cc1 = get_blob_from_file(certfiles[0]) -cc2 = get_blob_from_file(certfiles[1]) -cc3 = get_blob_from_file(certfiles[2]) -cc4 = get_blob_from_file(certfiles[3]) -cc5 = get_blob_from_file(certfiles[4]) +# TODO: Add more tests, like 4 more would be good. +rr1 = get_rrset_from_file(RRfiles[0]) create_ssl_context(cafile=cacertfile) @@ -76,15 +72,18 @@ def print_and_check_tree_size(expected, baseurl): tree_size = sth["tree_size"] assert_equal(tree_size, expected, "tree size", quiet=True) -def do_add_chain(chain, baseurl): +def do_add_chain(ignore1, ignore2): assert 0, "use do_add_rr() instead" # FIXME: remove + +def do_add_rr(rrset, baseurl): global failures - blob = ''.join(chain) try: - result = add_chain(baseurl, {"blob":base64.b64encode(blob)}) + result = add_chain(baseurl, {"chain":base64.b64encode(rrset)}) except ValueError, e: print_error("%s", e) try: - signed_entry = pack_cert(blob) + print "result:", result + signed_entry = pack_cert(c14n_dsrr(rrset)) + print "signed_entry:", repr(signed_entry) check_sct_signature(baseurl, signed_entry, result, publickey=logpublickey) print_success("signature check succeeded") except AssertionError, e: @@ -95,8 +94,8 @@ def do_add_chain(chain, baseurl): return result def get_and_validate_proof(timestamp, chain, leaf_index, nentries, baseurl): - blob = ''.join(chain) - merkle_tree_leaf = pack_mtl(timestamp, blob) + dsrr = c14n_dsrr(chain) + merkle_tree_leaf = pack_mtl(timestamp, dsrr) leaf_hash = get_leaf_hash(merkle_tree_leaf) sth = get_sth(baseurl) proof = get_proof_by_hash(baseurl, leaf_hash, sth["tree_size"]) @@ -109,7 +108,7 @@ def get_and_validate_proof(timestamp, chain, leaf_index, nentries, baseurl): root_hash = base64.b64decode(sth["sha256_root_hash"]) assert_equal(root_hash, calc_root_hash, "verified root hash", nodata=True, quiet=True) - get_and_check_entry(timestamp, blob, leaf_index, baseurl) + get_and_check_entry(timestamp, chain, leaf_index, baseurl) def get_and_validate_consistency_proof(sth1, sth2, size1, size2, baseurl): consistency_proof = [base64.decodestring(entry) for entry in get_consistency_proof(baseurl, size1, size2)] @@ -121,15 +120,17 @@ def get_and_validate_consistency_proof(sth1, sth2, size1, size2, baseurl): def get_and_check_entry(timestamp, chain, leaf_index, baseurl): - blob = ''.join(chain) 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, blob) + merkle_tree_leaf = pack_mtl(timestamp, c14n_dsrr(chain)) leaf_input = base64.decodestring(fetched_entry["leaf_input"]) - extra_data = base64.decodestring(fetched_entry["extra_data"]) assert_equal(leaf_input, merkle_tree_leaf, "entry", nodata=True, quiet=True) - assert_equal(extra_data, '\x00\x00\x00', "extra_data", quiet=True) + extra_data = base64.decodestring(fetched_entry["extra_data"]) + chain_fetched = unpack_rrset(decode_certificate_chain(extra_data)[0]) + chain_submitted = unpack_rrset(chain)[1:] + # FIXME: Might not have submited trust anchors. + assert_equal(chain_fetched, chain_submitted, "chain", quiet=True) def merge(): return subprocess.call(["../tools/merge", "--config", "../test/catlfish-test.cfg", @@ -141,9 +142,9 @@ assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) for baseurl in baseurls: print_and_check_tree_size(0, baseurl) -testgroup("cert1") +testgroup("rr1") -result1 = do_add_chain(cc1, baseurls[0]) +result1 = do_add_rr(rr1, baseurls[0]) mergeresult = merge() assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) @@ -154,7 +155,7 @@ for baseurl in baseurls: print_and_check_tree_size(1, baseurl) size_sth[1] = base64.b64decode(get_sth(baseurls[0])["sha256_root_hash"]) -result2 = do_add_chain(cc1, baseurls[0]) +result2 = do_add_rr(rr1, baseurls[0]) assert_equal(result2["timestamp"], result1["timestamp"], "timestamp") @@ -167,79 +168,23 @@ size1_v2_sth = base64.b64decode(get_sth(baseurls[0])["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 - -get_and_validate_proof(result1["timestamp"], cc1, 0, 0, baseurls[0]) - -testgroup("cert2") - -result3 = do_add_chain(cc2, baseurls[0]) - -mergeresult = merge() -assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) - -for baseurl in baseurls: - print_and_check_tree_size(2, baseurl) -size_sth[2] = base64.b64decode(get_sth(baseurls[0])["sha256_root_hash"]) - -get_and_validate_proof(result1["timestamp"], cc1, 0, 1, baseurls[0]) -get_and_validate_proof(result3["timestamp"], cc2, 1, 1, baseurls[0]) - -testgroup("cert3") - -result4 = do_add_chain(cc3, baseurls[0]) - -mergeresult = merge() -assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) - -for baseurl in baseurls: - print_and_check_tree_size(3, baseurl) -size_sth[3] = base64.b64decode(get_sth(baseurls[0])["sha256_root_hash"]) - -get_and_validate_proof(result1["timestamp"], cc1, 0, 2, baseurls[0]) -get_and_validate_proof(result3["timestamp"], cc2, 1, 2, baseurls[0]) -get_and_validate_proof(result4["timestamp"], cc3, 2, 1, baseurls[0]) - -testgroup("cert4") - -result5 = do_add_chain(cc4, baseurls[0]) +# TODO: add an invalid chain and check that it generates an error and +# that treesize still is 1 -mergeresult = merge() -assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) - -for baseurl in baseurls: - print_and_check_tree_size(4, baseurl) -size_sth[4] = base64.b64decode(get_sth(baseurls[0])["sha256_root_hash"]) - -get_and_validate_proof(result1["timestamp"], cc1, 0, 2, baseurls[0]) -get_and_validate_proof(result3["timestamp"], cc2, 1, 2, baseurls[0]) -get_and_validate_proof(result4["timestamp"], cc3, 2, 2, baseurls[0]) -get_and_validate_proof(result5["timestamp"], cc4, 3, 2, baseurls[0]) - -testgroup("cert5") - -result6 = do_add_chain(cc5, baseurls[0]) - -mergeresult = merge() -assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) - -for baseurl in baseurls: - print_and_check_tree_size(5, baseurl) -size_sth[5] = base64.b64decode(get_sth(baseurls[0])["sha256_root_hash"]) +get_and_validate_proof(result1["timestamp"], rr1, 0, 0, baseurls[0]) -get_and_validate_proof(result1["timestamp"], cc1, 0, 3, baseurls[0]) -get_and_validate_proof(result3["timestamp"], cc2, 1, 3, baseurls[0]) -get_and_validate_proof(result4["timestamp"], cc3, 2, 3, baseurls[0]) -get_and_validate_proof(result5["timestamp"], cc4, 3, 3, baseurls[0]) -get_and_validate_proof(result6["timestamp"], cc5, 4, 1, baseurls[0]) +testgroup("proofs") mergeresult = merge() assert_equal(mergeresult, 0, "merge", quiet=True, fatal=True) -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, baseurls[0]) +for first_size in range(1, 1): + for second_size in range(first_size + 1, 2): + get_and_validate_consistency_proof(size_sth[first_size], + size_sth[second_size], + first_size, + second_size, + baseurls[0]) print "-------" if failures: |