templates/c/firmware-info/tools/identity.py

"""Build provenance and exact-file SHA-256, shared by CCS, Keil and CLI builds.

prepare runs BEFORE compiling; seal runs AFTER binary conversion and size checks.
The digest is external to the image, avoiding the self-hashing paradox. No MCU
protocol encoding lives here: the generated header contains scalar constants.
"""
from __future__ import annotations

import argparse
from datetime import datetime, timedelta, timezone
import hashlib
import json
import os
import re
import shutil
import tempfile
from pathlib import Path
import subprocess


def git_state(repository: Path) -> dict:
    """Record full commits, and flag untracked sources as well as tracked edits.

    Generated outputs must be ignored in the repository. Unlike Climate's older
    helper, a new, untracked source file must not look like a clean release.
    """
    def git(*args):
        return subprocess.check_output(
            ["git", "-C", str(repository), *args], stderr=subprocess.DEVNULL,
            text=True, encoding="utf-8").strip()
    try:
        commit = git("rev-parse", "HEAD")
        dirty = bool(git("status", "--porcelain", "--untracked-files=normal"))
        return {"commit": commit, "dirty": dirty}
    except (OSError, subprocess.CalledProcessError):
        return {"commit": None, "dirty": None}


def write_atomic(path: Path, text: str):
    path.parent.mkdir(parents=True, exist_ok=True)
    temporary = path.with_name(path.name + ".tmp")
    # Built-in open supports newline on Python versions before 3.10.
    with open(temporary, "w", encoding="utf-8", newline="\n") as stream:
        stream.write(text)
    temporary.replace(path)


def next_version(version):  # Вычисляет следующий номер по ограничениям общего контракта firmware-info v1.
    result = list(version)  # Сохраняет исходный номер до успешной проверки полного переполнения.
    for index, limit in ((2, 999), (1, 255), (0, 255)):  # Patch имеет диапазон 0..999, minor и major занимают по одному октету 0..255.
        if result[index] < limit:  # Находит младший разряд, который ещё можно увеличить.
            result[index] += 1  # Резервирует следующий номер попытки сборки, включая будущие неудачные компиляции.
            return result  # Старшие разряды сохраняются, переполненные младшие уже обнулены.
        result[index] = 0  # Переносит единицу в следующий разряд после достижения предела текущего.
    raise ValueError("Firmware version exhausted at 255.255.999")  # Не допускает повторного использования номера после полного переполнения.


