fuchsia_upload.py

#!/usr/bin/env python3
"""Publishes Fuchsia VIM3 firmware source code.

This has three parts:
  1. build firmware (split into fuchsia_build.py)
  2. create YAML files for CIPD package
  3. publish source code to the open-source repo

We could potentially just develop in the open - there's nothing internal in this
codebase - except that we don't really have a good host at the moment for
developing GPL code publicly (http://b/213950490). So for now we just manually
flush code out on every release so that the CIPD binaries match the public
source code.

In case CIPD is not available it is expected that all necessary tools would be
in PATH.
"""

import argparse
import json
import logging
import os
import shutil
import subprocess
import sys
import tempfile
import textwrap
from pathlib import Path
from typing import Optional
from fuchsia_build import fuchsia_build, ensure_deps, cipd_dep_dir_clean

_MY_DIR = Path(__file__).parent.resolve()

_REPO_NAME = "u-boot"
_UPSTREAM_BRANCH = "origin/vim3"
_THIRD_PARTY_REPO_URL = f"https://third-party-mirror.googlesource.com/{_REPO_NAME}/"


_LICENSE_SEPARATOR = "\n===================================\n\n"
# Explicitly enumerate known license files so that if a new one comes in,
# the script will fail and notify us. Once the new file has been reviewed and
# approved we can add it here.
_KNOWN_LICENSE_FILES = (
    "bsd-2-clause.txt",
    "bsd-3-clause.txt",
    "eCos-2.0.txt",
    "gpl-2.0.txt",
    "ibm-pibs.txt",
    "isc.txt",
    "lgpl-2.0.txt",
    "lgpl-2.1.txt",
)
# Files in the licenses dir that are informational only and don't need to be
# included in the license dump.
_INFORMATIONAL_LICENSE_FILES = ("Exceptions", "README")

# u-boot.bin.unsigned is the combined BL2 + TPL format that goes in the eMMC
# boot0/boot1 partitions. Unclear why u-boot.bin.signed.encrypted doesn't work,
# but we don't need signatures for dev boards so we're OK with this.
_BUILD_ARTIFACT_NAME = "u-boot.bin.unsigned"
_BUILD_ARTIFACT_PATH = _MY_DIR / "build" / _BUILD_ARTIFACT_NAME

_CIPD_FILES_DIR = _MY_DIR / "cipd_out"
_CIPD_YAML_FILE_NAME = _CIPD_FILES_DIR / "cipd.yaml"
_CIPD_PACKAGE = "fuchsia/prebuilt/third_party/firmware/vim3"

_JAVABIN = "java"
_COPYBARA_PATH = "copybara"


def setup_gob_auth() -> bool:
    """Attempts to set up Git authentication for GoB using LUCI credential helper.

    This configures the local Git repository to use the 'luci' credential helper
    (git-credential-luci) for GoB hosts (*.googlesource.com).

    The credentials (OAuth tokens) are managed by the ambient LUCI/Swarming bot
    environment. The helper retrieves these tokens dynamically from the local
    LUCI authentication service (communicating via the socket/file pointed to
    by the LUCI_CONTEXT environment variable).

    This setup assumes that:
    1. git-credential-luci is available on PATH (typically part of depot_tools).
    2. The script is running within an active LUCI task context that has been
       configured with a service account (handled automatically by the LUCI
       builder configuration). No manual key registration is required.

    Returns:
        True if authentication was successfully set up, False otherwise.
    """
    print("Attempting to set up GoB authentication using LUCI credential helper...")
    os.environ["GIT_TERMINAL_PROMPT"] = "0"
    credential_helper_str = "credential.https://*.googlesource.com.helper"

    # Check if git-credential-luci is available on PATH
    if not shutil.which("git-credential-luci"):
        print(
            "git-credential-luci helper not found on PATH. Cannot configure GoB auth."
        )
        return False

    try:
        # Clear any existing helper for this pattern to avoid conflicts/multi-value errors
        subprocess.run(
            ["git", "config", "--local", "--unset-all", credential_helper_str],
            check=False,
        )

        # Configure git to use the 'luci' helper for googlesource.com
        subprocess.run(
            ["git", "config", "--local", credential_helper_str, "luci"], check=True
        )
        print(
            "Successfully configured Git to use LUCI credential helper for googlesource.com"
        )

        # Verify auth
        print("Verifying authentication by running git ls-remote...")
        # Note: We must NOT bypass credential helper here because we want to test it.
        try:
            res = subprocess.run(
                ["git", "ls-remote", _THIRD_PARTY_REPO_URL],
                capture_output=True,
                text=True,
                timeout=10,
            )
            if res.returncode == 0:
                print("Auth verification SUCCESS")
                return True
            else:
                print(f"Auth verification FAILED: {res.stderr}")
                return False
        except subprocess.TimeoutExpired:
            print("Auth verification TIMEOUT (10s). git ls-remote hung.")
            return False

    except Exception as e:
        print(f"Error configuring GoB auth: {e}")
        return False


