From e54e051600fc7f46fb7eafce74d311a61d5bd823 Mon Sep 17 00:00:00 2001
From: Magnus Ahltorp <map@kth.se>
Date: Fri, 26 Sep 2014 14:50:10 +0200
Subject: Simplify submitcert
---
tools/certtools.py | 48 ++++++++++++++++++++++++++------
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)
--
cgit v1.1