aboutsummaryrefslogtreecommitdiff
path: root/siglog-witness.py
diff options
context:
space:
mode:
Diffstat (limited to 'siglog-witness.py')
-rwxr-xr-xsiglog-witness.py150
1 files changed, 93 insertions, 57 deletions
diff --git a/siglog-witness.py b/siglog-witness.py
index f3475d2..06a2c9c 100755
--- a/siglog-witness.py
+++ b/siglog-witness.py
@@ -31,10 +31,11 @@ SIGKEY_FILE_DEFAULT = CONFIG_DIR_DEFAULT + 'signing_key'
CONFIG_FILE = CONFIG_DIR_DEFAULT + 'siglog-witness.conf'
ERR_TREEHEAD_SIGNATURE_INVALID = 2
-ERR_CONSISTENCYPROOF_FETCH = 4
-ERR_CONSISTENCYPROOF_INVALID = 5
-ERR_TREEHEAD_FETCH = 6
-ERR_LOGKEYFILE = 7
+ERR_TREEHEAD_READ = 3
+ERR_TREEHEAD_FETCH = 4
+ERR_CONSISTENCYPROOF_FETCH = 5
+ERR_CONSISTENCYPROOF_INVALID = 6
+ERR_LOGKEY = 7
ERR_LOGKEY_FORMAT = 8
ERR_SIGKEYFILE_TYPE = 9
ERR_SIGKEYFILE_MODE = 10
@@ -170,30 +171,42 @@ def make_base_dir_maybe():
except FileNotFoundError:
os.makedirs(dirname, mode=0o700)
-def read_tree_head():
- filename = os.path.expanduser(g_args.base_dir) + 'signed_tree_head'
+def read_tree_head(filename):
try:
with open(filename, mode='r') as f:
return TreeHead(f.read())
except FileNotFoundError:
return None
+def read_tree_head_and_verify(log_verification_key):
+ fn = os.path.expanduser(g_args.base_dir) + 'signed_tree_head'
+ tree_head = read_tree_head(fn)
+ if not tree_head:
+ return None, (ERR_TREEHEAD_READ,
+ "ERROR: unable to read file {}".format(fn))
+
+ if not tree_head.signature_valid(log_verification_key):
+ return None, (ERR_TREEHEAD_SIGNATURE_INVALID,
+ "ERROR: signature of stored tree head invalid")
+
+ return tree_head, None
+
def store_tree_head(tree_head):
dirname = os.path.expanduser(g_args.base_dir)
with open(dirname + 'signed_tree_head', mode='w+b') as f:
f.write(tree_head.text())
-def fetch_tree_head(log_verification_key):
+def fetch_tree_head_and_verify(log_verification_key):
req = requests.get(g_args.base_url + 'st/v0/get-tree-head-to-sign')
if req.status_code != 200:
- return None, (ERR_TREEHEAD_FETCH, "ERROR: unable to fetch new tree head: {}".format(req.status_code))
+ return None, (ERR_TREEHEAD_FETCH,
+ "ERROR: unable to fetch new tree head: {}".format(req.status_code))
tree_head = TreeHead(req.content.decode())
-
if not tree_head.signature_valid(log_verification_key):
- return None, (ERR_TREEHEAD_SIGNATURE_INVALID, "ERROR: signature of fetched tree head not valid")
+ return None, (ERR_TREEHEAD_SIGNATURE_INVALID,
+ "ERROR: signature of fetched tree head invalid")
- assert(tree_head is not None)
return tree_head, None
def fetch_consistency_proof(first, second):
@@ -201,9 +214,9 @@ def fetch_consistency_proof(first, second):
post_data += 'new_size={}\n'.format(second)
req = requests.post(g_args.base_url + 'st/v0/get-consistency-proof', post_data)
if req.status_code != 200:
- print("ERROR: st/v0/get-consistency-proof({}) => {}".format(post_data, req))
- return None
- return ConsistencyProof(req.content.decode())
+ return None, (ERR_CONSISTENCYPROOF_FETCH,
+ "ERROR: unable to fetch consistency proof: {}".format(req.status_code))
+ return ConsistencyProof(req.content.decode()), None
def numbits(n):
p = 0
@@ -253,9 +266,9 @@ def consistency_proof_valid(first, second, proof):
return sn == 0 and fr == first.root_hash() and sr == second.root_hash()
-def sign_and_send_sig(signing_key, sth):
+def sign_send_store_tree_head(signing_key, tree_head):
hash = sha256(signing_key.verify_key.encode())
- signature = signing_key.sign(sth.serialise()).signature
+ signature = signing_key.sign(tree_head.serialise()).signature
post_data = 'signature={}\n'.format(hexlify(signature).decode('ascii'))
post_data += 'key_hash={}\n'.format(hash.hexdigest())
@@ -266,14 +279,18 @@ def sign_and_send_sig(signing_key, sth):
"ERROR: Unable to post signature to log: {} => {}: {}". format(req.url,
req.status_code,
req.text))
+ # Store only when all else is done. Next invocation will treat a
+ # stored tree head as having been verified.
+ store_tree_head(tree_head)
def ensure_log_verification_key():
if not g_args.log_verification_key:
- return None, (ERR_LOGKEYFILE, "ERROR: missing log verification key")
+ return None, (ERR_LOGKEY, "ERROR: missing log verification key")
try:
log_verification_key = nacl.signing.VerifyKey(g_args.log_verification_key, encoder=nacl.encoding.HexEncoder)
except:
- return None, (ERR_LOGKEY_FORMAT, "ERROR: invalid log verification key: {}".format(g_args.log_verification_key))
+ return None, (ERR_LOGKEY_FORMAT,
+ "ERROR: invalid log verification key: {}".format(g_args.log_verification_key))
assert(log_verification_key is not None)
return log_verification_key, None
@@ -292,21 +309,28 @@ def ensure_sigkey(fn):
s = os.stat(fn, follow_symlinks=False)
if not S_ISREG(s.st_mode):
- return None, (ERR_SIGKEYFILE_TYPE, "ERROR: Signing key file {} must be a regular file".format(fn))
+ return None, (ERR_SIGKEYFILE_TYPE,
+ "ERROR: Signing key file {} must be a regular file".format(fn))
if S_IMODE(s.st_mode) & 0o077 != 0:
- return None, (ERR_SIGKEYFILE_MODE, "ERROR: Signing key file {} permissions too lax: {:04o}".format(fn, S_IMODE(s.st_mode)))
+ return None, (ERR_SIGKEYFILE_MODE,
+ "ERROR: Signing key file {} permissions too lax: {:04o}".format(fn, S_IMODE(s.st_mode)))
with open(fn, 'r') as f:
try:
signing_key = nacl.signing.SigningKey(f.readline().strip(), nacl.encoding.HexEncoder)
except:
- return None, (ERR_SIGKEY_FORMAT, "ERROR: Invalid signing key in {}".format(fn))
+ return None, (ERR_SIGKEY_FORMAT,
+ "ERROR: Invalid signing key in {}".format(fn))
assert(signing_key is not None)
return signing_key, None
+def user_confirm(prompt):
+ if input(prompt + ' y/n> ').lower()[0] == 'y':
+ return True
+ return False
+
def main(args):
- msg = None
global g_args
g_args = Parser()
parse_config(CONFIG_FILE)
@@ -326,43 +350,55 @@ def main(args):
signing_key, err = ensure_sigkey(g_args.sigkey_file)
if err: return err
- new, err = fetch_tree_head(log_verification_key)
+ new = None # FIXME rename new -> new_tree_head
+ cur, err = read_tree_head_and_verify(log_verification_key) # FIXME rename cur -> cur_tree_head
+ if err:
+ new, err2 = fetch_tree_head_and_verify(log_verification_key)
+ if err2: return err2
+
+ if not g_args.bootstrap_log:
+ return err
+
+ print("\nWARNING: We have only seen one single tree head from the\n"
+ "log {},\n"
+ "representing a tree of size {}. We are therefore unable to\n"
+ "verify that the tree it represents is really a superset of an\n"
+ "earlier version of the tree in this log.\n"
+ "\nWe are effectively signing this tree head blindly.\n".format(g_args.base_url,
+ new.tree_size()))
+ if user_confirm("Really sign head for tree of size {} and upload "
+ "the signature?".format(new.tree_size())):
+ err3 = sign_send_store_tree_head(signing_key, new)
+ if err3: return err3
+
+ return 0, None
+
+ new, err = fetch_tree_head_and_verify(log_verification_key)
if err: return err
- # FIXME: if we're bootstrapping, ignore potential tree head on disk
- cur = read_tree_head()
- if not cur:
- print("INFO: No current tree head found in {}".format(g_args.base_dir))
- else:
- if not cur.signature_valid(log_verification_key):
- return ERR_TREEHEAD_SIGNATURE_INVALID, "ERROR: signature of current tree head invalid"
- if new.tree_size() <= cur.tree_size():
- msg = "INFO: Fetched tree already verified, size {}".format(cur.tree_size())
- else:
- proof = fetch_consistency_proof(cur.tree_size(), new.tree_size())
- if not proof:
- return ERR_CONSISTENCYPROOF_FETCH, "ERROR: unable to fetch consistency proof"
- if consistency_proof_valid(cur, new, proof):
- consistency_verified = True
- else:
- errmsg = "ERROR: failing consistency proof check for {}->{}".format(cur.tree_size(), new.tree_size())
- errmsg += "DEBUG: {}:{}->{}:{}\n {}".format(cur.tree_size(),
- cur.root_hash(),
- new.tree_size(),
- new.root_hash(),
- proof.path())
- return ERR_CONSISTENCYPROOF_INVALID, errmsg
-
- if g_args.bootstrap_log:
- # TODO maybe require user confirmation
- ignore_consistency = True
-
- store_tree_head(new)
- if consistency_verified or ignore_consistency:
- err = sign_and_send_sig(signing_key, new)
- if err: return err
-
- return 0, msg
+ if not cur.signature_valid(log_verification_key):
+ return ERR_TREEHEAD_SIGNATURE_INVALID, "ERROR: signature of current tree head invalid"
+
+ if new.tree_size() <= cur.tree_size():
+ return 0, "INFO: Fetched head of tree of size {} already seen".format(cur.tree_size())
+
+ proof, err = fetch_consistency_proof(cur.tree_size(), new.tree_size())
+ if err: return err
+
+ if not consistency_proof_valid(cur, new, proof):
+ errmsg = "ERROR: failing consistency proof check for {}->{}\n".format(cur.tree_size(),
+ new.tree_size())
+ errmsg += "DEBUG: {}:{}->{}:{}\n {}".format(cur.tree_size(),
+ cur.root_hash(),
+ new.tree_size(),
+ new.root_hash(),
+ proof.path())
+ return ERR_CONSISTENCYPROOF_INVALID, errmsg
+
+ err = sign_send_store_tree_head(signing_key, new)
+ if err: return err
+
+ return 0, None
if __name__ == '__main__':
status = main(sys.argv)