converter.py

#!/usr/bin/python3

import os
import yaml
import requests
import sys
import re
import logging
import argparse
import datetime

DEFAULT_EXPIRATION_DATE = datetime.datetime(2050, 1, 1).date()

def get_args():
    parser = argparse.ArgumentParser("GitHub Actions Approved Patterns Converter")
    parser.add_argument('--ghtoken', required=True, help="GitHub Token")
    parser.add_argument('--dhtoken', required=True, help="DockerHub Token")
    parser.add_argument('-f', '--filename', default="approved_patterns.yml", help="DockerHub Token")
    parser.add_argument('-v','--verbose', action='count', default=0, help="Verbosity")
    args = parser.parse_args()
    return args

class Log:
    def __init__(self, config):
        self.config = config
        self.log = logging.getLogger(__name__)
        self.verbosity = {
            0: logging.INFO,
            1: logging.CRITICAL,
            2: logging.ERROR,
            3: logging.WARNING,
            4: logging.INFO,
            5: logging.DEBUG,
        }

        self.stdout_fmt = logging.Formatter(
            "{asctime} [{levelname}] {funcName}: {message}", style="{"
        )

        if self.config["logfile"] == "stdout":
            self.to_stdout = logging.StreamHandler(sys.stdout)
            self.to_stdout.setLevel(self.verbosity[self.config["verbosity"]])
            self.to_stdout.setFormatter(self.stdout_fmt)
            self.log.setLevel(self.verbosity[self.config["verbosity"]])
            self.log.addHandler(self.to_stdout)
        else:
            self.log.setLevel(self.verbosity[self.config["verbosity"]])
            logging.basicConfig(
                format="%(asctime)s [%(levelname)s] %(funcName)s: %(message)s",
                filename=self.config["logfile"],
            )


