AppFactory Bot
70 lines
1.7 KiB
Python
70 lines
1.7 KiB
Python
import hmac
|
|
import hashlib
|
|
import subprocess
|
|
import threading
|
|
from fastapi import FastAPI, Request, Header, HTTPException
|
|
|
|
WEBHOOK_SECRET = "change-me-webhook-secret"
|
|
DEPLOY_SCRIPT = "/tools/deploy-app.sh"
|
|
|
|
app = FastAPI(title="AppFactory Webhook")
|
|
|
|
def verify_signature(body: bytes, signature: str | None):
|
|
if not signature:
|
|
raise HTTPException(status_code=401, detail="Missing signature")
|
|
|
|
expected = hmac.new(
|
|
WEBHOOK_SECRET.encode("utf-8"),
|
|
body,
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
valid_signatures = [
|
|
expected,
|
|
f"sha256={expected}",
|
|
]
|
|
|
|
if not any(hmac.compare_digest(s, signature) for s in valid_signatures):
|
|
raise HTTPException(status_code=401, detail="Invalid signature")
|
|
|
|
def run_deploy(app_id: str):
|
|
subprocess.run(
|
|
[DEPLOY_SCRIPT, app_id],
|
|
capture_output=True,
|
|
text=True
|
|
)
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
@app.post("/gitea")
|
|
async def gitea_webhook(
|
|
request: Request,
|
|
x_gitea_signature: str | None = Header(default=None)
|
|
):
|
|
body = await request.body()
|
|
verify_signature(body, x_gitea_signature)
|
|
|
|
payload = await request.json()
|
|
|
|
repo_name = payload.get("repository", {}).get("name")
|
|
ref = payload.get("ref", "")
|
|
|
|
if not repo_name:
|
|
raise HTTPException(status_code=400, detail="Missing repo name")
|
|
|
|
if ref != "refs/heads/main":
|
|
return {
|
|
"status": "ignored",
|
|
"reason": "not main branch",
|
|
"ref": ref
|
|
}
|
|
|
|
threading.Thread(target=run_deploy, args=(repo_name,), daemon=True).start()
|
|
|
|
return {
|
|
"status": "accepted",
|
|
"repo": repo_name
|
|
}
|