path: root/siglog-witness.py

#! /usr/bin/env python3

# Sign the most recently published tree head from a given ST log,
# after verifying a consistency proof from an already verified tree
# head to this new tree head.

# A verified tree head is expected to be found in the file
# ~/.config/siglog-witness/signed_tree_head . It's updated once a
# newer tree head has been verified successfully.

# If the config file ~/.config/siglog-witness/siglog-witness.conf
# exists and is readable, options are read from it. Options read from
# the config file can be overridden on the command line.

import sys
import os
from stat import *
import argparse
import requests
import struct
from binascii import hexlify, unhexlify
import nacl.encoding
import nacl.signing
from hashlib import sha256

# TODO maybe stop mixing dashes and underscores in directory names and filenames

BASE_URL_DEFAULT = 'http://tlog-poc.system-transparency.org:6965/'
CONFIG_DIR_DEFAULT = os.path.expanduser('~/.config/siglog-witness/')
SIGKEY_FILE_DEFAULT = CONFIG_DIR_DEFAULT + 'signing_key'
CONFIG_FILE = CONFIG_DIR_DEFAULT + 'siglog-witness.conf'

ERR_TREEHEAD_SIGNATURE_INVALID = 2
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
ERR_SIGKEY_FORMAT              = 11
ERR_NYI                        = 12
ERR_COSIG_POST                 = 13

class Parser:
    def __init__(self):
        p = argparse.ArgumentParser(
            description='Sign the most recently published tree head from a given siglog, after verifying it against an older tree.')

        p.add_argument('--bootstrap-log',
                       action='store_true',
                       help="Sign and save fetched tree head without verifying a consistency proof against a previous tree head. NOTE: User intervention required.")

        p.add_argument('-d', '--base-dir',
                       default=CONFIG_DIR_DEFAULT,
                       help="Configuration directory ({})".format(CONFIG_DIR_DEFAULT))

        p.add_argument('-l', '--log-verification-key',
                       help="Log verification key")

        p.add_argument('--save-config',
                       action='store_true',
                       help="Save command line options to the configuration file")

        p.add_argument('-s', '--sigkey-file',
                       default=SIGKEY_FILE_DEFAULT,
                       help="Signing key file ({})".format(SIGKEY_FILE_DEFAULT))

        p.add_argument('-u', '--base-url',
                       default=BASE_URL_DEFAULT,
                       help="Log base URL ({})".format(BASE_URL_DEFAULT))

        self.parser = p

def parse_config(filename):
    try:
        lines = []
        with open(filename, 'r') as f:
            line = f.readline()
            while line:
                lines.append(line.strip())
                line = f.readline()
            g_args.parser.parse_args(lines, namespace=g_args)
    except FileNotFoundError:
        pass

def parse_args(argv):
    g_args.parser.parse_args(namespace=g_args)

def parse_keyval(text):
    dictx = {}
    for line in text.split():
        (key, val) = line.split('=')
        if not key in dictx:
            dictx[key] = val
        else:
            if type(dictx[key]) is list:
                dictx[key] += [val]
            else:
                dictx[key] = [dictx[key], val]
    return dictx

class TreeHead:
    def __init__(self, sth_data):
        self._text = parse_keyval(sth_data)
        assert(len(self._text) == 5)
        assert('timestamp' in self._text)
        assert('tree_size' in self._text)
        assert('root_hash' in self._text)
        assert('signature' in self._text)
        assert('key_hash' in self._text)

    def text(self):
        text = 'timestamp={}\n'.format(self._text['timestamp'])
        text += 'tree_size={}\n'.format(self._text['tree_size'])
        text += 'root_hash={}\n'.format(self._text['root_hash'])
        text += 'signature={}\n'.format(self._text['signature'])
        text += 'key_hash={}\n'.format(self._text['key_hash'])
        return text.encode('ascii')

    def serialise(self):
        data = struct.pack('!QQ', self.timestamp(), self.tree_size())
        data += unhexlify(self._text['root_hash'])
        assert(len(data) == 48)
        return data

    def signature_valid(self, pubkey):
        # Guard against tree head with >1 signature -- don't try to
        # validate a cosigned tree head.
        assert(type(self._text['signature']) is str)
        sig = unhexlify(self._text['signature'])
        assert(len(sig) == 64)
        data = self.serialise()
        try:
            verified_data = pubkey.verify(sig + data)
        except nacl.exceptions.BadSignatureError:
            return False
        assert(verified_data == data)
        return True

    def timestamp(self):
        return int(self._text['timestamp'])
    def tree_size(self):
        return int(self._text['tree_size'])
    def root_hash(self):
        return unhexlify(self._text['root_hash'])

