onyx, a tech blog

Programming is my primary passion, but I also have a strong interest in Linux server administration. I find webhooks to be an invaluable tool for staying up to date, and more recently, they have become a convenient method for receiving real-time updates from GitHub.

Background

To address this, I opted for a centralized system, which offers several advantages. Primarily, it enables me to manage all webhooks from a single location. Additionally, it provides the flexibility to parse and, if needed, modify a webhook’s payload before forwarding it to Discord.

For instance, consider the excellent Linux backup tool GoBackup. It supports sending webhook notifications — on backup completion — to various platforms including Discord. However, the default webhook message looks like this:

systemd: [GoBackup] OK: Backup service has successfully

Backup of service completed successfully at 2023-10-23 09:17:47.443308376 +0100 BST

This format is perfectly adequate, offering plenty of detail, however, I prefer notifications that are structured in the following way:

<hostname> :: <Service>: <status message>

By including the webhook’s origin (the server hostname), followed by the Linux service name, and finally the status message, I can quickly grasp the essential information. Discord provides a timestamp and if I need more details, I simply refer to the logs.

Implementing the Webhook Proxy

Define our Endpoint

Only a single endpoint is needed. However, validating the incoming request is important. This is achieved by using FastAPI's dependency injection dependency injection in the path operation decorator.

@app.post(
    "/github/{repo_name}",
    dependencies=[
        Depends(validate_source_ip), Depends(validate_webhook_secret)
    ],
)
async def github_webhook(
    repo_name: str,
    request: Request,
    client: niquests.AsyncSession = Depends(get_request_client),
    x_real_ip: str = Header(...),
    x_hub_signature: str = Header(...),
    x_github_event: str = Header(...),
):
    ...

Validate the Source IP

In my case, the server is placed behind a reverse proxy. To determine the source IP, it is read from the X-Real-IP header. This IP is then checked against a list of valid IP ranges, which can be retrieved via the GitHub API.

async def validate_source_ip(
    request_client: Annotated[
        niquests.AsyncSession, Depends(get_request_client)
    ],
    x_real_ip: str = Header(...),
) -> None:
    resp = await request_client.get("https://api.github.com/meta")
    allowed_ips = resp.json().get("hooks", [])

    if not any(
        ip_address(x_real_ip) in ip_network(valid_ip)
        for valid_ip in allowed_ips
    ):
        ERR_MSG = f"IP address {x_real_ip} is not allowed"
        raise HTTPException(status_code=403, detail=ERR_MSG)

Validate the Webhook Secret

Next, the webhook secret is tested. If either of these dependency functions fails, a HTTPException is raised to prevent the request from being processed.

async def validate_webhook_secret(
    request: Request, x_hub_signature: str = Header(...)
) -> None:
    if not x_hub_signature:
        raise HTTPException(
            status_code=400, detail="Missing X-Hub-Signature header"
        )

    mac = hmac.new(
        settings.GITHUB_WEBHOOK_SECRET.encode(),
        msg=await request.body(),
        digestmod=hashlib.sha1,
    )

    if not hmac.compare_digest("sha1=" + mac.hexdigest(), x_hub_signature):
        raise HTTPException(status_code=403, detail="Invalid HMAC signature")

Match the Event and Trigger the Handler

If all validation checks succeed, the event type is read from the X-Github-Event header, and the appropriate handler is triggered.

    match x_github_event:
        case "ping":
            return {"message": "pong"}
        case "issues":
            await handlers.issues(payload, client)
        case _:
            logger.debug(f"unhandled event {x_github_event}")

The request payload can now be processed and converted into a Discord webhook with embedded content:

async def issues(payload, request_client: AsyncSession):
    if payload["action"] not in ["opened", "closed", "reopened", "deleted"]:
        return

    embed = {
        ...
    }
    data = {
        "username": "Github-OpenGist",
        "embeds": [embed],
    }
    await send_discord_webhook(request_client=request_client, data=data)

Finally the Discord webhook is sent:

async def send_discord_webhook(request_client: AsyncSession, data):
    resp = await request_client.post(settings.DISCORD_WEBHOOK_URL, json=data)
    if resp.status_code != 204:
        raise HTTPException(
            status_code=400, detail="Failed to send Discord webhook"
        )

The result is a nicely formatted Discord message:

Conclusion

With everything up and running I can relay and receive Github webhook notification through a proxy server. Other than being written with a different framework the codebase was also converted from sync to async which is much faster for these types of IO operations.

I've posted a version of this proxy server that implements only Github Issues. It can easily be extended to support other Github events.

Gist notes:

• Sending the Discord webhook is handled by the excellent niquests library.
• Managing the request client resource is done via a FastAPI lifespan.
• Loading the proxy server configuration is achieved with Pydantic Settings.

Subscribe to this blog's RSS feed