diff options
Diffstat (limited to 'tools')
| -rw-r--r-- | tools/certtools.py | 48 | ||||
| -rwxr-xr-x | tools/submitcert.py | 80 |
2 files changed, 71 insertions, 57 deletions
diff --git a/tools/certtools.py b/tools/certtools.py index 13dad17..fe345cd 100644 --- a/tools/certtools.py +++ b/tools/certtools.py @@ -8,6 +8,20 @@ import sys import hashlib import ecdsa +publickeys = { + "https://ct.googleapis.com/pilot/": + "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEfahLEimAoz2t01p3uMziiLOl/fHTD" + "M0YDOhBRuiBARsV4UvxG2LdNgoIGLrtCzWE0J5APC2em4JlvR8EEEFMoA==", + + "https://127.0.0.1:8080/": + "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE4qWq6afhBUi0OdcWUYhyJLNXTkGqQ9" + "PMS5lqoCgkV2h1ZvpNjBH2u8UbgcOQwqDo66z6BWQJGolozZYmNHE2kQ==", + + "https://flimsy.ct.nordu.net/": + "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE4qWq6afhBUi0OdcWUYhyJLNXTkGqQ9" + "PMS5lqoCgkV2h1ZvpNjBH2u8UbgcOQwqDo66z6BWQJGolozZYmNHE2kQ==", +} + def get_cert_info(s): p = subprocess.Popen( ["openssl", "x509", "-noout", "-subject", "-issuer", "-inform", "der"], @@ -15,7 +29,8 @@ def get_cert_info(s): stderr=subprocess.PIPE) parsed = p.communicate(s) if parsed[1]: - print "error:", parsed[1] + print "ERROR:", parsed[1] + sys.exit(1) result = {} for line in parsed[0].split("\n"): (key, sep, value) = line.partition("=") @@ -34,7 +49,7 @@ def get_certs_from_file(certfile): cert = "" incert = True elif line == "-----END CERTIFICATE-----": - certs.append(cert) + certs.append(base64.decodestring(cert)) incert = False elif incert: cert += line @@ -49,7 +64,6 @@ def get_root_cert(issuer): for accepted_cert in accepted_certs: subject = get_cert_info(base64.decodestring(accepted_cert))["subject"] if subject == issuer: - print "found root cert" root_cert = base64.decodestring(accepted_cert) return root_cert @@ -62,12 +76,11 @@ def get_proof_by_hash(baseurl, hash, tree_size): try: params = urllib.urlencode({"hash":base64.b64encode(hash), "tree_size":tree_size}) - print params result = \ urllib2.urlopen(baseurl + "ct/v1/get-proof-by-hash?" + params).read() return json.loads(result) except urllib2.HTTPError, e: - print e.read() + print "ERROR:", e.read() sys.exit(1) def tls_array(data, length_len): @@ -90,7 +103,7 @@ def add_chain(baseurl, submission): return json.loads(urllib2.urlopen(baseurl + "ct/v1/add-chain", json.dumps(submission)).read()) except urllib2.HTTPError, e: - print e.read() + print "ERROR:", e.read() sys.exit(1) def get_entries(baseurl, start, end): @@ -99,7 +112,7 @@ def get_entries(baseurl, start, end): result = urllib2.urlopen(baseurl + "ct/v1/get-entries?" + params).read() return json.loads(result) except urllib2.HTTPError, e: - print e.read() + print "ERROR:", e.read() sys.exit(1) def decode_certificate_chain(packed_certchain): @@ -118,7 +131,8 @@ def decode_signature(signature): assert rest == "" return (hash_alg, signature_alg, unpacked_signature) -def check_signature(publickey, leafcert, sct): +def check_signature(baseurl, leafcert, sct): + publickey = base64.decodestring(publickeys[baseurl]) calculated_logid = hashlib.sha256(publickey).digest() received_logid = base64.decodestring(sct["id"]) assert calculated_logid == received_logid, \ @@ -146,3 +160,21 @@ def check_signature(publickey, leafcert, sct): vk = ecdsa.VerifyingKey.from_der(publickey) vk.verify(unpacked_signature, signed_struct, hashfunc=hashlib.sha256, sigdecode=ecdsa.util.sigdecode_der) + +def pack_mtl(timestamp, leafcert): + entry_type = struct.pack(">H", 0) + extensions = "" + + timestamped_entry = struct.pack(">Q", timestamp) + entry_type + \ + tls_array(leafcert, 3) + tls_array(extensions, 2) + version = struct.pack(">b", 0) + leaf_type = struct.pack(">b", 0) + merkle_tree_leaf = version + leaf_type + timestamped_entry + return merkle_tree_leaf + +def get_leaf_hash(merkle_tree_leaf): + leaf_hash = hashlib.sha256() + leaf_hash.update(struct.pack(">b", 0)) + leaf_hash.update(merkle_tree_leaf) + + return leaf_hash.digest() diff --git a/tools/submitcert.py b/tools/submitcert.py index 702ffb3..e8d8901 100755 --- a/tools/submitcert.py +++ b/tools/submitcert.py @@ -6,6 +6,7 @@ import base64 import sys import struct import hashlib +import itertools from certtools import * baseurl = sys.argv[1] @@ -13,62 +14,29 @@ certfile = sys.argv[2] lookup_in_log = True -publickeys = { - "https://ct.googleapis.com/pilot/": - "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAEfahLEimAoz2t01p3uMziiLOl/fHTD" - "M0YDOhBRuiBARsV4UvxG2LdNgoIGLrtCzWE0J5APC2em4JlvR8EEEFMoA==", - - "https://127.0.0.1:8080/": - "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE4qWq6afhBUi0OdcWUYhyJLNXTkGqQ9" - "PMS5lqoCgkV2h1ZvpNjBH2u8UbgcOQwqDo66z6BWQJGolozZYmNHE2kQ==", - - "https://flimsy.ct.nordu.net/": - "MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAE4qWq6afhBUi0OdcWUYhyJLNXTkGqQ9" - "PMS5lqoCgkV2h1ZvpNjBH2u8UbgcOQwqDo66z6BWQJGolozZYmNHE2kQ==", -} - - certs = get_certs_from_file(certfile) -result = add_chain(baseurl, {"chain":certs}) - -print result - -publickey = base64.decodestring(publickeys[baseurl]) - -check_signature(publickey, base64.decodestring(certs[0]), result) +result = add_chain(baseurl, {"chain":map(base64.b64encode, certs)}) -for cert in certs: - print get_cert_info(base64.decodestring(cert)) +try: + check_signature(baseurl, certs[0], result) +except AssertionError, e: + print "ERROR:", e + sys.exit(1) +except ecdsa.keys.BadSignatureError, e: + print "ERROR: bad signature" + sys.exit(1) +print "signature check succeeded" if lookup_in_log: - last_issuer = get_cert_info(base64.decodestring(certs[-1]))["issuer"] - last_subject = get_cert_info(base64.decodestring(certs[-1]))["subject"] - entry_type = struct.pack(">H", 0) + merkle_tree_leaf = pack_mtl(result["timestamp"], certs[0]) - extensions = "" - - timestamped_entry = struct.pack(">Q", result["timestamp"]) + entry_type + \ - tls_array(base64.decodestring(certs[0]), 3) + tls_array(extensions, 2) - version = struct.pack(">b", 0) - leaf_type = struct.pack(">b", 0) - merkle_tree_leaf = version + leaf_type + timestamped_entry - - print "merkle_tree_leaf:", base64.b64encode(merkle_tree_leaf) - - leaf_hash = hashlib.sha256() - leaf_hash.update(struct.pack(">b", 0)) - leaf_hash.update(merkle_tree_leaf) - - print base64.b64encode(leaf_hash.digest()) + leaf_hash = get_leaf_hash(merkle_tree_leaf) sth = get_sth(baseurl) - print sth - - proof = get_proof_by_hash(baseurl, leaf_hash.digest(), sth["tree_size"]) - print proof + proof = get_proof_by_hash(baseurl, leaf_hash, sth["tree_size"]) leaf_index = proof["leaf_index"] @@ -76,8 +44,6 @@ if lookup_in_log: fetched_entry = entries["entries"][0] - print fetched_entry - print "does the leaf_input of the fetched entry match what we calculated:", \ base64.decodestring(fetched_entry["leaf_input"]) == merkle_tree_leaf @@ -85,4 +51,20 @@ if lookup_in_log: certchain = decode_certificate_chain(base64.decodestring(extra_data)) - print [base64.b64encode(cert) for cert in certchain] + submittedcertchain = certs[1:] + + for (submittedcert, fetchedcert, i) in zip(submittedcertchain, + certchain, itertools.count(1)): + print "cert", i, "in chain is the same:", submittedcert == fetchedcert + + if len(certchain) == len(submittedcertchain) + 1: + last_issuer = get_cert_info(certs[-1])["issuer"] + root_subject = get_cert_info(certchain[-1])["subject"] + print "issuer of last cert in submitted chain and " \ + "subject of last cert in fetched chain is the same:", \ + last_issuer == root_subject + elif len(certchain) == len(submittedcertchain): + print "cert chains are the same length" + else: + print "ERROR: fetched cert chain has length", len(certchain), + print "and submitted chain has length", len(submittedcertchain) |