Merge pull request #17 from Cloud-Code-AI/6-feat-add-optional-signatu…

…re-check-for-github-bot

feat: Added github signature check for app

api/github_helper/utils.py

@@ -3,6 +3,8 @@
 import time
 import requests
 import logging
+import hmac
+import hashlib
 
 logger = logging.getLogger(__name__)
 
@@ -40,3 +42,21 @@ def get_diff_text(url, access_token):
     response = requests.get(url, headers=headers)
     logger.debug(f"Diff API response: {url}, Resp: {response.text}")
     return response.text
+
+
+def is_github_signature_valid(headers, body):
+    """
+    Validate the signature of the incoming request against the secret.
+    """
+    github_secret = os.environ.get("GITHUB_APP_WEBHOOK_SECRET", "").encode()
+    signature = headers.get('X-Hub-Signature-256')
+
+    if not signature or not github_secret:
+        return False
+
+    sha_name, signature = signature.split('=')
+    if sha_name != 'sha256':
+        return False
+
+    mac = hmac.new(github_secret, msg=body, digestmod=hashlib.sha256)
+    return hmac.compare_digest(mac.hexdigest(), signature)

api/main.py

@@ -1,21 +1,29 @@
-from fastapi import FastAPI, Request, BackgroundTasks
+from fastapi import FastAPI, Request, BackgroundTasks, HTTPException
 from fastapi.responses import JSONResponse
 from api.github_helper.pull_requests import process_pull_request, ACTIONS_TO_PROCESS_PR
+from api.github_helper.utils import (
+    is_github_signature_valid
+)
 import logging
 
 logging.basicConfig(
     level=logging.INFO, format="%(asctime)s - %(name)s - %(levelname)s - %(message)s"
 )
 logger = logging.getLogger(__name__)
 
-
 app = FastAPI()
 
 
 @app.post("/github-webhook")
 async def handle_webhook(request: Request, background_tasks: BackgroundTasks):
     payload = await request.json()
+    body = await request.body()
     event = request.headers.get("X-GitHub-Event")
+    # Check if the Signature is valid
+    # TODO: Make this optional based on user config settings
+    if not is_github_signature_valid(request.headers, body):
+        return HTTPException(status_code=404, detail="Invalid Signature")
+
     if event == "pull_request" and payload["action"] in ACTIONS_TO_PROCESS_PR:
         background_tasks.add_task(process_pull_request, payload)
     else: