flask-gogs-maker/gogsmaker.py

""" GogsMaker

A webhook-handler for Gogs running `make` when needed. """

import os
import sys
import subprocess
import hmac
import logging
from hashlib import sha256
from threading import Thread
from functools import wraps
import coloredlogs
from flask import Flask, request
import settings

LOGGER_NAME = __name__

app = Flask(__name__)


class UnmonitoredRepository(Exception):
    pass


class GitError(Exception):
    def __init__(self, what):
        super().__init__()
        self.what = what

    def __str__(self):
        return self.what


def get_hook(url):
    """ Get the hook matching an URL, or raise UnmonitoredRepository """
    for hook in settings.HOOKS:
        if hook["url"] == url:
            return hook
    raise UnmonitoredRepository


def repo_path(hook):
    """ Get the path at which the hook's repo is cloned """
    return os.path.join(settings.CLONE_ROOT, hook["name"])


def subprocess_run(command, **kwargs):
    """ Run subprocess with default arguments """
    args = {
        "check": True,
        "stdout": subprocess.DEVNULL,
        "stderr": subprocess.PIPE,
    }
    args.update(kwargs)

    return subprocess.run(command, **args)


class MakeWorker(Thread):
    """ A make job """

    def __init__(self, hook):
        super().__init__()
        self.hook = hook
        self.name = "makeworker-{}".format(hook["name"])
        self.path = repo_path(hook)

    def run(self):
        """ Run the make job """
        try:
            subprocess_run(["make", "-C", self.path, "--"] + self.hook["targets"])
        except subprocess.CalledProcessError as error:
            logging.error(
                ("Hook %s: make failed with status %s. " "Error output:\n%s\n"),
                self.hook["name"],
                error.returncode,
                error.stderr.decode("utf-8"),
            )


def update_repo(hook, clone_url):
    """ Update (or clone) the given repository. May raise GitError. """
    path = repo_path(hook)
    if os.path.isdir(os.path.join(path, ".git")):  # Repo is already cloned
        try:
            subprocess_run(["git", "-C", path, "reset", "--hard"])  # Just in case.
            subprocess_run(["git", "-C", path, "pull"])
        except subprocess.CalledProcessError as error:
            logging.error(
                ("Hook %s: git failed with status %s. " "Error output:\n%s\n"),
                hook["name"],
                error.returncode,
                error.stderr.decode("utf-8"),
            )
            raise GitError("Cannot pull {}".format(hook["name"]))

    else:  # Repo is to be cloned
        try:
            subprocess_run(["mkdir", "-p", path])
            subprocess_run(["git", "clone", clone_url, path], check=True)
        except subprocess.CalledProcessError as error:
            logging.error(
                ("Hook %s: git failed cloning with status %s. " "Error output:\n%s"),
                hook["name"],
                error.returncode,
                error.stderr.decode("utf-8"),
            )
            raise GitError("Cannot clone {}".format(clone_url))


def check_signature(received_sig, hook, payload):
    """ Check Gogs signature """
    digest = hmac.new(
        hook["secret"].encode("utf-8"), msg=payload, digestmod=sha256
    ).hexdigest()
    return hmac.compare_digest(digest, received_sig)


def gogs_payload(required):
    def wrapper(fct):
        @wraps(fct)
        def wrapped(*args, **kwargs):
            payload = request.json
            if payload is None:
                return "Expected json\n", 415

            for field in required + ["repository/html_url"]:
                path = field.split("/")
                explore = payload
                for section in path:
                    if section not in explore:
                        return (
                            "Invalid json: missing {}\n".format("/".join(path)),
                            400,
                        )
                    explore = explore[section]

            try:
                hook = get_hook(payload["repository"]["html_url"])
            except UnmonitoredRepository:
                return "Unmonitored repository\n", 403

            if not settings.DEBUG:
                received_sig = request.headers["X-Gogs-Signature"]
                payload_raw = request.data
                if not check_signature(received_sig, hook, payload_raw):
                    return "Invaild signature\n", 403

            return fct(payload, hook, *args, **kwargs)

        return wrapped

    return wrapper


@app.route("/", methods=["POST"])
@gogs_payload(["repository/clone_url"])
def view_root(payload, hook):
    clone_url = payload["repository"]["clone_url"]
    if "clone_url" in hook:
        clone_url = hook["clone_url"]

    try:
        update_repo(hook, clone_url)
    except GitError as error:
        return "Git error: {}\n".format(error), 500

    worker = MakeWorker(hook)
    worker.start()

    return "OK\n", 200


@app.before_first_request  # FIXME this should be run on startup...
def startup_actions():
    setup_logger()
    check_settings()


def setup_logger():
    """ Setup the default logger """
    coloredlogs.install(
        fmt="%(asctime)s [%(levelname)s] %(message)s",
    )


def check_settings():
    """ Check the supplied settings """
    if settings.DEBUG:
        logging.warning(
            "GogsMaker is running in DEBUG MODE, this is "
            "unsuitable for production environments!"
        )

    required_keys = ["name", "url", "targets", "secret"]
    for hook_id, hook in enumerate(settings.HOOKS):
        for key in required_keys:
            if key not in hook:
                if key == "name":
                    descr = "#{}".format(hook_id)
                else:
                    descr = "{} (#{})".format(hook["name"], hook_id)

                logging.critical(
                    ("Configuration error: hook %s lacks " "attribute %s."), descr, key
                )
                sys.exit(1)