flask-gogs-maker/gogsmaker.py

""" GogsMaker

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

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

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'])


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:
            print('Hook {}: failed to make'.format(self.hook['name']),
                  file=sys.stderr)


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'],
                           check=True)  # Just in case.
            subprocess.run(['git', '-C', path, 'pull'], check=True)
        except subprocess.CalledProcessError:
            raise GitError("Cannot pull {}".format(hook['name']))

    else:  # Simply update
        try:
            subprocess.run(['mkdir', '-p', path])
            subprocess.run(['git', 'clone', clone_url, path], check=True)
        except subprocess.CalledProcessError:
            raise GitError("Cannot clone {}".format(clone_url))


def check_signature(received_sig, hook, payload):
    ''' Check Gogs signature '''
    digest = hmac.new(hook.secret,
                      msg=payload,
                      digestmod=sha256).digest()
    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']

    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


# TODO: @app.before_first_request settings checkup