163 lines
4.9 KiB
Python
163 lines
4.9 KiB
Python
""" 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 termcolor import colored
|
|
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'].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']
|
|
|
|
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 check_settings():
|
|
''' Check the supplied settings '''
|
|
if settings.DEBUG:
|
|
print(colored('WARNING! ', 'red', attrs=['bold'])
|
|
+ '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)
|
|
|
|
print((colored('FATAL! ', 'red', attrs=['bold'])
|
|
+ 'Configuration error: hook {} lacks attribute {}.')
|
|
.format(descr, key))
|
|
sys.exit(1)
|