From c86a4bb1b9f337e7b923e46eef24e1967684313b Mon Sep 17 00:00:00 2001
From: =?UTF-8?q?Th=C3=A9ophile=20Bastian?= <contact@tobast.fr>
Date: Fri, 15 Apr 2022 18:31:39 +0200
Subject: [PATCH] Implement fifo hooks

---
 config.sample.yml   |  4 +++
 gitea_hooks/app.py  | 68 +++++++++++++++++++++++++++++++++++++++++++++
 gitea_hooks/conf.py | 33 ++++++++++++++++++++++
 3 files changed, 105 insertions(+)
 create mode 100644 config.sample.yml
 create mode 100644 gitea_hooks/app.py
 create mode 100644 gitea_hooks/conf.py

diff --git a/config.sample.yml b/config.sample.yml
new file mode 100644
index 0000000..56e65bc
--- /dev/null
+++ b/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
diff --git a/gitea_hooks/app.py b/gitea_hooks/app.py
new file mode 100644
index 0000000..e68040f
--- /dev/null
+++ b/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
diff --git a/gitea_hooks/conf.py b/gitea_hooks/conf.py
new file mode 100644
index 0000000..630a6e5
--- /dev/null
+++ b/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))