Added tests

Author: TCA166

pyproject.toml

@@ -19,3 +19,6 @@ dependencies = [
 [project.urls]
 Homepage = "https://github.com/TCA166/gitWebhook"
 
+[options]
+package_dir = src
+

src/webhook.py

@@ -9,7 +9,7 @@
 
 GITLAB_HEADER = "X-Gitlab-Token"
 
-def verifyGithubSignature(request: Request, token:str) -> bool:
+def verifyGithubRequest(request: Request, token:str) -> bool:
     """Verify the GitHub signature of a webhook request"""
     signature = request.headers.get(GITHUB_HEADER)
     if signature is None:
@@ -18,6 +18,10 @@ def verifyGithubSignature(request: Request, token:str) -> bool:
     expected_signature = "sha256=" + hash_object.hexdigest()
     return signature == expected_signature
 
+def verifyGitlabRequest(request: Request, token:str) -> bool:
+    """Verify the GitLab token of a webhook request"""
+    return request.headers.get(GITLAB_HEADER) == token
+
 class webhookBlueprint(Blueprint, gitWebhookBlueprintABC):
     """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 | None, log:Logger | None = None, name:str="webhook", *args, **kwargs):
@@ -39,12 +43,12 @@ def receiveWebhook(self) -> Response:
             abort(415)
         if self.webhookToken is not None: #verification
             if GITHUB_HEADER in request.headers:
-                if not verifyGithubSignature(request, self.webhookToken):
+                if not verifyGithubRequest(request, self.webhookToken):
                     if self.log is not None:
                         self.log.warning("A request with an invalid GitHub signature")
                     abort(401)
             elif GITLAB_HEADER:
-                if request.headers.get(GITLAB_HEADER) != self.webhookToken:
+                if not verifyGitlabRequest(request, self.webhookToken):
                     if self.log is not None:
                         self.log.warning("A request with an invalid GitLab token")
                     abort(401)

test.py

@@ -0,0 +1,96 @@
+import unittest
+from flask import Request, request, Flask
+from src.webhook import verifyGithubRequest, verifyGitlabRequest, webhookBlueprint
+from src.functionWebhook import functionWebhookBlueprint
+import random
+from hmac import new as hmacNew
+from hashlib import sha256
+
+VALID_TOKEN = "1234"
+
+class testingRequest(Request):
+    def __init__(self, token:str):
+        self.data = random.randbytes(50)
+        hash_object = hmacNew(token.encode("utf-8"), msg=self.data, digestmod=sha256)
+        self.headers = {"X-Hub-Signature-256": f"sha256={hash_object.hexdigest()}", "X-Gitlab-Token":token}
+    def get_data(self):
+        return self.data
+
+class TestVerification(unittest.TestCase):
+    def setUp(self) -> None:
+        self.validRequest = testingRequest(VALID_TOKEN)
+        self.invalidRequest = testingRequest("12345")
+        return super().setUp()
+    def testVerifyGithubRequest(self):
+        self.assertTrue(verifyGithubRequest(self.validRequest, VALID_TOKEN))
+        self.assertFalse(verifyGithubRequest(self.validRequest, "12345"))
+        self.assertFalse(verifyGithubRequest(self.invalidRequest, VALID_TOKEN))
+        self.assertFalse(verifyGithubRequest(self.invalidRequest, "12346"))
+    def testVerifyGitlabRequest(self):
+        self.assertTrue(verifyGitlabRequest(self.validRequest, "1234"))
+        self.assertFalse(verifyGitlabRequest(self.validRequest, "12345"))
+        self.assertFalse(verifyGitlabRequest(self.invalidRequest, "1234"))
+
+class TestWehbookBlueprint(unittest.TestCase):
+    def setUp(self) -> None:
+        self.webhook = webhookBlueprint(VALID_TOKEN, name="valid")
+        self.app = Flask(__name__)
+        self.app.register_blueprint(self.webhook, url_prefix="/valid")
+        self.webhookNoToken = webhookBlueprint(None, name="invalid")
+        self.app.register_blueprint(self.webhookNoToken, url_prefix="/noToken")
+        self.app.config.update({"TESTING": True})
+        self.client = self.app.test_client()
+        return super().setUp()
+    def testReceiveWebhookValid(self):
+        request = testingRequest(VALID_TOKEN)
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 415)
+        request.headers["Content-Type"] = "application/json"
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 400) #no json
+    def testReceiveWebhookInvalidNoCheck(self):
+        request = testingRequest("123")
+        resp = self.client.post("/noToken/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 415)
+        request.headers["Content-Type"] = "application/json"
+        resp = self.client.post("/noToken/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 400)
+    def testReceiveWebhookInvalidCheck(self):
+        request = testingRequest("123")
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 415)
+        request.headers["Content-Type"] = "application/json"
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 401)
+    def testProcessWebhook(self):
+        self.assertEqual(self.webhook.processWebhook({"test":"test"}), (200, "OK"))
+        self.assertEqual(self.webhookNoToken.processWebhook({"test":"test"}), (200, "OK"))
+
+class TestFunctionWebhookBlueprint(unittest.TestCase):
+    def setUp(self) -> None:
+        func = lambda x: x != x
+        self.webhook = functionWebhookBlueprint(VALID_TOKEN, name="valid", functions=[func])
+        self.app = Flask(__name__)
+        self.app.register_blueprint(self.webhook, url_prefix="/valid")
+        self.app.config.update({"TESTING": True})
+        self.client = self.app.test_client()
+        return super().setUp()
+    def testReceiveWebhookValid(self):
+        request = testingRequest(VALID_TOKEN)
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 415)
+        request.headers["Content-Type"] = "application/json"
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 400) #no json
+    def testReceiveWebhookInvalidCheck(self):
+        request = testingRequest("123")
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 415)
+        request.headers["Content-Type"] = "application/json"
+        resp = self.client.post("/valid/", headers=request.headers, data=request.data)
+        self.assertEqual(resp.status_code, 401)
+    def testProcessWebhook(self):
+        self.assertEqual(self.webhook.processWebhook({"test":"test"}), (400, "Function <lambda> returned false"))
+
+if __name__ == "__main__":
+    unittest.main()