class Converter:
    # Handles Converting the allowed_patterns list to an actions.yml that can be consumed by our workflow
    def __init__(self, args):
        self.allowlist = {}
        self.logger = Log(
            {
                "logfile": "stdout",
                "verbosity": args.verbose,
            }
        )

        # GitHub Session Handler
        self.gh = requests.Session()
        self.gh.headers.update(
            {
                "Accept": "application/vnd.github+json",
                "Authorization": f"Bearer {args.ghtoken}",
                "X-GitHub-Api-Version": "2022-11-28",
            }
        )

        # DockerHub Session Handler
        self.dh = requests.Session()
        self.dh.headers.update(
            {
                "Authorization": f"Bearer {args.dhtoken}",
                "Accept": "application/json",
            }
        )

    def gh_fetch(self, uri):
        self.logger.log.debug(f"Fetching {uri}...")
        try:
            data = self.gh.get(uri)
            self.logger.log.debug(data)
            data.raise_for_status()

        except:
            self.logger.log.error(f"{uri} Returned 404!")
            return None

        data = yaml.safe_load(data.content.decode("utf-8"))
        if isinstance(data, list):
            return data
        else:
            print(data["status"])

    def dh_fetch(self, uri):
        self.logger.log.debug(f"Fetching {uri}...")
        try:
            data = self.dh.get(uri)
            data.raise_for_status()
        except:
            self.logger.log.error(f"{uri} Failed!!!")
            return None

        data = yaml.safe_load(data.content.decode("utf-8"))["results"]
        if isinstance(data, list):
            return data
        else:
            print(data)

    def build_dh_action(self, action, tag):
        self.logger.log.info(f"Fetching tags for Dockerhub://{action}")
        dh_uri = f"https://registry.hub.docker.com/v2/repositories/"
        tags = self.dh_fetch(f"{dh_uri}/{action}/tags/?page_size=100")
        t = {}
        if "*" in tag:
            # set to the newest tagged image
            self.logger.log.info("Pinning DockerHub Image to newest Tagged Image")
            sha = max(tags, key=lambda x: x["name"])["digest"]
        else:
            self.logger.log.info(f"Ensuring {tag} is a valid tag")
            tl = [t["sha"] for t in tags if t["name"] == tag]
            if len(tl) == 0:
                sha = None
                self.logger.log.error(
                    f"Tag: {tag} is not found in {action} tags! Skipping..."
                )
            else:
                sha = tl[0]

        if sha:
            t[sha] = {"expires_at": DEFAULT_EXPIRATION_DATE}
        else:
            return None

        return t

    def build_gh_action(self, action, tag):
        self.logger.log.info(f"Fetching Details on {action}")

        gh_uri = f"https://api.github.com/repos/{action}"
        tags = self.gh_fetch(f"{gh_uri}/git/refs/tags")
        heads = self.gh_fetch(f"{gh_uri}/git/refs/heads")

        if tags and heads:
            nick = None
            t = {}
            self.logger.log.info(f"Parsing: {action}@{tag}")
            if "*" in tag:
                if len(tag.split()) > 1:
                    self.logger.log.info("Keeping globs around for now...")
                    sha = tag
                # We need to keep the globs, don't set to HEAD
                else:
                    # if globbed, set to hash of HEAD.
                    # self.logger.log.info("Pinning to the SHA of the current HEAD")
                    self.logger.log.info("Keeping globs around for now...")
                    #sha = max(tags, key=lambda x: x["ref"])["object"]["sha"]
                    sha = "*"
            elif tag == "latest":
                # set to the hash of 'refs/heads/latest'
                self.logger.log.info("Pinning to the SHA of refs/heads/latest")
                sha = [item for item in heads if item["ref"] == "refs/heads/latest"][0][
                    "object"
                ]["sha"]

            elif len(tag) == 40:
                # Lets pretend for now that any 40 character string is a SHA
                # TODO Validate that the 40 character string is a valid SHA
                self.logger.log.critical(
                    "Pretending that any 40 character string is a SHA..."
                )
                sha = tag

            else:
                # Check if the provided tag is valid, if so use it.
                nick = tag
                self.logger.log.info(f"Pinning to the SHA of refs/heads/{tag}")
                sha = next(
                    (
                        item["object"]["sha"]
                        for item in tags
                        if item["ref"] == f"refs/tags/{tag}"
                    ),
                    None,
                )
                if sha is None:
                    self.logger.log.error(
                        f"Tag: {tag} not found in https://api.github.com/repos/{action}/git/refs/tags"
                    )
                    self.logger.log.error(f"Skipping {action}@{tag}")
                    return None

            # Expiration Date set as a GLOBAL
            t[sha] = {"expires_at": f"{DEFAULT_EXPIRATION_DATE}"}
            
            # TODO Don't keep globs
            # Keep the globs for 6 months
            if sha == "*":
                t[sha]["keep"] = True
                t[sha]["expires_at"] = DEFAULT_EXPIRATION_DATE

            # TODO Don't keep tags
            # Keep tags as nicknames for their associated version for 6 months
            if nick:
                t[nick] = {"expires_at": DEFAULT_EXPIRATION_DATE}

            return t

    def parse_approved_patterns(self, file):
        allowlist = {}
        allowed = yaml.safe_load(file)
        for ap in allowed:
            # Do some work to make sure that the names are right first, _then_ parse the tags
            a = ap.split("/")

            # Parse Docker things first
            if a[0] == "docker:":
                # action = self.build_dh_action(ap)
                self.logger.log.critical("Parsing DockerHub entry")
                dkey, image, tag = ap.split(":")
                act = image.lstrip("//")
                action = self.build_dh_action(act, tag)

                # reset the action name to include the docker key `docker://`
                act = "://".join([dkey, act])

            # If it's not Docker it's GitHub
            else:
                # %s/%s
                if len(a) == 2:
                    org = a[0]
                    if a[1] != "*":
                        repo = a[1]
                    else:
                        self.logger.log.critical(
                            f"Invalid Entry (No repo provided): {ap}"
                        )
                        continue
                # %s, should not happen
                elif len(a) == 1:
                    print(a)

                # %s/%s/%s trunc'd to %s/%s
                elif len(a) >= 3:
                    org = a[0]
                    repo = a[1]

                act = f"{org}/{repo}"

                if "@" in act:
                    act, tag = act.split("@")
                else:
                    # In this case * is equivalent to HEAD of the default branch
                    tag = "*"

                action = self.build_gh_action(act, tag)

            if action:
                # Update the allowlist
                if act in self.allowlist:
                    allowlist[act].update(action)
                else:
                    allowlist[act] = action

        return allowlist


if __name__ == "__main__":
    args = get_args()
    c = Converter(args)
    c.logger.log.info("Parsing {FILENAME}")
    converted = c.parse_approved_patterns(open(args.filename))
    c.logger.log.info("Printing Generated actions.yml to file")
    with open("actions.yaml", "w+") as f:
        yaml.safe_dump(converted, f, default_flow_style=False)
        f.close()
    c.logger.log.info("Done!")