def license_file_contents() -> str:
    """Creates and returns the combined license file contents.

    Raises:
        FileExistsError if an unexpected license file was found.
    """
    licenses = []
    license_dir = _MY_DIR / "Licenses"

    def _add_license(name: str):
        contents = (license_dir / name).read_text()
        # No need to duplicate licenses if they're the exact same.
        if contents not in licenses:
            licenses.append(contents)

    dir_contents = sorted(item.name for item in license_dir.iterdir() if item.is_file())
    for name in dir_contents:
        if name in _KNOWN_LICENSE_FILES:
            _add_license(name)
        elif name in _INFORMATIONAL_LICENSE_FILES:
            pass
        else:
            raise FileExistsError(f"Unknown u-boot license file: {name}")

    return _LICENSE_SEPARATOR.join(licenses)


def ensure_git_is_clean(repo_dir: Path, upstream_branch: str) -> str:
    """Checks if the given git repo is clean.

    A clean repo is one that is on the upstream branch and has no local changes.

    Args:
        repo_dir: a path to the repo root dir.
        upstream_branch: upstream branch we should match.

    Raises:
        ValueError if the given repo is not clean.

    Returns:
        The current clean git revision.
    """

    def git(command):
        return subprocess.run(
            ["git"] + command, cwd=repo_dir, check=True, capture_output=True, text=True
        ).stdout.strip()

    # Update the local repo.
    git(["fetch"])

    # Check the HEAD revision to look for committed changes.
    head_revision = git(["rev-parse", "HEAD"])
    upstream_revision = git(["rev-parse", upstream_branch])
    if head_revision != upstream_revision:
        raise ValueError(f"HEAD {head_revision} != upstream {upstream_revision}")

    # Check for any uncommitted file changes. This will print the name of
    # any uncommitted files, so empty string means no local changes.
    status = git(["status", "--porcelain"])
    if status:
        raise ValueError(f"Uncommitted local changes:\n{status}")

    return head_revision


def create_copybara_config(revision: Optional[str]):
    """Creates the Copybara config file contents.

    Args:
        revision: u-boot source revision to publish, or None for ToT.

    Returns:
        The Copybara config file contents.
    """
    # When dry-running with local changes, our local revision doesn't exist on
    # the host so copybara can't find it. In this case just use the tip of tree.
    if not revision:
        logging.warning("No source revision given, using 'vim3' ToT")
        revision = "vim3"

    # http://go/copybara-reference.
    contents = textwrap.dedent(
        f"""\
        core.workflow(
            name = "default",
            origin = git.origin(
                url = "https://turquoise-internal.googlesource.com/third_party/u-boot",
                ref = "{revision}",
            ),
            destination = git.destination(
                url = "{_THIRD_PARTY_REPO_URL}",
                push = "vim3",
            ),
            mode = "ITERATIVE",
            authoring = authoring.pass_thru(
                default = "Fuchsia firmware team <fuchsia-firmware@google.com>"
            ),
        )
        """
    )

    return contents


def create_cipd_files(manifest: str, revision: Optional[str]):
    """Create CIPD yaml files

    Args
        manifest: filename for YAML manifest file provided by CIPD builder
        revision: u-boot source revision, or None for ToT
    """
    shutil.rmtree(_CIPD_FILES_DIR, ignore_errors=True)
    _CIPD_FILES_DIR.mkdir(parents=True, exist_ok=True)

    # Source revision is only used in metadata for CIPD so it doesn't have
    # to be a real ref.
    if not revision:
        revision = "__local_dirty__"

    metadata_files = {
        "OWNERS": "fuchsia-firmware@google.com",
        "manifest.json": json.dumps({_REPO_NAME: revision}, indent=2, sort_keys=True),
        "LICENSE": license_file_contents(),
    }

    for name, contents in metadata_files.items():
        (_CIPD_FILES_DIR / name).write_text(contents)

    shutil.copyfile(_BUILD_ARTIFACT_PATH, _CIPD_FILES_DIR / _BUILD_ARTIFACT_NAME)

    # Create CIPD yaml file
    with open(_CIPD_YAML_FILE_NAME, "w") as cipd_yaml_file:
        logging.info("Writing CIPD yaml file: `%s`", _CIPD_YAML_FILE_NAME)
        cipd_yaml_file.write(
            textwrap.dedent(
                f"""\
                package: {_CIPD_PACKAGE}
                description: vim3 firmware image
                install_mode: copy
                data:
            """
            )
        )

        cipd_yaml_file.write(f"  - file: {_BUILD_ARTIFACT_NAME}\n")
        for name in metadata_files:
            cipd_yaml_file.write(f"  - file: {name}\n")

    create_cipd_yaml_manifest(revision, manifest)


