Implement Webhook Monitoring

.dockerignore

@@ -0,0 +1,5 @@
+.idea/
+**/json/
+.github/
+.venv/
+!env/.env.example

src/data/json_handler.py

@@ -1,28 +1,25 @@
 # from git import Repo
 import os
 from json import load
-from git import Repo
+import src.utils.log as log
+import git
 
 
 raw_json_data = {}
 json_data = {}
 helper_data = {}
 
 
-def check_if_json_files_exist():
-    path = "./src/data/json"
-    if os.path.exists(path):
-        directory = os.listdir("./src/data/json")
-        return len(directory) == 0
-    return False
-
-
 def get_jsons_from_github():
     path = "./src/data/json"
+    git_url = "https://github.com/helldivers-2/json"
     if not os.path.exists(path):
-        git_url = "https://github.com/helldivers-2/json"
-
-        Repo.clone_from(git_url, path)
+        log.msg("Clone JSON Repository")
+        git.Repo.clone_from(git_url, path)
+    else:
+        log.msg("Pull JSON Repository")
+        repo = git.Repo(path)
+        repo.remotes.origin.pull()
 
 
 def get_json_files():
@@ -168,7 +165,12 @@ def expand_json():
             page.pop("assets")
 
 
-get_jsons_from_github()
-get_json_files()
-sort_json_dicts()
-expand_json()
+def update_json(check_files: bool = True):
+    log.msg("Init / Update JSON")
+    get_jsons_from_github()
+    get_json_files()
+    sort_json_dicts()
+    expand_json()
+
+
+update_json()

src/main.py

@@ -10,7 +10,7 @@
 from src.utils.middleware.authentication import authenticate
 
 # Routes Import
-from src.routes import base, admin, raw, v1
+from src.routes import base, admin, raw, v1, webhook_in
 
 # -------- INIT API -------- #
 app: FastAPI = FastAPI(
@@ -44,3 +44,6 @@
 app.include_router(admin.router)
 app.include_router(raw.router)
 app.include_router(v1.router)
+
+# -------- WEBHOOK -------- #
+app.include_router(webhook_in.router)

src/routes/webhook_in.py

@@ -0,0 +1,53 @@
+from fastapi import APIRouter, status, HTTPException
+from fastapi.requests import Request
+from starlette.responses import Response
+import json
+from src.data.json_handler import update_json
+import src.utils.log as log
+from src.cfg.settings import security
+import hashlib
+import hmac
+
+router = APIRouter(
+    prefix="/wh",
+    tags=["webhook"],
+    include_in_schema=False,
+)
+
+
+class Webhook:
+    def __init__(self, token):
+        self.token = token
+
+    def github_verify(self, payload, headers):
+        signature_header = headers.get("x-hub-signature-256", None)
+        if not signature_header:
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="x-hub-signature-256 header missing",
+            )
+        hash_object = hmac.new(
+            self.token.encode("utf-8"), msg=payload, digestmod=hashlib.sha256
+        )
+        expected_signature = "sha256=" + hash_object.hexdigest()
+        if not hmac.compare_digest(signature_header, expected_signature):
+            raise HTTPException(
+                status_code=status.HTTP_403_FORBIDDEN,
+                detail="Request signatures invalid.",
+            )
+
+
+@router.post(path="/github", status_code=status.HTTP_204_NO_CONTENT)
+async def github_webhook(request: Request, response: Response):
+    headers = request.headers
+    payload = await request.body()
+    json_payload = json.loads(payload)
+    try:
+        wh = Webhook(security["token"])
+        msg = wh.github_verify(payload, headers)
+    except HTTPException as e:
+        response.status_code = status.HTTP_400_BAD_REQUEST
+        return
+
+    if json_payload["action"] == "push":
+        update_json()