Improve webhook signature handling

app/main.py

@@ -13,7 +13,6 @@ def read_env_file_value(key: str, default: str = "") -> str:
         with open("/opt/appfactory/config/appfactory.env", "r", encoding="utf-8") as f:
             for line in f:
                 line = line.strip()
-
                 if line.startswith(f"{key}="):
                     return line.split("=", 1)[1].strip().strip('"')
     except Exception:
@@ -24,7 +23,7 @@ def read_env_file_value(key: str, default: str = "") -> str:
 
 WEBHOOK_SECRET = os.getenv(
     "WEBHOOK_SECRET",
-    read_env_file_value("WEBHOOK_SECRET", "change-me-webhook-secret")
+    read_env_file_value("WEBHOOK_SECRET", "")
 )
 
 APP_DEPLOY_SCRIPT = "/tools/deploy-app.sh"
@@ -35,7 +34,6 @@ CORE_SERVICES = {
     "appfactory-webhook",
 }
 
-
 app = FastAPI(title="AppFactory Webhook")
 
 
@@ -45,6 +43,9 @@ def health():
 
 
 def verify_signature(body: bytes, signature: str | None):
+    if not WEBHOOK_SECRET:
+        raise HTTPException(status_code=500, detail="Webhook secret is not configured")
+
     if not signature:
         raise HTTPException(status_code=401, detail="Missing signature")
 
@@ -54,12 +55,17 @@ def verify_signature(body: bytes, signature: str | None):
         hashlib.sha256
     ).hexdigest()
 
-    valid_signatures = {
+    accepted = {
         digest,
         f"sha256={digest}",
     }
 
-    if not any(hmac.compare_digest(signature, valid) for valid in valid_signatures):
+    signature = signature.strip()
+
+    if not any(hmac.compare_digest(signature, item) for item in accepted):
+        print("Invalid signature", flush=True)
+        print(f"Received prefix: {signature[:16]}", flush=True)
+        print(f"Expected prefix: {digest[:16]}", flush=True)
         raise HTTPException(status_code=401, detail="Invalid signature")
 
 
@@ -67,11 +73,7 @@ def run_command(command: list[str], repo_name: str):
     print(f"Deploy started: {repo_name}", flush=True)
     print(f"Command: {' '.join(command)}", flush=True)
 
-    result = subprocess.run(
-        command,
-        capture_output=True,
-        text=True
-    )
+    result = subprocess.run(command, capture_output=True, text=True)
 
     print(f"Deploy finished: {repo_name}", flush=True)
     print(f"Return code: {result.returncode}", flush=True)
@@ -89,13 +91,16 @@ def run_command(command: list[str], repo_name: str):
 async def gitea_webhook(
     request: Request,
     x_gitea_signature: str | None = Header(default=None),
+    x_hub_signature: str | None = Header(default=None),
+    x_hub_signature_256: str | None = Header(default=None),
 ):
     body = await request.body()
 
-    verify_signature(body, x_gitea_signature)
+    signature = x_gitea_signature or x_hub_signature_256 or x_hub_signature
+
+    verify_signature(body, signature)
 
     payload = json.loads(body.decode("utf-8"))
-
     repo_name = payload.get("repository", {}).get("name")
 
     if not repo_name:
@@ -106,12 +111,11 @@ async def gitea_webhook(
     else:
         command = [APP_DEPLOY_SCRIPT, repo_name]
 
-    thread = threading.Thread(
+    threading.Thread(
         target=run_command,
         args=(command, repo_name),
         daemon=True,
-    )
-    thread.start()
+    ).start()
 
     return {
         "status": "accepted",