def prepare(repository: Path, templates: Path, config: Path, output: Path) -> dict:
    settings = json.loads(config.read_text(encoding="utf-8-sig"))
    version = settings["version"]
    if not isinstance(version, list) or len(version) != 3 or any(type(value) is not int or not 0 <= value <= limit for value, limit in zip(version, (255, 255, 999))):  # Проверяет JSON-массив трёх целых разрядов до любого изменения сохраняемого счётчика.
        raise ValueError("version must be [major 0..255, minor 0..255, patch 0..999]")  # Ошибочные типы, логические значения и выход за пределы контракта останавливают pre-build.
    if type(settings.get("increment_version", False)) is not bool:  # Отсутствующий флаг сохраняет ручной режим старых проектов; строки true/false не считаются настройкой.
        raise ValueError("increment_version must be true or false")  # Требует явное логическое значение JSON вместо неявного преобразования Python.
    for field in ("product", "target"):
        if not isinstance(settings.get(field), str) or not settings[field].strip():
            raise ValueError(f"missing {field}")
    if settings.get("increment_version", False):  # Прямые вызовы templates из CCS, Keil и CLI используют один и тот же сохраняемый итератор.
        version = settings["version"] = next_version(version)  # Вычисляет номер для JSON, заголовка и метаданных только один раз.
        with tempfile.NamedTemporaryFile(mode="w", encoding="utf-8", newline="\n", dir=config.parent, prefix=".firmware-version-", delete=False) as stream:  # Временный файл на том же диске позволяет атомарно заменить настройки без частичной записи.
            temporary = Path(stream.name)  # Запоминает только собственный файл, чтобы не затронуть пользовательские данные при очистке.
            try: stream.write(json.dumps(settings, ensure_ascii=False, indent=2) + "\n")  # Сохраняет новый номер и все дополнительные поля проекта в UTF-8.
            except BaseException: stream.close(); temporary.unlink(); raise  # Ошибка записи оставляет исходный JSON целым и удаляет незавершённый файл.
        try: temporary.replace(config)  # Занимает номер до дальнейших шагов, поэтому отказ генерации или компиляции не повторит его.
        finally: temporary.unlink(missing_ok=True)  # Удаляет временный файл при ошибке замены; после успеха он уже переименован.
    source, dependency = git_state(repository), git_state(templates)
    # A clean application built with a modified templates checkout is dirty too.
    known = source["commit"] is not None and dependency["commit"] is not None
    dirty = bool(source["dirty"] or dependency["dirty"])
    build_id = (source["commit"][:7] + "+" if dirty else source["commit"][:8]) if known else "NOGIT000"
    # UTC has an explicit interpretation across build PCs; SOURCE_DATE_EPOCH is
    # supported for reproducible release builds, not silently inferred from Git.
    epoch = os.environ.get("SOURCE_DATE_EPOCH")
    stamp = datetime.fromtimestamp(int(epoch), timezone.utc) if epoch else datetime.now(timezone.utc)
    stamp = stamp.replace(microsecond=0)
    if not 2000 <= stamp.year <= 2199:
        raise ValueError("build year outside firmware-info v1 range")
    metadata = {"schema": 1, "product": settings["product"], "target": settings["target"],
                "version": version, "build_id": build_id, "built_at_utc": stamp.isoformat(),
                "source": source, "templates": dependency,
                "identity_contract": 1, "identity_register_count": 12}
    constants = dict(zip(("FIRMWARE_VERSION_MAJOR", "FIRMWARE_VERSION_MINOR", "FIRMWARE_VERSION_PATCH"), version))
    constants.update(dict(zip(("FI_YEAR", "FI_MONTH", "FI_DAY", "FI_HOUR", "FI_MINUTE", "FI_SECOND"),
                              (stamp.year, stamp.month, stamp.day, stamp.hour, stamp.minute, stamp.second))))
    constants.update({f"FI_ID_{i}": ord(c) for i, c in enumerate(build_id)})
    header = "/* Generated before compilation. Do not edit or commit. UTC stamp. */\n#ifndef FIRMWARE_IDENTITY_H\n#define FIRMWARE_IDENTITY_H\n"
    header += f'#define FIRMWARE_BUILD_ID "{build_id}"\n'
    header += "".join(f"#define {key} {value}U\n" for key, value in constants.items())
    header += "#endif\n"
    # Rewriting the header forces the identity compilation unit to be rebuilt.
    # Sharing this directory between concurrent configurations is unsupported.
    write_atomic(output / "firmware_identity.h", header)
    write_atomic(output / "firmware_identity.json", json.dumps(metadata, indent=2) + "\n")
    return metadata


def seal(image: Path, metadata_path: Path) -> dict:
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    data = image.read_bytes()
    if not data:
        raise ValueError("empty firmware image")
    # Seal records the exact file sent to the programmer, including SCI8 headers.
    # It does not claim that the device has measured RAM/Flash or authenticated it.
    metadata.update({"file": image.name, "size_bytes": len(data),
                     "sha256": hashlib.sha256(data).hexdigest()})
    write_atomic(image.with_suffix(image.suffix + ".manifest.json"), json.dumps(metadata, indent=2) + "\n")
    write_atomic(image.with_suffix(image.suffix + ".sha256"), f'{metadata["sha256"]}  {image.name}\n')
    return metadata


def verify(image: Path, manifest: Path) -> dict:
    metadata = json.loads(manifest.read_text(encoding="utf-8"))
    data = image.read_bytes()
    if len(data) != metadata["size_bytes"] or hashlib.sha256(data).hexdigest() != metadata["sha256"]:
        raise ValueError("firmware size/SHA-256 does not match manifest")
    return metadata


def build_folder_name(metadata: dict) -> str:
    """Use the firmware commit and pre-build time rendered in UTC+3."""
    source = metadata.get("source") or {}
    dependency = metadata.get("templates") or {}
    commit = source.get("commit")
    if commit is not None:
        if not isinstance(commit, str) or not re.fullmatch(r"[0-9a-fA-F]{40,64}", commit):
            raise ValueError("invalid firmware commit in pre-build metadata")
        identity = commit[:8].lower()
        if source.get("dirty") or dependency.get("dirty"):
            identity += "+"
    else:
        identity = metadata.get("build_id", "NOGIT000")
        if not isinstance(identity, str) or not re.fullmatch(r"[0-9a-fA-F]{8}|[0-9a-fA-F]{7}\+|NOGIT000", identity):
            raise ValueError("invalid build ID in pre-build metadata")
    stamp_text = metadata.get("built_at_utc")
    if not isinstance(stamp_text, str):
        raise ValueError("missing pre-build UTC timestamp")
    stamp = datetime.fromisoformat(stamp_text.replace("Z", "+00:00"))
    if stamp.tzinfo is None:
        raise ValueError("pre-build timestamp must include a timezone")
    stamp = stamp.astimezone(timezone(timedelta(hours=3)))
    return identity + "_" + stamp.strftime("%Y-%m-%d_%H-%M-%S_UTC+3")


def export_build(folder: Path, export: Path, image_name: str) -> Path:
    """Copy the current complete build folder, without scanning old archives."""
    folder, export = folder.resolve(), export.resolve()
    if export == folder or folder in export.parents:
        raise ValueError("export directory must be outside the current build folder")
    destination = export / folder.name
    verify(folder / image_name, folder / (image_name + ".manifest.json"))
    export.mkdir(parents=True, exist_ok=True)
    if destination != folder:
        shutil.copytree(folder, destination, dirs_exist_ok=True)
    verify(destination / image_name, destination / (image_name + ".manifest.json"))
    print(f"[firmware-build] Exported folder: {destination}")
    return destination

def archive_build(image: Path, metadata_path: Path, output: Path = None,
                  extra_files=(), export: Path = None) -> dict:
    """Keep a byte-identical build snapshot, alongside the existing flat output."""
    image, metadata_path = image.resolve(), metadata_path.resolve()
    metadata = json.loads(metadata_path.read_text(encoding="utf-8"))
    name = build_folder_name(metadata)
    output = (output if output is not None else image.parent / "firmware").resolve()
    output.mkdir(parents=True, exist_ok=True)
    destination = output / name
    # Seal and verify staged copies before publishing files in the build folder.
    with tempfile.TemporaryDirectory(prefix=".archive-", dir=output) as temporary:
        stage = Path(temporary)
        shutil.copy2(image, stage / image.name)
        for file in extra_files:
            file = Path(file)
            shutil.copy2(file, stage / file.name)
        shutil.copy2(metadata_path, stage / "firmware_identity.json")
        seal(stage / image.name, stage / "firmware_identity.json")
        verify(stage / image.name, stage / (image.name + ".manifest.json"))
        destination.mkdir(parents=True, exist_ok=True)
        for file in stage.iterdir():
            file.replace(destination / file.name)
    print(f"[firmware-build] Folder: {destination}")
    result = {"directory": str(destination), "file": image.name}
    if export is not None:
        result["exported_directory"] = str(export_build(destination, export, image.name))
    return result

def main():
    parser = argparse.ArgumentParser(description=__doc__)
    sub = parser.add_subparsers(dest="command", required=True)
    p = sub.add_parser("prepare")
    for name in ("repository", "templates", "config", "output"):
        p.add_argument("--" + name, type=Path, required=True)
    p = sub.add_parser("seal")
    p.add_argument("--image", type=Path, required=True)
    p.add_argument("--metadata", dest="metadata_path", type=Path, required=True)
    p = sub.add_parser("verify")
    p.add_argument("--image", type=Path, required=True)
    p.add_argument("--manifest", type=Path, required=True)
    p = sub.add_parser("archive")
    p.add_argument("--image", type=Path, required=True)
    p.add_argument("--metadata", dest="metadata_path", type=Path, required=True)
    p.add_argument("--output", type=Path)
    p.add_argument("--export", type=Path)
    args = vars(parser.parse_args())
    command = args.pop("command")
    result = {"prepare": prepare, "seal": seal, "verify": verify, "archive": archive_build}[command](**args)
    print(json.dumps(result, ensure_ascii=False))


if __name__ == "__main__":
    main()