trezor-firmware/core/translations/cli.py

from __future__ import annotations

import datetime
import json
import logging
import typing as t
import subprocess
from pathlib import Path

import click

from trezorlib import cosi, models, merkle_tree
from trezorlib._internal import translations

HERE = Path(__file__).parent.resolve()
LOG = logging.getLogger(__name__)

ALL_MODELS = {models.T2B1, models.T2T1, models.T3T1}

PRIVATE_KEYS_DEV = [byte * 32 for byte in (b"\xdd", b"\xde", b"\xdf")]

PUBLIC_KEYS_PROD = [
    bytes.fromhex(key)
    for key in (
        "62cea8257b0bca15b33a76405a79c4881eb20ee82346813181a50291d6caec67",
        "594ef09e51139372e528c6c5b8742ee1806f9114caeaeedb04be6a98e1ce3020",
    )
]

VERSION_H = HERE.parent / "embed" / "firmware" / "version.h"
SIGNATURES_JSON = HERE / "signatures.json"


class SignedInfo(t.TypedDict):
    merkle_root: str
    signature: str
    datetime: str
    commit: str


class UnsignedInfo(t.TypedDict):
    merkle_root: str
    datetime: str
    commit: str


class SignatureFile(t.TypedDict):
    current: UnsignedInfo
    history: list[SignedInfo]


def _version_from_version_h() -> translations.VersionTuple:
    defines: t.Dict[str, int] = {}
    with open(VERSION_H) as f:
        for line in f:
            try:
                define, symbol, number = line.rstrip().split()
                assert define == "#define"
                defines[symbol] = int(number)
            except Exception:
                # not a #define, not a number, wrong number of parts
                continue

    return (
        defines["VERSION_MAJOR"],
        defines["VERSION_MINOR"],
        defines["VERSION_PATCH"],
        defines["VERSION_BUILD"],
    )


def make_tree_info(merkle_root: bytes) -> UnsignedInfo:
    now = datetime.datetime.utcnow()
    commit = (
        subprocess.check_output(["git", "rev-parse", "HEAD"], cwd=HERE)
        .decode("ascii")
        .strip()
    )
    return UnsignedInfo(
        merkle_root=merkle_root.hex(), datetime=now.isoformat(), commit=commit
    )


def sign_info(info: UnsignedInfo, signature: bytes) -> SignedInfo:
    return SignedInfo(signature=signature.hex(), **info)


def update_merkle_root(signature_file: SignatureFile, merkle_root: bytes) -> bool:
    """Update signatures.json with the new Merkle root.

    Returns True if the signature file was updated, False if it was already up-to-date.
    """
    current = signature_file["current"]

    if current["merkle_root"] == merkle_root.hex():
        # Merkle root is already up to date
        return False

    # overwrite with a new one
    signature_file["current"] = make_tree_info(merkle_root)
    SIGNATURES_JSON.write_text(json.dumps(signature_file, indent=2))
    return True


def generate_all_blobs(rewrite_version: bool) -> list[translations.TranslationsBlob]:
    order = translations.order_from_json(json.loads((HERE / "order.json").read_text()))
    fonts_dir = HERE / "fonts"

    current_version = _version_from_version_h()
    current_version_str = ".".join(str(v) for v in current_version)

    common_version = None

    all_languages = [lang_file.stem for lang_file in HERE.glob("??.json")]
    all_blobs: list[translations.TranslationsBlob] = []
    for lang in all_languages:
        if lang == "en":
            continue

        for model in ALL_MODELS:
            try:
                blob_json = json.loads((HERE / f"{lang}.json").read_text())
                blob_version = translations.version_from_json(
                    blob_json["header"]["version"]
                )
                if rewrite_version:
                    version = current_version
                    if blob_version != current_version:
                        blob_json["header"]["version"] = current_version_str
                        (HERE / f"{lang}.json").write_text(
                            json.dumps(blob_json, indent=2) + "\n"
                        )

                else:
                    version = blob_version
                    if common_version is None:
                        common_version = version
                    elif blob_version != common_version:
                        raise ValueError(
                            f"Language {lang} has version {version} but expected {common_version}"
                        )

                blob = translations.blob_from_defs(
                    blob_json, order, model, version, fonts_dir
                )
                all_blobs.append(blob)
            except Exception as e:
                import traceback

                traceback.print_exc()
                LOG.warning(f"Failed to build {lang} for {model.internal_name}: {e}")
                continue

            LOG.info(f"Built {lang} for {model.internal_name}")

    return all_blobs


