onyx_online / Github Webhook Proxy

from fastapi import FastAPI, Request, Depends, Header from contextlib import asynccontextmanager import niquests import logging import handlers from dependencies import get_request_client logger = logging.getLogger(__name__) @asynccontextmanager async def lifespan(app: FastAPI): app.state.request_client = niquests.AsyncSession() yield await app.state.request_client.close() app = FastAPI(lifespan=lifespan) @app.post("/github/{repo_name}") async def github_webhook( request: Request, repo_name: str, client: niquests.AsyncSession = Depends(get_request_client), x_real_ip: str = Header(...), x_github_event: str = Header(...), x_hub_signature: str = Header(...), ): payload = await request.json() match x_github_event: case "issues": await handlers.issues(payload, client) case _: logger.debug(f"unhandled event {x_github_event}") return { "message": f"received webhook for {repo_name}: {x_github_event}", "status": "success", }

from fastapi import Request, Header, Depends import niquests from fastapi import HTTPException import logging import hashlib import hmac from settings import settings from ipaddress import ip_address, ip_network from typing import Annotated logger = logging.getLogger(__name__) def get_request_client(request: Request) -> niquests.AsyncSession: return request.app.state.request_client async def validate_github_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 ): logger.debug(f"Received request from invalid IP: {x_real_ip}") ERR_MSG = f"IP address {x_real_ip} is not allowed" raise HTTPException(status_code=403, detail=ERR_MSG) async def validate_webhook_secret( request: Request, x_hub_signature: str = Header(...) ) -> None: if not x_hub_signature: logger.debug("Missing X-Hub-Signature header") 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): logger.debug("Invalid HMAC signature") raise HTTPException(status_code=403, detail="Invalid HMAC signature")

from niquests import AsyncSession from settings import settings import logging from fastapi import HTTPException logger = logging.getLogger(__name__) async def issues(payload, request_client: AsyncSession): if payload["action"] not in ["opened", "closed", "reopened", "deleted"]: return embed = { "author": { "name": payload["sender"]["login"], "icon_url": payload["sender"]["avatar_url"], }, "title": f"[{payload['repository']['full_name']}] Issue {payload['action'].capitalize()}: #{payload['issue']['number']} {payload['issue']['title']}", "url": payload["issue"]["html_url"], "description": payload["issue"].get("body"), } data = { "username": "Github", "embeds": [embed], } await send_discord_webhook(request_client=request_client, data=data) 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") logger.info(f"Discord webhook sent with status code {resp.status_code}")

from pydantic_settings import BaseSettings, SettingsConfigDict class Settings(BaseSettings): """Application settings.""" GITHUB_WEBHOOK_SECRET: str DISCORD_WEBHOOK_URL: str model_config = SettingsConfigDict(env_file=".env", env_file_encoding="utf-8") settings = Settings()