coinbootmaker_helper

#!/usr/bin/env python3

# Copyright (C) 2021 Gunter Miegel coinboot.io
#
# This file is part of Coinboot.
# This software may be modified and distributed under the terms
# of the MIT license.  See the LICENSE file for details.
#
# Please notice even while this script is licensed with the
# MIT license the software packaged by this script may be licensed by
# an other license with different terms and conditions.
# You have to agree to the license of the packaged software to use it.

"""
Usage:
    coinbootmaker_helper create plugin <path> [--upload]
    coinbootmaker_helper create readme
"""

import re
import os
import subprocess
import fileinput
import logging
import boto3
from tabulate import tabulate
from docopt import docopt
from botocore.exceptions import ClientError
from strictyaml import load, Map, Str, Int, Seq, YAMLError


def call_coinbootmaker(script_name):
    """Run plugin creation script with coinbootmaker

    :param script_name: name of the plugin creation script
    :returns: output of running the plugin creation script

    """
    coinbootmaker_output = subprocess.run(
        ["./coinbootmaker", "-p", script_name], stdout=subprocess.PIPE, check=True
    )
    return coinbootmaker_output


def extract_full_archive_name(subprocess_output):
    """Get composed full name of created Coinboot plugin archive

    :param subprocess_output: output of subprocess call
    :returns: filename of created Coinboot plugin archive

    """
    # We look for a line looking like that:
    # 'Created Coinboot Plugin: coinboot_ethminer_v0.18.0_20210603.1555.tar.gz'
    for line in subprocess_output.stdout.decode("utf-8").split("\n"):
        match = re.search("Created Coinboot Plugin: (.*)", line)
        if match:
            fullname_created_archive = match.group(1)
            return fullname_created_archive
    return None


def detect_kernel(subprocess_output):
    """Determine kernel version of coinbootmaker built environment

    :param subprocess_output: output of calling coinbootmaker
    :returns: detected kernel version else None
    """
    for line in subprocess_output.stdout.decode("utf-8").split("\n"):
        match = re.search("lib\/modules\/(.*-generic)\/.*$", line)
        if match:
            kernel = match.group(1)
            return kernel
    return None


def upload_file(
    s3_client, file_name, bucket, coinbootmaker_file, kernel, object_name=None
):
    """Upload a file to an S3 bucket

    :param file_name: file to upload
    :param bucket: destination bucket
    :param object_name: S3 object name. If not specified then file_name is used (Default value = None)
    :param s3_client: S3-client instance created by boto
    :param coinbootmaker_file: dict holding metadata and script of Coinbootmaker file
    :returns: True if file was uploaded, else False

    """
    # If S3 object_name was not specified, use file_name
    if object_name is None:
        object_name = file_name

    if kernel is None:
        kernel = "all"

    # Upload the file
    try:
        response = s3_client.upload_file(
            file_name,
            bucket,
            object_name,
            # Metadata prefix 'x-amz-meta-' is added automatically.
            ExtraArgs={
                "ACL": "public-read",
                "Metadata": {
                    "archive_name": coinbootmaker_file["archive_name"],
                    "description": coinbootmaker_file["description"],
                    "maintainer": coinbootmaker_file["maintainer"],
                    "plugin": coinbootmaker_file["plugin"],
                    "source": coinbootmaker_file["source"],
                    "version": coinbootmaker_file["version"],
                    "kernel": kernel,
                },
            },
        )
    except ClientError as e:
        logging.error(e)
        return False
    return True


def create_markdown_table_of_plugins(s3_client, bucket):
    """Create a list of pluings present at a defined S3 bucket

    :param s3_client: S3-client instance created by boto
    :param bucket: destination bucket

    """

    metadata_keys_sorted = [
        "plugin",
        "version",
        "kernel",
        "description",
        "maintainer",
        "source",
    ]
    header_keys_sorted = [
        metadata_key.capitalize() for metadata_key in metadata_keys_sorted
    ]
    header_keys_sorted.append("URL")

    list_of_plugins = []

    list_of_plugins.append(header_keys_sorted)

    # TODO: Error handling for S3 API call - failing API or objects, metadata missing
    response = s3_client.list_objects(Bucket=bucket)
    if "Contents" in response:
        response_contents = response["Contents"]
        for object in response_contents:
            metadata = s3_client.head_object(Bucket=bucket, Key=object["Key"])[
                "Metadata"
            ]
            line = []

            for key in metadata_keys_sorted:
                line.append(metadata.get(key))
            line.append(
                "https://s3.eu-central-1.wasabisys.com/coinboot/" + object["Key"]
            )
            list_of_plugins.append(line)
        markdown_formatted = tabulate(
            list_of_plugins, headers="firstrow", tablefmt="github"
        )
        return markdown_formatted.split("\n")
    else:
        print("Bucket is empty")
        return None


def concat_with_readme(readme_template_file, readme_file, list_of_plugins):
    """Create a markdown table in the README file with the provided list of plugins

    :param readme_template_file: template file for the README
    :param readme_file: the final README file
    :param list_of_plugins: list of plugins to create a markdown table from

    """
    readme_file_content = []

    for line in fileinput.input(readme_template_file):
        line = line.rstrip("\r\n")
        if line == "<!-- PLACEHOLDER FOR MARKDOWN PLUGIN TABLE -->":
            readme_file_content.extend(list_of_plugins)
        else:
            readme_file_content.append(line)

    with open(readme_file, "w") as f:
        f.write("\n".join(readme_file_content))


def setup_s3_client():
    """Setup the S3 client

    :returns: S3 client object

    """
    s3_client = boto3.client(
        "s3",
        endpoint_url="https://s3.eu-central-1.wasabisys.com",
        aws_access_key_id=os.environ["AWS_ACCESS_KEY_ID"],
        aws_secret_access_key=os.environ["AWS_SECRET_ACCESS_KEY"],
    )
    return s3_client


def main():
    """ """
    args = docopt(__doc__)
    if args["plugin"]:
        path = args["<path>"].replace("plugins/", "")
        schema = Map(
            {
                "plugin": Str(),
                "archive_name": Str(),
                "version": Str(),
                "description": Str(),
                "maintainer": Str(),
                "source": Str(),
                "run": Str(),
            }
        )

        with open(path, "r") as f:
            try:
                coinbootmaker_file = load(f.read(), schema)
                script_name = path.replace("src", "")
                coinbootmaker_output = call_coinbootmaker(script_name)
                archive_name = extract_full_archive_name(coinbootmaker_output)
                print(archive_name)
            except YAMLError as error:
                print(path + ": " + str(error))

            if args["--upload"]:
                s3_client = setup_s3_client()
                kernel = detect_kernel(coinbootmaker_output)
                if kernel:
                    prefix = kernel
                else:
                    prefix = "all"

                print("Bucket prefix is: " + prefix)

                upload_file(
                    s3_client,
                    "build/" + archive_name,
                    "coinboot",
                    coinbootmaker_file.data,
                    kernel,
                    prefix + "/" + archive_name,
                )

    if args["readme"]:
        s3_client = setup_s3_client()
        table_of_plugins = create_markdown_table_of_plugins(s3_client, "coinboot")
        if table_of_plugins:
            concat_with_readme("README_template.md", "README.md", table_of_plugins)


if __name__ == "__main__":
    main()