#!/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
# file, You can obtain one at http://mozilla.org/MPL/2.0/.

"""
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
https://www.gstatic.com/ct/log_list/v3/log_list.json
See more information at https://certificate.transparency.dev/google/
"""

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
 * file, You can obtain one at http://mozilla.org/MPL/2.0/. */

/* 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 {
  // See bug 1338873 about making these fields const.
  const char* name;
  CTLogState state;
  uint64_t timestamp;
  // Index within kCTLogOperatorList.
  size_t operatorIndex;
  const char* key;
  size_t keyLength;
};

struct CTLogOperatorInfo {
  // See bug 1338873 about making these fields const.
  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([f"\\x{c:02x}" 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, f"No operators with id {target_name} defined."
    assert len(matches) == 1, f"Found multiple operators with id {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=f"operated by {operator_name}".replace("/", "|"),
                state=state,
                timestamp=timestamp,
                spaces=spaces,
                timestamp_comment=timestamp_comment,
                # Maximum line width is 80.
                indented_log_key="\n".join(
                    [f'     "{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 "
        "previously-downloaded file. See https://certificate.transparency.dev/google/",
        epilog="Example: ./mach python %s" % os.path.basename(sys.argv[0]),
    )

    arg_parser.add_argument(
        "--url",
        default="https://www.gstatic.com/ct/log_list/v3/log_list.json",
        help="download the known CT logs JSON file from the specified URL (default: %(default)s)",
    )
    arg_parser.add_argument(
        "--signature-url",
        default="https://www.gstatic.com/ct/log_list/v3/log_list.sig",
        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()