Source code
Revision control
Copy as Markdown
Other Tools
#!/usr/bin/env python3
# This Source Code Form is subject to the terms of the Mozilla Public
# License, v. 2.0. If a copy of the MPL was not distributed with this
"""
Parses a JSON file listing the known Certificate Transparency logs
(log_list.json) and generates a C++ header file to be included in Firefox.
The current log_list.json file available under security/manager/tools
was originally downloaded from
"""
import argparse
import base64
import datetime
import json
import os.path
import ssl
import sys
import textwrap
import time
from string import Template
from urllib.request import urlopen
import buildconfig
import certifi
import mozpack.path as mozpath
import rsa
from pyasn1.codec.der import decoder
from pyasn1_modules import pem, rfc2314
OUTPUT_TEMPLATE = """\
/* -*- Mode: C++; tab-width: 8; indent-tabs-mode: nil; c-basic-offset: 2 -*- */
/* vim: set ts=8 sts=2 et sw=2 tw=80: */
/* This Source Code Form is subject to the terms of the Mozilla Public
* License, v. 2.0. If a copy of the MPL was not distributed with this
/* This file was automatically generated by $prog. */
#ifndef $include_guard
#define $include_guard
#include "CTLog.h"
#include "prtime.h"
#include <stddef.h>
static const PRTime kCTExpirationTime = INT64_C($expiration_time);
namespace mozilla::ct {
enum class CTLogState {
Admissible, // Qualified, Usable, or ReadOnly
Retired,
};
struct CTLogInfo {
const char* name;
CTLogState state;
uint64_t timestamp;
// Index within kCTLogOperatorList.
size_t operatorIndex;
const char* key;
size_t keyLength;
};
struct CTLogOperatorInfo {
const char* name;
mozilla::ct::CTLogOperatorId id;
};
const CTLogInfo kCTLogList[] = {
$logs
};
const CTLogOperatorInfo kCTLogOperatorList[] = {
$operators
};
} // namespace mozilla::ct
#endif // $include_guard
"""
def get_timestamp(time_str):
"""
Convert a time string such as "2017-01-01T00:00:00Z" to an integer
representing milliseconds since the epoch.
Timezones in the string are not supported and will result in an exception.
"""
t = datetime.datetime.strptime(time_str, "%Y-%m-%dT%H:%M:%SZ")
epoch = datetime.datetime.utcfromtimestamp(0)
seconds_since_epoch = (t - epoch).total_seconds()
return int(seconds_since_epoch * 1000)
def get_hex_lines(blob, width):
"""Convert a binary string to a multiline text of C escape sequences."""
text = "".join(["\\x{:02x}".format(c) for c in blob])
# When escaped, a single byte takes 4 chars (e.g. "\x00").
# Make sure we don't break an escaped byte between the lines.
return textwrap.wrap(text, width - width % 4)
def get_operator_index(json_data, target_name):
"""Return operator's entry from the JSON along with its array index."""
matches = [
(operator, index)
for (index, operator) in enumerate(json_data["operators"])
if operator["name"] == target_name
]
assert len(matches) != 0, "No operators with id {0} defined.".format(target_name)
assert len(matches) == 1, "Found multiple operators with id {0}.".format(
target_name
)
return matches[0][1]
LOG_INFO_TEMPLATE = """\
{$description, $state,
$timestamp, // $timestamp_comment
$operator_index,$spaces // $operator_comment
$indented_log_key,
$log_key_len}"""
class UnhandledLogStateException(Exception):
pass
def map_state(state):
"""
Maps a log state string to the appropriate CTLogState enum value or None,
if the log state indicates that the log should not be included. Valid
states to be included are 'qualified', 'usable', 'readonly', or 'retired'.
Valid states that are not to be included are 'pending' or 'rejected'.
"""
if state == "qualified" or state == "usable" or state == "readonly":
return "CTLogState::Admissible"
elif state == "retired":
return "CTLogState::Retired"
elif state == "pending" or state == "rejected":
return None
else:
raise UnhandledLogStateException("unhandled log state '%s'" % state)
def get_log_info_structs(json_data):
"""Return array of CTLogInfo initializers for the known logs."""
tmpl = Template(LOG_INFO_TEMPLATE)
initializers = []
for operator in json_data["operators"]:
operator_name = operator["name"]
for log in operator["logs"]:
log_key = base64.b64decode(log["key"])
operator_index = get_operator_index(json_data, operator_name)
state = list(log["state"].keys())[0]
timestamp_comment = log["state"][state]["timestamp"]
timestamp = get_timestamp(timestamp_comment)
state = map_state(state)
if state is None:
continue
is_test_log = "test_only" in operator and operator["test_only"]
prefix = ""
suffix = ","
if is_test_log:
prefix = "#ifdef DEBUG\n"
suffix = ",\n#endif // DEBUG"
num_spaces = len(str(timestamp)) - len(str(operator_index))
spaces = " " * num_spaces
toappend = tmpl.substitute(
# Use json.dumps for C-escaping strings.
# Not perfect but close enough.
description=json.dumps(log["description"]),
operator_index=operator_index,
operator_comment="operated by {0}".
# The comment must not contain "/".
format(operator_name).replace("/", "|"),
state=state,
timestamp=timestamp,
spaces=spaces,
timestamp_comment=timestamp_comment,
# Maximum line width is 80.
indented_log_key="\n".join(
[' "{0}"'.format(l) for l in get_hex_lines(log_key, 74)]
),
log_key_len=len(log_key),
)
initializers.append(prefix + toappend + suffix)
return initializers
def get_log_operator_structs(json_data):
"""Return array of CTLogOperatorInfo initializers."""
tmpl = Template(" {$name, $id}")
initializers = []
currentId = 0
for operator in json_data["operators"]:
prefix = ""
suffix = ","
is_test_log = "test_only" in operator and operator["test_only"]
if is_test_log:
prefix = "#ifdef DEBUG\n"
suffix = ",\n#endif // DEBUG"
toappend = tmpl.substitute(name=json.dumps(operator["name"]), id=currentId)
currentId += 1
initializers.append(prefix + toappend + suffix)
return initializers
TEN_WEEKS_IN_SECONDS = 60 * 60 * 24 * 7 * 10
MICROSECONDS_PER_SECOND = 1000000
def generate_cpp_header_file(json_data, out_file):
"""Generate the C++ header file for the known logs."""
filename = os.path.basename(out_file.name)
include_guard = filename.replace(".", "_").replace("/", "_")
log_info_initializers = get_log_info_structs(json_data)
operator_info_initializers = get_log_operator_structs(json_data)
expiration_time = (
int(time.time()) + TEN_WEEKS_IN_SECONDS
) * MICROSECONDS_PER_SECOND
out_file.write(
Template(OUTPUT_TEMPLATE).substitute(
prog=os.path.basename(sys.argv[0]),
include_guard=include_guard,
logs="\n".join(log_info_initializers),
operators="\n".join(operator_info_initializers),
expiration_time=expiration_time,
)
)
def patch_in_test_logs(json_data):
"""Insert Mozilla-specific test log data."""
max_id = len(json_data["operators"])
mozilla_test_operator_1 = {
"name": "Mozilla Test Org 1",
"id": max_id + 1,
"test_only": True,
"logs": [
{
"description": "Mozilla Test RSA Log 1",
# `openssl x509 -noout -pubkey -in <path/to/default-ee.pem>`
"key": """
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAuohRqESOFtZB/W62iAY2
ED08E9nq5DVKtOz1aFdsJHvBxyWo4NgfvbGcBptuGobya+KvWnVramRxCHqlWqdF
h/cc1SScAn7NQ/weadA4ICmTqyDDSeTbuUzCa2wO7RWCD/F+rWkasdMCOosqQe6n
cOAPDY39ZgsrsCSSpH25iGF5kLFXkD3SO8XguEgfqDfTiEPvJxbYVbdmWqp+ApAv
OnsQgAYkzBxsl62WYVu34pYSwHUxowyR3bTK9/ytHSXTCe+5Fw6naOGzey8ib2nj
tIqVYR3uJtYlnauRCE42yxwkBCy/Fosv5fGPmRcxuLP+SSP6clHEMdUDrNoYCjXt
jQIDAQAB
""",
"state": {
"qualified": {
"timestamp": "2024-07-22T16:44:26Z",
},
},
},
{
"description": "Mozilla Test EC Log",
# `openssl x509 -noout -pubkey -in <path/to/root_secp256r1_256.pem`
"key": """
MFkwEwYHKoZIzj0CAQYIKoZIzj0DAQcDQgAET7+7u2Hg+PmxpgpZrIcE4uwFC0I+
PPcukj8sT3lLRVwqadIzRWw2xBGdBwbgDu3I0ZOQ15kbey0HowTqoEqmwA==
""",
"state": {
"qualified": {
"timestamp": "2024-07-22T16:44:26Z",
},
},
},
],
}
mozilla_test_operator_2 = {
"name": "Mozilla Test Org 2",
"id": max_id + 2,
"test_only": True,
"logs": [
{
"description": "Mozilla Test RSA Log 2",
# `openssl x509 -noout -pubkey -in <path/to/other-test-ca.pem>`
"key": """
MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAwXXGUmYJn3cIKmeR8bh2
w39c5TiwbErNIrHL1G+mWtoq3UHIwkmKxKOzwfYUh/QbaYlBvYClHDwSAkTFhKTE
SDMF5ROMAQbPCL6ahidguuai6PNvI8XZgxO53683g0XazlHU1tzSpss8xwbrzTBw
7JjM5AqlkdcpWn9xxb5maR0rLf7ISURZC8Wj6kn9k7HXU0BfF3N2mZWGZiVHl+1C
aQiICBFCIGmYikP+5Izmh4HdIramnNKDdRMfkysSjOKG+n0lHAYq0n7wFvGHzdVO
gys1uJMPdLqQqovHYWckKrH9bWIUDRjEwLjGj8N0hFcyStfehuZVLx0eGR1xIWjT
uwIDAQAB
""",
"state": {
"qualified": {
"timestamp": "2024-07-22T16:44:26Z",
},
},
}
],
}
json_data["operators"].append(mozilla_test_operator_1)
json_data["operators"].append(mozilla_test_operator_2)
def get_content_at(url):
print("Fetching URL: ", url)
ssl_context = ssl.create_default_context(cafile=certifi.where())
f = urlopen(url, context=ssl_context)
return f.read()
def read_rsa_key(path):
"""
Read the PEM subject public key info at the given path and
return it as an RSA public key.
"""
with open(path) as f:
spki = pem.readPemFromFile(
f, "-----BEGIN PUBLIC KEY-----", "-----END PUBLIC KEY-----"
)
decoded, _ = decoder.decode(spki, rfc2314.SubjectPublicKeyInfo())
return rsa.PublicKey.load_pkcs1(
decoded["subjectPublicKey"].asOctets(), format="DER"
)
class UnsupportedSignatureHashAlgorithmException(Exception):
pass
def run(args):
"""
Load the input JSON file and generate the C++ header according to the
command line arguments.
"""
if args.json_file:
print("Reading file: ", args.json_file)
with open(args.json_file, "rb") as json_file:
json_text = json_file.read()
else:
json_text = get_content_at(args.url)
signature = get_content_at(args.signature_url)
key = read_rsa_key(args.key_file)
print("Validating signature...")
hash_alg = rsa.verify(json_text, signature, key)
if hash_alg != "SHA-256":
raise UnsupportedSignatureHashAlgorithmException(
"unsupported hash algorithm '%s'" % hash_alg
)
print("Writing output: ", args.json_file_out)
with open(args.json_file_out, "wb") as json_file_out:
json_file_out.write(json_text)
json_data = json.loads(json_text)
patch_in_test_logs(json_data)
print("Writing output: ", args.out)
with open(args.out, "w") as out_file:
generate_cpp_header_file(json_data, out_file)
print("Done.")
def parse_arguments_and_run():
"""Parse the command line arguments and run the program."""
arg_parser = argparse.ArgumentParser(
description="Parses a JSON file listing the known "
"Certificate Transparency logs and generates "
"a C++ header file to be included in Firefox."
"Downloads the JSON file from the known source "
"of truth by default, but can also operate on a "
epilog="Example: ./mach python %s" % os.path.basename(sys.argv[0]),
)
arg_parser.add_argument(
"--url",
help="download the known CT logs JSON file from the specified URL (default: %(default)s)",
)
arg_parser.add_argument(
"--signature-url",
help="download the signature on the known CT logs JSON file from the specified URL (default: %(default)s)",
)
arg_parser.add_argument(
"--key-file",
default=mozpath.join(
buildconfig.topsrcdir, "security", "manager", "tools", "log_list_pubkey.pem"
),
help="verify the signature on the downloaded CT logs JSON file with the key in the specified file (default: %(default)s)",
)
arg_parser.add_argument(
"--json-file",
nargs="?",
const=mozpath.join(
buildconfig.topsrcdir, "security", "manager", "tools", "log_list.json"
),
help="read the known CT logs JSON data from the specified file (default: %(const)s)",
)
arg_parser.add_argument(
"--json-file-out",
default=mozpath.join(
buildconfig.topsrcdir, "security", "manager", "tools", "log_list.json"
),
help="write the known CT logs JSON data to the specified file when downloading it from the given url (default: %(default)s)",
)
arg_parser.add_argument(
"--out",
default=mozpath.join(buildconfig.topsrcdir, "security", "ct", "CTKnownLogs.h"),
help="path and filename of the header file to be generated (default: %(default)s)",
)
run(arg_parser.parse_args())
if __name__ == "__main__":
parse_arguments_and_run()