appfactory-webhook/app/main.py

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", ""))

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 WEBHOOK_SECRET:
        raise HTTPException(status_code=500, detail="Webhook secret is not configured")

    if not signature:
        raise HTTPException(status_code=401, detail="Missing signature")

    digest = hmac.new(WEBHOOK_SECRET.encode("utf-8"), body, hashlib.sha256).hexdigest()
    accepted = {digest, f"sha256={digest}"}

    signature = signature.strip()

    if not any(hmac.compare_digest(signature, item) for item in accepted):
        raise HTTPException(status_code=401, detail="Invalid signature")


def first_commit(payload: dict) -> dict:
    commits = payload.get("commits") or []
    if commits:
        return commits[-1] or {}
    return payload.get("head_commit") or {}


def extract_attribution(payload: dict) -> dict:
    pusher = payload.get("pusher") or {}
    sender = payload.get("sender") or {}
    commit = first_commit(payload)
    author = commit.get("author") or {}

    pusher_name = (
        pusher.get("full_name")
        or pusher.get("username")
        or pusher.get("name")
        or sender.get("full_name")
        or sender.get("login")
        or sender.get("username")
        or "unknown"
    )

    pusher_username = (
        pusher.get("username")
        or pusher.get("login")
        or sender.get("login")
        or sender.get("username")
        or pusher_name
        or "unknown"
    )

    commit_author = (
        author.get("name")
        or author.get("username")
        or author.get("email")
        or "unknown"
    )

    commit_sha = commit.get("id") or commit.get("sha") or payload.get("after") or "unknown"

    return {
        "pusher": str(pusher_name),
        "pusher_username": str(pusher_username),
        "commit_author": str(commit_author),
        "commit_sha": str(commit_sha),
    }


def run_command(command: list[str], repo_name: str, deploy_env: dict[str, str]):
    env = os.environ.copy()
    env.update(deploy_env)

    print(f"Deploy started: {repo_name}", flush=True)
    print(f"Command: {' '.join(command)}", flush=True)
    print(f"Trigger source: {env.get('APPFACTORY_TRIGGER_SOURCE')}", flush=True)
    print(f"Pusher: {env.get('APPFACTORY_PUSHER')}", flush=True)
    print(f"Commit author: {env.get('APPFACTORY_COMMIT_AUTHOR')}", flush=True)

    result = subprocess.run(command, capture_output=True, text=True, env=env)

    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),
    x_hub_signature: str | None = Header(default=None),
    x_hub_signature_256: str | None = Header(default=None),
):
    body = await request.body()
    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:
        raise HTTPException(status_code=400, detail="Missing repository name")

    attr = extract_attribution(payload)

    if repo_name in CORE_SERVICES:
        command = [CORE_DEPLOY_SCRIPT, repo_name]
    else:
        command = [APP_DEPLOY_SCRIPT, repo_name]

    deploy_env = {
        "APPFACTORY_TRIGGER_SOURCE": "webhook",
        "APPFACTORY_TRIGGERED_BY_USERNAME": attr["pusher_username"],
        "APPFACTORY_TRIGGERED_BY_DISPLAY_NAME": attr["pusher"],
        "APPFACTORY_COMMIT_AUTHOR": attr["commit_author"],
        "APPFACTORY_PUSHER": attr["pusher"],
        "APPFACTORY_COMMIT_SHA": attr["commit_sha"],
    }

    threading.Thread(
        target=run_command,
        args=(command, repo_name, deploy_env),
        daemon=True,
    ).start()

    return {
        "status": "accepted",
        "repo": repo_name,
        "core": repo_name in CORE_SERVICES,
        "triggered_by": attr["pusher_username"],
        "commit_author": attr["commit_author"],
    }