"""Webhook receiver for Gitea and GitHub events (T49). Mounted at /webhooks/{source} in api.py. Validates HMAC signatures, normalises payloads to EventEnvelope, validates against the event type registry, and publishes to NATS subject activity.events. Config: WEBHOOK_SECRET_GITEA — shared secret for Gitea HMAC-SHA256 WEBHOOK_SECRET_GITHUB — shared secret for GitHub HMAC-SHA256 NATS_URL — NATS server URL (default: nats://localhost:4222) """ from __future__ import annotations import hashlib import hmac import json import logging import os import uuid from datetime import datetime, timezone from typing import Any from fastapi import APIRouter, Header, HTTPException, Request from activity_core.models import EventEnvelope logger = logging.getLogger(__name__) WEBHOOK_SECRET_GITEA = os.environ.get("WEBHOOK_SECRET_GITEA", "") WEBHOOK_SECRET_GITHUB = os.environ.get("WEBHOOK_SECRET_GITHUB", "") NATS_URL = os.environ.get("NATS_URL", "nats://localhost:4222") router = APIRouter(prefix="/webhooks", tags=["webhooks"]) def _verify_hmac(body: bytes, signature: str, secret: str) -> bool: """Return True if signature matches HMAC-SHA256(body, secret).""" if not secret: return False expected = "sha256=" + hmac.new( secret.encode(), body, hashlib.sha256 ).hexdigest() return hmac.compare_digest(expected.encode(), signature.encode()) # ── Gitea normalisers ───────────────────────────────────────────────────────── def _gitea_repo_created(payload: dict, now: datetime) -> EventEnvelope | None: """Gitea repository event (action=created) → gitea.repo.created.""" if payload.get("action") != "created": return None repo = payload.get("repository") or payload name = repo.get("full_name", "") slug = repo.get("name", name.split("/")[-1] if "/" in name else name) return EventEnvelope( id=str(uuid.uuid4()), type="gitea.repo.created", timestamp=now, publisher="gitea/webhook", attributes={ "repo_full_name": name, "repo_slug": slug, "owner": (repo.get("owner") or {}).get("login", ""), "html_url": repo.get("html_url", ""), "created_at": repo.get("created", now.isoformat()), }, ) def _gitea_push(payload: dict, now: datetime) -> EventEnvelope | None: """Gitea push event → gitea.push.""" repo = payload.get("repository") or {} ref = payload.get("ref", "") branch = ref.removeprefix("refs/heads/") if ref.startswith("refs/heads/") else ref commits = payload.get("commits") or [] return EventEnvelope( id=str(uuid.uuid4()), type="gitea.push", timestamp=now, publisher="gitea/webhook", attributes={ "repo_full_name": repo.get("full_name", ""), "branch": branch, "pusher": (payload.get("pusher") or {}).get("login", ""), "commits_count": len(commits), "compare_url": payload.get("compare", ""), "pushed_at": now.isoformat(), }, ) def _gitea_issue_closed(payload: dict, now: datetime) -> EventEnvelope | None: """Gitea issues event (action=closed) → gitea.issue.closed.""" if payload.get("action") != "closed": return None issue = payload.get("issue") or {} repo = payload.get("repository") or {} return EventEnvelope( id=str(uuid.uuid4()), type="gitea.issue.closed", timestamp=now, publisher="gitea/webhook", attributes={ "repo_full_name": repo.get("full_name", ""), "issue_number": issue.get("number", 0), "issue_title": issue.get("title", ""), "closer": (payload.get("sender") or {}).get("login", ""), "closed_at": now.isoformat(), }, ) _GITEA_NORMALISERS: dict[str, Any] = { "repository": _gitea_repo_created, "push": _gitea_push, "issues": _gitea_issue_closed, } # ── GitHub normalisers ──────────────────────────────────────────────────────── def _github_push(payload: dict, now: datetime) -> EventEnvelope | None: """GitHub push event → gitea.push (reuse same event type).""" repo = payload.get("repository") or {} ref = payload.get("ref", "") branch = ref.removeprefix("refs/heads/") if ref.startswith("refs/heads/") else ref commits = payload.get("commits") or [] return EventEnvelope( id=str(uuid.uuid4()), type="gitea.push", timestamp=now, publisher="github/webhook", attributes={ "repo_full_name": repo.get("full_name", ""), "branch": branch, "pusher": (payload.get("pusher") or {}).get("name", ""), "commits_count": len(commits), "compare_url": payload.get("compare", ""), "pushed_at": now.isoformat(), }, ) _GITHUB_NORMALISERS: dict[str, Any] = { "push": _github_push, } async def _publish_to_nats(envelope: EventEnvelope) -> None: """Publish the normalised envelope to NATS subject activity.events.""" try: import nats nc = await nats.connect(NATS_URL) subject = f"activity.{envelope.type}" await nc.publish(subject, envelope.model_dump_json().encode()) await nc.flush() await nc.close() logger.info("published %r to NATS subject %r", envelope.id, subject) except Exception as exc: logger.error("failed to publish %r to NATS: %s", envelope.id, exc) raise # ── Route handlers ──────────────────────────────────────────────────────────── @router.post("/gitea") async def receive_gitea( request: Request, x_gitea_event: str = Header(default=""), x_gitea_signature_256: str = Header(default="", alias="X-Gitea-Signature-256"), ) -> dict[str, str]: body = await request.body() if WEBHOOK_SECRET_GITEA: if not _verify_hmac(body, x_gitea_signature_256, WEBHOOK_SECRET_GITEA): raise HTTPException(status_code=401, detail="invalid signature") try: payload = json.loads(body) except json.JSONDecodeError: raise HTTPException(status_code=400, detail="invalid JSON body") now = datetime.now(tz=timezone.utc) event_type = x_gitea_event.lower() normaliser = _GITEA_NORMALISERS.get(event_type) if normaliser is None: raise HTTPException( status_code=422, detail=f"unsupported Gitea event type: {event_type!r}", ) envelope = normaliser(payload, now) if envelope is None: return {"status": "ignored", "reason": "action not mapped"} from activity_core.event_type_registry import is_event_type_allowed if not is_event_type_allowed(envelope.type): raise HTTPException( status_code=422, detail=f"event type {envelope.type!r} not registered or not accepted", ) await _publish_to_nats(envelope) return {"status": "accepted", "event_id": envelope.id} @router.post("/github") async def receive_github( request: Request, x_github_event: str = Header(default="", alias="X-GitHub-Event"), x_hub_signature_256: str = Header(default="", alias="X-Hub-Signature-256"), ) -> dict[str, str]: body = await request.body() if WEBHOOK_SECRET_GITHUB: if not _verify_hmac(body, x_hub_signature_256, WEBHOOK_SECRET_GITHUB): raise HTTPException(status_code=401, detail="invalid signature") try: payload = json.loads(body) except json.JSONDecodeError: raise HTTPException(status_code=400, detail="invalid JSON body") now = datetime.now(tz=timezone.utc) event_type = x_github_event.lower() normaliser = _GITHUB_NORMALISERS.get(event_type) if normaliser is None: raise HTTPException( status_code=422, detail=f"unsupported GitHub event type: {event_type!r}", ) envelope = normaliser(payload, now) if envelope is None: return {"status": "ignored", "reason": "action not mapped"} from activity_core.event_type_registry import is_event_type_allowed if not is_event_type_allowed(envelope.type): raise HTTPException( status_code=422, detail=f"event type {envelope.type!r} not registered or not accepted", ) await _publish_to_nats(envelope) return {"status": "accepted", "event_id": envelope.id}