| 12 | |
| 13 | |
| 14 | class GitHubWebhookServer: |
| 15 | def __init__(self, discord_bot=None): |
| 16 | self.app = web.Application() |
| 17 | self.runner: web.AppRunner | None = None |
| 18 | self.site: web.TCPSite | None = None |
| 19 | self.discord_bot = discord_bot |
| 20 | |
| 21 | # Setup routes |
| 22 | self.app.router.add_post("/webhook", self.handle_webhook) |
| 23 | self.app.router.add_get("/health", self.health_check) |
| 24 | |
| 25 | async def start(self): |
| 26 | """Start the webhook server.""" |
| 27 | self.runner = web.AppRunner(self.app) |
| 28 | await self.runner.setup() |
| 29 | self.site = web.TCPSite(self.runner, "0.0.0.0", config.webhook_port) |
| 30 | await self.site.start() |
| 31 | logger.info(f"GitHub webhook server started on port {config.webhook_port}") |
| 32 | |
| 33 | async def stop(self): |
| 34 | """Stop the webhook server.""" |
| 35 | if self.runner: |
| 36 | await self.runner.cleanup() |
| 37 | logger.info("GitHub webhook server stopped") |
| 38 | |
| 39 | async def health_check(self, request: web.Request) -> web.Response: |
| 40 | """Health check endpoint.""" |
| 41 | return web.json_response({"status": "ok", "service": "polly-webhook"}) |
| 42 | |
| 43 | def verify_signature(self, payload: bytes, signature: str) -> bool: |
| 44 | """Verify GitHub webhook signature.""" |
| 45 | if not config.webhook_secret: |
| 46 | # SECURITY: Reject all webhooks if no secret configured |
| 47 | logger.error("GITHUB_WEBHOOK_SECRET not configured - rejecting webhook for security") |
| 48 | return False |
| 49 | |
| 50 | if not signature: |
| 51 | logger.warning("Webhook received without signature header") |
| 52 | return False |
| 53 | |
| 54 | # GitHub sends signature as "sha256=<hash>" |
| 55 | if signature.startswith("sha256="): |
| 56 | signature = signature[7:] |
| 57 | |
| 58 | expected = hmac.new(config.webhook_secret.encode(), payload, hashlib.sha256).hexdigest() |
| 59 | |
| 60 | return hmac.compare_digest(expected, signature) |
| 61 | |
| 62 | async def handle_webhook(self, request: web.Request) -> web.Response: |
| 63 | """Handle incoming GitHub webhook.""" |
| 64 | # Read payload |
| 65 | payload = await request.read() |
| 66 | |
| 67 | # Verify signature |
| 68 | signature = request.headers.get("X-Hub-Signature-256", "") |
| 69 | if not self.verify_signature(payload, signature): |
| 70 | logger.warning("Invalid webhook signature") |
| 71 | return web.json_response({"error": "Invalid signature"}, status=401) |
no outgoing calls
no test coverage detected