import typing as t import hmac import hashlib import json from collections import defaultdict from flask import Flask, request, abort from .conf import Configuration config = Configuration() app = Flask(__name__) def webhook_receiver(hook_type): if hook_type not in config.raw: raise Exception(f"Badly configured route: no conf of type {hook_type}") relevant_hooks = config.raw[hook_type] def inner(func): def wrapped(hook_name: str): if request.content_length is None or request.content_length > 32000: abort(400) if request.content_type != "application/json": return "Expected json", 415 if hook_name not in relevant_hooks: abort(404) hook_conf = relevant_hooks[hook_name] raw_payload: bytes = request.get_data(cache=False) provided_sig: str = request.headers["X-Gitea-Signature"] computed_sig = hmac.new( hook_conf["secret"].encode("utf-8"), raw_payload, hashlib.sha256 ).hexdigest() if not hmac.compare_digest(provided_sig, computed_sig): abort(403) try: payload = json.loads(raw_payload) except json.JSONDecodeError: return "Bad JSON", 400 return func(payload, hook_name, hook_conf) return wrapped return inner @app.route("/") def root() -> t.Tuple[str, int]: """Root web handler -- pointless in this case""" return "Not supported.", 400 @app.route("/fifo/", methods=["POST"]) @webhook_receiver("fifo_hooks") def fifo_hooks(payload, hook_name, hook_conf): """Fifo web handler -- write 1 to a unix fifo""" try: with open(hook_conf["fifo_path"], "w") as fifo: fifo.write("1") except FileNotFoundError: return "No such fifo", 500 except PermissionError: return "Permission denied on FIFO", 500 return "OK", 200