AppFactory Bot
116 lines
2.7 KiB
Python
116 lines
2.7 KiB
Python
import hashlib
|
|
import hmac
|
|
import json
|
|
import os
|
|
import subprocess
|
|
import threading
|
|
|
|
from fastapi import FastAPI, Header, HTTPException, Request
|
|
|
|
|
|
def read_env_file_value(key: str, default: str = "") -> str:
|
|
try:
|
|
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:
|
|
pass
|
|
|
|
return default
|
|
|
|
|
|
WEBHOOK_SECRET = os.getenv(
|
|
"WEBHOOK_SECRET",
|
|
read_env_file_value("WEBHOOK_SECRET", "change-me-webhook-secret")
|
|
)
|
|
|
|
APP_DEPLOY_SCRIPT = "/tools/deploy-app.sh"
|
|
CORE_DEPLOY_SCRIPT = "/tools/deploy-core-service.sh"
|
|
|
|
CORE_SERVICES = {
|
|
"appfactory-portal",
|
|
"appfactory-webhook",
|
|
}
|
|
|
|
|
|
app = FastAPI(title="AppFactory Webhook")
|
|
|
|
|
|
@app.get("/health")
|
|
def health():
|
|
return {"status": "ok"}
|
|
|
|
|
|
def verify_signature(body: bytes, signature: str | None):
|
|
if not signature:
|
|
raise HTTPException(status_code=401, detail="Missing signature")
|
|
|
|
expected = "sha256=" + hmac.new(
|
|
WEBHOOK_SECRET.encode("utf-8"),
|
|
body,
|
|
hashlib.sha256
|
|
).hexdigest()
|
|
|
|
if not hmac.compare_digest(expected, signature):
|
|
raise HTTPException(status_code=401, detail="Invalid signature")
|
|
|
|
|
|
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
|
|
)
|
|
|
|
print(f"Deploy finished: {repo_name}", flush=True)
|
|
print(f"Return code: {result.returncode}", flush=True)
|
|
|
|
if result.stdout:
|
|
print("STDOUT:", flush=True)
|
|
print(result.stdout, flush=True)
|
|
|
|
if result.stderr:
|
|
print("STDERR:", flush=True)
|
|
print(result.stderr, flush=True)
|
|
|
|
|
|
@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 = json.loads(body.decode("utf-8"))
|
|
|
|
repo_name = payload.get("repository", {}).get("name")
|
|
|
|
if not repo_name:
|
|
raise HTTPException(status_code=400, detail="Missing repository name")
|
|
|
|
if repo_name in CORE_SERVICES:
|
|
command = [CORE_DEPLOY_SCRIPT, repo_name]
|
|
else:
|
|
command = [APP_DEPLOY_SCRIPT, repo_name]
|
|
|
|
thread = threading.Thread(
|
|
target=run_command,
|
|
args=(command, repo_name),
|
|
daemon=True,
|
|
)
|
|
thread.start()
|
|
|
|
return {
|
|
"status": "accepted",
|
|
"repo": repo_name,
|
|
"core": repo_name in CORE_SERVICES,
|
|
}
|