first commit

Author: TCA166

.gitignore

@@ -0,0 +1,2 @@
+.vscode/
+tokens.json

README.md

@@ -0,0 +1,27 @@
+# githubWebhook
+
+A Flask blueprint for receiving GitHub or GitLab webhooks.
+
+## Setup
+
+1. Clone this repo
+2. Install packages from requirements.txt
+3. Register the webhookBlueprint within a Flask app of your choice
+4. Now you just need to setup the webhook on the git side, and you are done
+
+## Default behavior
+
+The default webhookBlueprint requires a git token to be provided for verification.
+For GitHub that would be the secret string that you provide [during creation](https://docs.github.com/en/webhooks/using-webhooks/creating-webhooks#creating-a-repository-webhook) and for GitLab that would be the [secret token](https://docs.gitlab.com/ee/user/project/integrations/webhooks.html#validate-payloads-by-using-a-secret-token).
+webhookBlueprint creates a POST endpoint / which receives webhook requests, verifies them and passes their JSON content to webhookBlueprint.processWebhook().
+webhookBlueprint.processWebhook() by default creates a new git subprocess that performs a git pull.
+After the git pull the default method attempts to run an optionally provided unittest.TestSuite.
+If these tests fail the method attempts to abort the last merge.
+The results of the tests and the abort merge is returned to git.
+For security reasons only webhooks with means of verification will be accepted.
+
+## Customization
+
+The class was built with customization in mind.
+If you need the processing method to do something else I suggest you simply override the processWebhook method with your own method in the singleton blueprint instance.
+Alternatively if you really need to build upon the blueprint, or want a different interface you can always just create a subclass.

requirements.txt

@@ -0,0 +1 @@
+Flask >= 3.0.0

webhook.py

@@ -0,0 +1,54 @@
+from flask import Blueprint, request, Response, abort, Request
+from hashlib import sha256
+from hmac import new as hmacNew
+from subprocess import run
+from unittest import TestSuite, TestResult
+
+def verifyGithubSignature(request: Request, token:str) -> bool:
+    """Verify the GitHub signature of a webhook request"""
+    signature = request.headers.get("X-Hub-Signature-256")
+    if signature is None:
+        return False
+    hash_object = hmacNew(token.encode("utf-8"), msg=request.get_data(), digestmod=sha256)
+    expected_signature = "sha256=" + hash_object.hexdigest()
+    return signature == expected_signature
+
+class webhookBlueprint(Blueprint):
+    """Wrapper over the flask blueprint that creates an endpoint for receiving and processing git webhooks. Overwrite the processWebhook method to process the webhook data."""
+    def __init__(self, webhookToken:str, tests:TestSuite=None, name:str="webhook", import_name:str=__name__, *args, **kwargs):
+        super().__init__(name, import_name, *args, **kwargs)
+        self.webhookToken = webhookToken
+        self.tests = tests
+        self.route("/", methods=["POST"])(self.receiveWebhook)
+    def receiveWebhook(self) -> Response:
+        """Receive webhook from GitHub and process it using the processWebhook method."""
+        if "X-Hub-Signature-256" in request.headers:
+            if not verifyGithubSignature(request, self.webhookToken):
+                abort(401)
+        elif "X-Gitlab-Token":
+            if request.headers.get("X-Gitlab-Token") != self.token:
+                abort(401)
+        else:
+            abort(400, "Unsupported webhook source")
+        #at this point the webhook is verified
+        return self.processWebhook(request.json)
+    def processWebhook(self, data:dict) -> tuple[int, str]:
+        """Process the webhook. Return a tuple of (status code, message)"""
+        process = run(["/usr/bin/git", "pull"], env=dict(GIT_SSH_COMMAND="/usr/bin/ssh"))
+        if process.returncode != 0:
+            return 500, process.stderr.decode("utf-8")
+        if self.tests is not None:
+            result:TestResult = self.tests.run()
+            if result.wasSuccessful():
+                return 200, "Tests passed"
+            else:
+                abortProcess = run(["/usr/bin/git", "merge", "--abort"], env=dict(GIT_SSH_COMMAND="/usr/bin/ssh"))
+                return 428, f"Tests did not pass, Errors: {result.errors}, Failures: {result.failures}. Merge abort status: {abortProcess.returncode}"
+        else:
+            return 200, "Webhook received successfully"
+        
+if __name__ == "__main__":
+    from flask import Flask
+    app = Flask(__name__)
+    app.register_blueprint(webhookBlueprint("token"))
+    app.run()

wsgi.py

@@ -0,0 +1,12 @@
+from webhook import webhookBlueprint
+from flask import Flask
+import json
+
+app = Flask(__name__)
+with open("tokens.json", "r") as f:
+    token = json.load(f)["webhookGit"]
+wb = webhookBlueprint(token, url_prefix="/")
+app.register_blueprint(wb)
+
+if __name__ == "__main__":
+    app.run()