69 lines
2.0 KiB
Python
69 lines
2.0 KiB
Python
import typing as t
|
|
import hmac
|
|
import hashlib
|
|
import json
|
|
from collections import defaultdict
|
|
from flask import Flask, request
|
|
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:
|
|
return "Too much content", 400
|
|
if request.content_type != "application/json":
|
|
return "Expected json", 415
|
|
|
|
if hook_name not in relevant_hooks:
|
|
return "No such hook", 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):
|
|
return "Bad credentials", 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/<string:hook_name>", 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\n")
|
|
except FileNotFoundError:
|
|
return "No such fifo", 500
|
|
except PermissionError:
|
|
return "Permission denied on FIFO", 500
|
|
|
|
return "OK", 200
|