Implement fifo hooks

.gitignore

@@ -1,3 +1,5 @@
+config.yml
+
 # ---> Python
 # Byte-compiled / optimized / DLL files
 __pycache__/

config.sample.yml

@@ -0,0 +1,4 @@
+fifo_hooks:
+  test_repo:
+    secret: very_secret  # same as given to gitea
+    fifo_path: /var/run/test_repo

gitea_hooks/app.py

@@ -0,0 +1,68 @@
+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/<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")
+    except FileNotFoundError:
+        return "No such fifo", 500
+    except PermissionError:
+        return "Permission denied on FIFO", 500
+
+    return "OK", 200

gitea_hooks/conf.py

@@ -0,0 +1,33 @@
+import yaml
+
+
+class BadConfig(Exception):
+    """Raised when the configuration is incorrect"""
+
+
+class Configuration:
+    """Parses the configuration file, and caches it.
+
+    The configuration is a YAML file.
+    """
+
+    def __init__(self):
+        try:
+            with open("./config.yml", "r") as handle:
+                self.raw = yaml.safe_load(handle)
+        except yaml.YAMLError as exn:
+            raise BadConfig("Cannot parse yaml") from exn
+        except FileNotFoundError as exn:
+            raise BadConfig("Configuration file not found") from exn
+        except PermissionError as exn:
+            raise BadConfig("Configuration file not readable") from exn
+        self._healthcheck()
+
+    def _healthcheck(self):
+        """Checks that the configuration is ok"""
+        if "fifo_hooks" not in self.raw:
+            raise BadConfig("No `fifo_hooks`")
+        for hook, data in self.raw["fifo_hooks"].items():
+            for mandatory in ("secret", "fifo_path"):
+                if mandatory not in data:
+                    raise BadConfig("Fifo hook {}: no `{}`".format(hook, mandatory))

requirements.txt

@@ -0,0 +1,2 @@
+Flask==2.1.*
+pyyaml