apache-http-server/support/ctauditscts

#!/usr/bin/env python
#
# Licensed to the Apache Software Foundation (ASF) under one or more
# contributor license agreements.  See the NOTICE file distributed with
# this work for additional information regarding copyright ownership.
# The ASF licenses this file to You under the Apache License, Version 2.0
# (the "License"); you may not use this file except in compliance with
# the License.  You may obtain a copy of the License at
#
#     http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS,
# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
# See the License for the specific language governing permissions and
# limitations under the License.

import binascii
import os
import sqlite3
import ssl
import struct
import sys
import tempfile

from contextlib import closing

SERVER_START = 1
KEY_START = 2
CERT_START = 3
SCT_START = 4


def usage():
    print >> sys.stderr, ('Usage: %s /path/to/audit/files ' +
                          '[/path/to/log-config-db]') % sys.argv[0]
    sys.exit(1)


def audit(fn, tmp, already_checked, cur):
    print 'Auditing %s...' % fn

    # First, parse the audit file into a series of related
    #
    #   1. PEM file with certificate chain
    #   2. Individual SCT files
    #
    # Next,  for each SCT, invoke verify_single_proof to verify.
    log_bytes = open(fn, 'rb').read()
    offset = 0
    while offset < len(log_bytes):
        print 'Got package from server...'
        val = struct.unpack_from('>H', log_bytes, offset)
        assert val[0] == SERVER_START
        offset += 2

        assert struct.unpack_from('>H', log_bytes, offset)[0] == KEY_START
        offset += 2

        key_size = struct.unpack_from('>H', log_bytes, offset)[0]
        assert key_size > 0
        offset += 2

        key = log_bytes[offset:offset + key_size]
        offset += key_size

        # at least one certificate
        assert struct.unpack_from('>H', log_bytes, offset)[0] == CERT_START

        # for each certificate:
        leaf = None
        while struct.unpack_from('>H', log_bytes, offset)[0] == CERT_START:
            offset += 2
            val = struct.unpack_from('BBB', log_bytes, offset)
            offset += 3
            der_size = (val[0] << 16) | (val[1] << 8) | (val[2] << 0)
            print '  Certificate size:', hex(der_size)
            if not leaf:
                leaf = (offset, der_size)
            offset += der_size

        pem = ssl.DER_cert_to_PEM_cert(log_bytes[leaf[0]:leaf[0] + leaf[1]])

        tmp_leaf_pem = tempfile.mkstemp(text=True)
        with closing(os.fdopen(tmp_leaf_pem[0], 'w')) as f:
            f.write(pem)

        # at least one SCT
        assert struct.unpack_from('>H', log_bytes, offset)[0] == SCT_START

        # for each SCT:
        while offset < len(log_bytes) and \
                struct.unpack_from('>H', log_bytes, offset)[0] == SCT_START:
            offset += 2
            len_offset = offset
            sct_size = struct.unpack_from('>H', log_bytes, len_offset)[0]
            offset += 2
            print '  SCT size:', hex(sct_size)
            log_id = log_bytes[offset + 1:offset + 1 + 32]
            log_id_hex = binascii.hexlify(log_id).upper()
            print '    Log id: %s' % log_id_hex
            timestamp_ms = struct.unpack_from('>Q', log_bytes, offset + 33)[0]
            print '    Timestamp: %s' % timestamp_ms

            #  If we ever need the full SCT: sct = (offset, sct_size)
            offset += sct_size

            if key in already_checked:
                print '  (SCTs already checked)'
                continue

            already_checked[key] = True

            log_url_arg = ''
            if cur:
                stmt = 'SELECT * FROM loginfo WHERE log_id = ?'
                cur.execute(stmt, [log_id_hex])
                recs = list(cur.fetchall())
                if len(recs) > 0 and recs[0][6] is not None:
                    log_url = recs[0][6]

                    # verify_single_proof doesn't accept <scheme>://
                    if '://' in log_url:
                        log_url = log_url.split('://')[1]
                    log_url_arg = '--log_url %s' % log_url

                    print '    Log URL: ' + log_url

            cmd = 'verify_single_proof.py --cert %s --timestamp %s %s' % \
                  (tmp_leaf_pem[1], timestamp_ms, log_url_arg)
            print '>%s<' % cmd
            os.system(cmd)

        os.unlink(tmp_leaf_pem[1])


def main():
    if len(sys.argv) != 2 and len(sys.argv) != 3:
        usage()

    top = sys.argv[1]
    tmp = '/tmp'

    if len(sys.argv) == 3:
        cxn = sqlite3.connect(sys.argv[2])
        cur = cxn.cursor()
    else:
        cur = None

    # could serialize this between runs to further limit duplicate checking
    already_checked = dict()

    for dirpath, dnames, fnames in os.walk(top):
        fnames = [fn for fn in fnames if fn[-4:] == '.out']
        for fn in fnames:
            audit(os.path.join(dirpath, fn), tmp, already_checked, cur)


if __name__ == "__main__":
    main()