Receive Webhooks from Github and start a deploy script

app.py

import ipaddress
import json
import os
import subprocess
from enum import Enum

import uvicorn
from dotenv import load_dotenv
from fastapi import (
    BackgroundTasks,
    Depends,
    FastAPI,
    Header,
    HTTPException,
    Request,
    status,
)
from httpx import AsyncClient

load_dotenv()

app = FastAPI()

DEPLOY_SCRIPTS_FILE = os.getenv("DEPLOY_SCRIPTS_FILE", "deploy_scripts.json")
GITHUB_IPS_ONLY = os.getenv("GITHUB_IPS_ONLY", "True").lower() in ["true", "1"]


with open(DEPLOY_SCRIPTS_FILE) as fh:
    # load the deploy scripts from the project directory
    DEPLOY_SCRIPTS = json.load(fh)


def deploy_application(script_name, *args):
    subprocess.run([script_name, *args])


AppNames = Enum("AppNames", [(k, k) for k in DEPLOY_SCRIPTS.keys()], type=str)


async def gate_by_github_ip(request: Request):
    # Allow GitHub IPs only
    if GITHUB_IPS_ONLY:
        try:
            src_ip = ipaddress.ip_address(request.client.host)
        except ValueError:
            raise HTTPException(
                status.HTTP_400_BAD_REQUEST, "Could not hook sender ip address"
            )
        async with AsyncClient() as client:
            allowlist = await client.get("https://api.github.com/meta")
        for valid_ip in allowlist.json()["hooks"]:
            if src_ip in ipaddress.ip_network(valid_ip):
                return
        else:
            raise HTTPException(
                status.HTTP_403_FORBIDDEN, "Not a GitHub hooks ip address"
            )


@app.post("/webhook/{app_name}", dependencies=[Depends(gate_by_github_ip)])
async def receive_payload(
    request: Request,
    app_name: AppNames,
    background_tasks: BackgroundTasks,
    x_github_event: str = Header(...),
):

    if x_github_event == "create":
        payload = await request.json()
        if payload.get("ref_type") == "tag":
            tag_name = payload.get("ref")
            script_name = DEPLOY_SCRIPTS[app_name]
            background_tasks.add_task(deploy_application, script_name, tag_name)
            return {"message": f"Deployment started for tag {tag_name}"}

    elif x_github_event == "push":
        payload = await request.json()
        default_branch = payload["repository"]["default_branch"]
        # check if event is referencing the default branch
        if (
            "ref" in payload
            and payload["ref"] == f"refs/heads/{default_branch}"
        ):
            # check if app_name is declared in config
            script_name = DEPLOY_SCRIPTS[app_name]
            background_tasks.add_task(deploy_application, script_name)
            return {"message": "Deployment started"}

    elif x_github_event == "ping":
        return {"message": "pong"}

    else:
        return {"message": "Unable to process action"}


if __name__ == "__main__":
    uvicorn.run("main:app", host="127.0.0.1", port=5000, log_level="info")

deploy_scripts.json

{
    "mysite": "/usr/bin/deploy_app.sh"
}

nginx.conf

location /webhook {
    proxy_pass http://127.0.0.1:5002;  # Uvicorn app on port 5001
    proxy_set_header Host $host;
    proxy_set_header X-Real-IP $remote_addr;
    proxy_set_header X-Forwarded-For $proxy_add_x_forwarded_for;
    proxy_set_header X-Forwarded-Proto $scheme;

    # Websockets support
    proxy_http_version 1.1;
    proxy_set_header Upgrade $http_upgrade;
    proxy_set_header Connection "upgrade";
}

requirements.txt

fastapi
uvicorn[standard]
pytest
pytest-env
requests
httpx
pytest-asyncio
python-dotenv

webhook-rec.service

[Unit]
Description=FastAPI Webhook Receive
After=network.target

[Service]
User=root
Group=root
WorkingDirectory=/var/www/webhook_receive
# EnvironmentFile=/var/www/webhook_receive/.env
ExecStart=/var/www/webhook_receive/.venv/bin/uvicorn webhook_receive.main:app --host=0.0.0.0 --port=5002
Restart=always

# Logging configuration
StandardOutput=append:/var/log/webhook_receive/output.log
StandardError=append:/var/log/webhook_receive/error.log

[Install]
WantedBy=multi-user.target