def create_cipd_yaml_manifest(revision: Optional[str], filename: str):
    """Creates YAML manifest for CIPD.

    Args:
        revision: u-boot source revision to publish, or None for ToT.
        filename: name for the file to write to.
    """

    # When dry-running with local changes, our local revision doesn't exist on
    # the host. In this case just use the tip of tree.
    if not revision:
        logging.warning("No source revision given, using 'vim3' ToT")
        revision = "vim3"

    contents = json.dumps(
        [
            {
                "path": str(_CIPD_YAML_FILE_NAME.relative_to(_MY_DIR)),
                "tags": {_REPO_NAME: revision},
            }
        ],
        indent=4,
        sort_keys=True,
    )

    with open(filename, "w") as yaml_manifest_file:
        logging.info("Writing YAML file: `%s`", filename)
        yaml_manifest_file.write(contents)


def publish_source(
    revision: Optional[str],
    copybara_path: str,
    last_rev: Optional[str],
    dry_run: bool,
    push_justification: str,
):
    """Publishes this source to the public repo.

    Args:
        revision: u-boot source revision to publish, or None for ToT.
        copybara_path: path to the copybara executable.
        last_rev: last source revision that was pushed to destination. Use this
                  only to initialize the repo the first time so we can tell
                  Copybara where to start.
        dry_run: True to skip submitting anything.
        push_justification: BugID for push justification.

    Raises:
        subprocess.CalledProcessError if copybara fails.
    """
    config = create_copybara_config(revision)

    config_path = Path(tempfile.mkdtemp()) / "copy.bara.sky"
    config_path.write_text(config)

    # CIPD binary version has `copybara.jar` that depends on build time Java environment.
    # That is not available in builders. And `copybara_deploy.jar` that includes all dependencies.
    # In order to use second one '--singlejar' option is required
    command = [
        copybara_path,
        "--singlejar",
        str(config_path),
        "--git-push-option",
        "push-justification=" + push_justification,
    ]
    if last_rev:
        command += ["--last-rev", last_rev]
    if dry_run:
        command.append("--dry-run")
    logging.info("Copybara command: `%s`", " ".join(command))

    # Let stdout/sterr through to the console since this may take a while.
    try:
        subprocess.run(command, env=dict(os.environ, JAVABIN=_JAVABIN), check=True)
    except subprocess.CalledProcessError as error:
        if error.returncode == 4:  # NO_OP
            logging.info("Copybara reported nothing to do")
        else:
            raise


def _parse_args() -> argparse.Namespace:
    parser = argparse.ArgumentParser(
        description=__doc__, formatter_class=argparse.RawDescriptionHelpFormatter
    )

    parser.add_argument(
        "--cipd", default="cipd", help="Path to CIPD tool; looks on $PATH by default"
    )
    parser.add_argument(
        "--copybara",
        default=_COPYBARA_PATH,
        help=f"Path to copybara tool; default {_COPYBARA_PATH}",
    )
    parser.add_argument("--dry-run", action="store_true", help="Don't upload anything")
    parser.add_argument(
        "--cipd-yaml-manifest",
        default="",
        help="Write YAML manifest for CIPD in specified file",
    )
    parser.add_argument(
        "--last-rev",
        help="The last source revision that was released."
        " Only use this on the first run to tell copybara where to start.",
    )
    parser.add_argument(
        "--push-justification",
        default="b/302031093",
        help="BugID for git push justification.",
    )
    parser.add_argument(
        "--clean-cipd",
        action="store_true",
        help="Clear CIPD dependencies dir before start",
    )

    return parser.parse_args()


def is_on_builder() -> bool:
    """Checks if the script is running on a LUCI/Swarming builder."""
    return "SWARMING_TASK_ID" in os.environ or "BUILDBUCKET_BUILDER" in os.environ


def _main() -> int:
    logging.basicConfig(level=logging.INFO)
    args = _parse_args()

    if is_on_builder():
        setup_gob_auth()

    if args.clean_cipd:
        cipd_dep_dir_clean()

    # Versions are pinned to current integration version in most cases.
    # It is done to prevent unexpected breaking when integration moves forward.
    # It is required to do manual testing before bumping the versions.
    list_of_packages = [
        (
            "copybara",
            "infra/3pp/tools/copybara",
            "version:3@2450737ba36adafb13b5794da8948edfcb78afe3.cr1",
            "copybara",
        ),
        ("java", "flutter/java/openjdk/linux-amd64", "version:21", "bin"),
    ]
    ensure_deps(args.cipd, list_of_packages)

    try:
        revision = ensure_git_is_clean(_MY_DIR, _UPSTREAM_BRANCH)
    except ValueError as exception:
        if args.dry_run:
            logging.warning("%s", exception)
            logging.warning("Ignoring dirty local repo for dry run")
            revision = None
        else:
            raise

    publish_source(
        revision,
        args.copybara,
        args.last_rev,
        args.dry_run,
        args.push_justification,
    )

    try:
        fuchsia_build(
            cipd=args.cipd,
            clean_cipd=False,
        )
    except subprocess.CalledProcessError as e:
        logging.error(f"Build failed with exit code {e.returncode}")
        return e.returncode

    if args.cipd_yaml_manifest:
        create_cipd_files(manifest=args.cipd_yaml_manifest, revision=revision)

    return 0


if __name__ == "__main__":
    sys.exit(_main())