def build_all_blobs(
    all_blobs: list[translations.TranslationsBlob],
    merkle_tree: merkle_tree.MerkleTree,
    sigmask: int,
    signature: bytes,
    production: bool = False,
) -> None:
    for blob in all_blobs:
        proof = translations.Proof(
            merkle_proof=merkle_tree.get_proof(blob.header_bytes),
            signature=signature,
            sigmask=sigmask,
        )
        blob.proof = proof
        header = blob.header
        model = header.model.value.decode("ascii")
        version = ".".join(str(v) for v in header.firmware_version[:3])
        if production:
            suffix = ""
        else:
            suffix = "-unsigned"
        filename = f"translation-{model}-{header.language}-{version}{suffix}.bin"
        (HERE / filename).write_bytes(blob.build())
        LOG.info(f"Wrote {header.language} for {model} v{version}: {filename}")


@click.group()
def cli() -> None:
    pass


@cli.command()
@click.option("--signed", is_flag=True, help="Generate signed blobs.")
def gen(signed: bool | None) -> None:
    """Generate all language blobs for all models.

    The generated blobs will be signed with the development keys.
    """
    all_blobs = generate_all_blobs(rewrite_version=True)
    tree = merkle_tree.MerkleTree(b.header_bytes for b in all_blobs)
    root = tree.get_root_hash()

    signature_file: SignatureFile = json.loads(SIGNATURES_JSON.read_text())

    if signed:
        for entry in signature_file["history"]:
            if entry["merkle_root"] == root.hex():
                signature_hex = entry["signature"]
                signature_bytes = bytes.fromhex(signature_hex)
                sigmask, signature = signature_bytes[0], signature_bytes[1:]
                build_all_blobs(all_blobs, tree, sigmask, signature, production=True)
                return
        else:
            raise click.ClickException(
                "No matching signature found in signatures.json. Run `cli.py sign` first."
            )
        
    signature = cosi.sign_with_privkeys(root, PRIVATE_KEYS_DEV)
    sigmask = 0b111
    build_all_blobs(all_blobs, tree, sigmask, signature)
    if update_merkle_root(signature_file, root):
        SIGNATURES_JSON.write_text(json.dumps(signature_file, indent=2) + "\n")
        click.echo("Updated signatures.json")
    else:
        click.echo("signatures.json is already up-to-date")


@cli.command()
def merkle_root() -> None:
    """Print the Merkle root of all language blobs."""
    all_blobs = generate_all_blobs(rewrite_version=False)
    tree = merkle_tree.MerkleTree(b.header_bytes for b in all_blobs)
    root = tree.get_root_hash()

    signature_file: SignatureFile = json.loads(SIGNATURES_JSON.read_text())
    if signature_file["current"]["merkle_root"] != root.hex():
        raise click.ClickException(
            f"Merkle root mismatch!\n"
            f"Expected:                  {root.hex()}\n"
            f"Stored in signatures.json: {signature_file['current']['merkle_root']}\n"
            "Run `cli.py gen` to update the stored Merkle root."
        )

    click.echo(root.hex())


@cli.command()
@click.argument("signature_hex")
@click.option("--force", is_flag=True, help="Write even if the signature is invalid.")
def sign(signature_hex: str, force: bool | None) -> None:
    """Insert a signature into language blobs."""
    all_blobs = generate_all_blobs(rewrite_version=False)
    tree = merkle_tree.MerkleTree(b.header_bytes for b in all_blobs)
    root = tree.get_root_hash()

    signature_file: SignatureFile = json.loads(SIGNATURES_JSON.read_text())
    if signature_file["current"]["merkle_root"] != root.hex():
        raise click.ClickException(
            f"Merkle root mismatch!\n"
            f"Expected:                  {root.hex()}\n"
            f"Stored in signatures.json: {signature_file['current']['merkle_root']}"
        )

    # Update signature file data. It will be written only if the signature verifies.
    tree_info = make_tree_info(root)
    signed_info = sign_info(tree_info, bytes.fromhex(signature_hex))
    signature_file["history"].insert(0, signed_info)

    signature_bytes = bytes.fromhex(signature_hex)
    sigmask, signature = signature_bytes[0], signature_bytes[1:]

    try:
        cosi.verify(signature, root, 2, PUBLIC_KEYS_PROD, sigmask)
    except Exception as e:
        if force:
            LOG.warning(f"Invalid signature: {e}. --force is provided, writing anyway.")
            pass
        else:
            raise click.ClickException(f"Invalid signature: {e}") from e

    SIGNATURES_JSON.write_text(json.dumps(signature_file, indent=2) + "\n")
    build_all_blobs(all_blobs, tree, sigmask, signature, production=True)


if __name__ == "__main__":
    cli()