class ConsistencyProof():
    def __init__(self, consistency_proof_data):
        self._text = parse_keyval(consistency_proof_data)
        assert(len(self._text) == 3)
        assert('old_size' in self._text)
        assert('new_size' in self._text)
        assert('consistency_path' in self._text)

    def old_size(self):
        return int(self._text['old_size'])
    def new_size(self):
        return int(self._text['new_size'])
    def path(self):
        if type(self._text['consistency_path']) is list:
            return [unhexlify(e) for e in self._text['consistency_path']]
        else:
            return [unhexlify(self._text['consistency_path'])]

def make_base_dir_maybe():
    dirname = os.path.expanduser(g_args.base_dir)
    try:
        os.stat(dirname)
    except FileNotFoundError:
        os.makedirs(dirname, mode=0o700)

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_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))

    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 invalid")

    return tree_head, None

def fetch_consistency_proof(first, second):
    post_data = 'old_size={}\n'.format(first)
    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:
        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
    while n > 0:
        if n & 1:
            p += 1
        n >>= 1
    return p

# Implements the algorithm for consistency proof verification outlined
# in RFC6962-BIS, see
# https://datatracker.ietf.org/doc/html/draft-ietf-trans-rfc6962-bis-39#section-2.1.4.2
def consistency_proof_valid(first, second, proof):
    assert(first.tree_size() == proof.old_size())
    assert(second.tree_size() == proof.new_size())

    path = proof.path()
    if len(path) == 0:
        return False
    if numbits(first.tree_size()) == 1:
        path = [first.root_hash()] + path

    fn = first.tree_size() - 1
    sn = second.tree_size() - 1
    while fn & 1:
        fn >>= 1
        sn >>= 1

    fr = path[0]
    sr = path[0]

    for c in path[1:]:
        if sn == 0:
            return False

        if fn & 1 or fn == sn:
            fr = sha256(b'\x01' + c + fr).digest()
            sr = sha256(b'\x01' + c + sr).digest()
            while fn != 0 and fn & 1 == 0:
                fn >>= 1
                sn >>= 1
        else:
            sr = sha256(b'\x01' + sr + c).digest()

        fn >>= 1
        sn >>= 1

    return sn == 0 and fr == first.root_hash() and sr == second.root_hash()

def sign_send_store_tree_head(signing_key, tree_head):
    hash = sha256(signing_key.verify_key.encode())
    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())

    req = requests.post(g_args.base_url + 'st/v0/add-cosignature', post_data)
    if req.status_code != 200:
        return (ERR_COSIG_POST,
                "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_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))

    assert(log_verification_key is not None)
    return log_verification_key, None

# Read signature key from file, or generate one and write it to file.
def ensure_sigkey(fn):
    try:
        os.stat(fn, follow_symlinks=False)
    except FileNotFoundError:
        print("INFO: Signing key file {} not found -- generating new signing key".format(fn))
        signing_key = nacl.signing.SigningKey.generate()
        print("INFO: verification key: {}".format(signing_key.verify_key.encode(nacl.encoding.HexEncoder).decode('ascii')))
        with open(fn, 'w') as f:
            os.chmod(f.fileno(), S_IRUSR)
            f.write(signing_key.encode(encoder=nacl.encoding.HexEncoder).decode('ascii'))

    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))
    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)))

    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))

    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):
    global g_args
    g_args = Parser()
    parse_config(CONFIG_FILE)
    parse_args(args)
    if g_args.save_config:
        # TODO write to config file
        return ERR_NYI, "ERROR: --save-config is not yet implemented"

    consistency_verified = False
    ignore_consistency = False

    make_base_dir_maybe()

    log_verification_key, err = ensure_log_verification_key()
    if err: return err

    signing_key, err = ensure_sigkey(g_args.sigkey_file)
    if err: return err

    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

    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)
    if status[1]:
        print(status[1])
    sys.exit(status[0])