#!/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 hashlib
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,
};

enum class CTLogFormat {
  RFC6962,
  Tiled,
};

struct CTLogInfo {
  // See bug 1338873 about making these fields const.
  const char* name;
  CTLogState state;
  CTLogFormat format;
  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
"""


LOG_NAME_TABLE_TEMPLATE = """\
/* 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. */

// prettier-ignore
export const logNameTable = $logNameTable
"""


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, $log_format,
     $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 in {"qualified", "usable", "readonly"}:
        return "CTLogState::Admissible"
    elif state == "retired":
        return "CTLogState::Retired"
    elif state in {"pending", "rejected"}:
        return None
    else:
        raise UnhandledLogStateException(f"unhandled log state '{state}'")


def get_initializer_for_log(log, operator, json_data, log_format):
    log_key = base64.b64decode(log["key"])
    operator_name = operator["name"]
    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:
        return None
    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
    tmpl = Template(LOG_INFO_TEMPLATE)
    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,
        log_format=log_format,
        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),
    )
    return prefix + toappend + suffix


def get_log_info_structs(json_data):
    """Return array of CTLogInfo initializers for the known logs."""
    initializers = []
    for operator in json_data["operators"]:
        if "logs" in operator:
            for log in operator["logs"]:
                initializer = get_initializer_for_log(
                    log, operator, json_data, "CTLogFormat::RFC6962"
                )
                if initializer:
                    initializers.append(initializer)
        if "tiled_logs" in operator:
            for log in operator["tiled_logs"]:
                initializer = get_initializer_for_log(
                    log, operator, json_data, "CTLogFormat::Tiled"
                )
                if initializer:
                    initializers.append(initializer)
    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 generate_log_name_table_file(json_data, out_file):
    """Generate an mjs file that defines a map from log["log_id"] to log["description"] for each of the known logs."""
    entries = {}
    for operator in json_data["operators"]:
        for log in operator.get("logs", []):
            entries[log["log_id"]] = log.get("description", "")
        for log in operator.get("tiled_logs", []):
            entries[log["log_id"]] = log.get("description", "")
    logNameTable = json.dumps(entries, indent=2, sort_keys=True)
    out_file.write(
        Template(LOG_NAME_TABLE_TEMPLATE).substitute(
            prog=os.path.basename(sys.argv[0]), logNameTable=logNameTable
        )
    )


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",
                    },
                },
            }
        ],
    }
    mozilla_test_operator_3 = {
        "name": "Mozilla Test Org 3",
        "id": max_id + 3,
        "test_only": True,
        "tiled_logs": [
            {
                "description": "Mozilla Test RSA Log 4",
                # `openssl x509 -noout -pubkey -in <path/to/evroot.pem>`
                "key": """
            MIIBIjANBgkqhkiG9w0BAQEFAAOCAQ8AMIIBCgKCAQEAtUmJXJ0AEI0Rofmfh6nj
            0aXbXfrs8YjaV79kE2iPLORyLP8QkDjBdALJOipHPb0oau0/3NNjz1opFHcBvdgY
            12Fb+TVr1dPIM2qqjAlxNooGw81EYbk+UUJOF0S7LqrUaqs4zhloIZYfh3FKFmNp
            Pwl2HN9Na6El6s7HvicNOI94n2zfeK4xRO0oxF55KThjp6IqSJgKNqQOctV5ybkl
            3/jHkzYv/WiXp8F1TF6XyWfD6t0aroqizM40igFpuA4ootcMGpYMbzNfLaCbm2Q/
            Wr+6SeiqqYHpYOJ9h0gL3VXdlBf6GFCfu1VMz4GkOX6LqBKKNL3yeGXBieVzT7Ip
            BQIDAQAB
        """,
                "state": {
                    "qualified": {
                        "timestamp": "2025-06-25T12:09:26Z",
                    },
                },
            }
        ],
    }

    def add_log_id(log):
        log["log_id"] = base64.b64encode(
            hashlib.sha256(base64.b64decode(log["key"])).digest()
        ).decode("utf-8")

    def add_log_ids(operator):
        for log in operator.get("tiled_logs", []):
            add_log_id(log)
        for log in operator.get("logs", []):
            add_log_id(log)

    add_log_ids(mozilla_test_operator_1)
    add_log_ids(mozilla_test_operator_2)
    add_log_ids(mozilla_test_operator_3)

    json_data["operators"].append(mozilla_test_operator_1)
    json_data["operators"].append(mozilla_test_operator_2)
    json_data["operators"].append(mozilla_test_operator_3)


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(
                f"unsupported hash algorithm '{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 cpp header output: ", args.out)
    with open(args.out, "w") as out_file:
        generate_cpp_header_file(json_data, out_file)

    print("Writing log name table output: ", args.log_name_table_out)
    with open(args.log_name_table_out, "w") as out_file:
        generate_log_name_table_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=f"Example: ./mach python {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(
        "--log-name-table-out",
        default=mozpath.join(
            buildconfig.topsrcdir,
            "toolkit",
            "components",
            "certviewer",
            "content",
            "components",
            "logNameTable.mjs",
        ),
        help="path and filename of the log name table to be generated (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()