Webhook Handlers

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()

@app.post("/webhooks/{source}")
async def receive_webhook(source: str, request: Request):
    """Generic webhook receiver with signature verification."""
    body = await request.body()
    headers = dict(request.headers)

    # Log the webhook
    print(f"Received webhook from {source}")
    print(f"Headers: {headers}")
    print(f"Body: {body.decode()}")

    # Verify signature if present
    signature = headers.get("x-hub-signature-256") or headers.get("x-webhook-signature")
    if signature:
        if not verify_signature(body, signature, secret="my-secret"):
            raise HTTPException(status_code=401, detail="Invalid signature")

    # Process the webhook
    return {"status": "received", "source": source}

from fastapi import FastAPI, Request, HTTPException
import hmac
import hashlib

app = FastAPI()

GITHUB_SECRET = b"my-webhook-secret"

def verify_github_signature(body: bytes, signature: str, secret: bytes) -> bool:
    """Verify GitHub webhook signature (HMAC-SHA256)."""
    expected = "sha256=" + hmac.new(secret, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, signature)

@app.post("/webhooks/github")
async def github_webhook(request: Request):
    body = await request.body()
    signature = request.headers.get("X-Hub-Signature-256", "")

    if not verify_github_signature(body, signature, GITHUB_SECRET):
        raise HTTPException(status_code=401, detail="Invalid signature")

    import json
    payload = json.loads(body)
    event = request.headers.get("X-GitHub-Event")

    if event == "push":
        branch = payload["ref"].replace("refs/heads/", "")
        commits = payload["commits"]
        print(f"Push to {branch}: {len(commits)} commits")
        for commit in commits[:5]:
            print(f"  {commit['message'][:80]}")

    elif event == "pull_request":
        action = payload["action"]
        pr = payload["pull_request"]
        print(f"PR #{pr['number']}: {action} - {pr['title']}")

    return {"status": "ok"}

from fastapi import FastAPI, Request, HTTPException
from fastapi.responses import JSONResponse
import hashlib
import hmac
import time

app = FastAPI()

SLACK_SIGNING_SECRET = "your-signing-secret"

def verify_slack_signature(timestamp: str, body: str, signature: str) -> bool:
    """Verify Slack request signature."""
    if abs(time.time() - float(timestamp)) > 300:
        return False  # Request too old
    sig_basestring = f"v0:{timestamp}:{body}"
    computed = "v0=" + hmac.new(
        SLACK_SIGNING_SECRET.encode(), sig_basestring.encode(), hashlib.sha256
    ).hexdigest()
    return hmac.compare_digest(computed, signature)

@app.post("/webhooks/slack")
async def slack_webhook(request: Request):
    body = await request.body()
    timestamp = request.headers.get("X-Slack-Request-Timestamp", "")
    signature = request.headers.get("X-Slack-Signature", "")

    if not verify_slack_signature(timestamp, body.decode(), signature):
        raise HTTPException(status_code=401, detail="Invalid signature")

    import json
    payload = json.loads(body)

    # Handle URL verification challenge
    if payload.get("type") == "url_verification":
        return {"challenge": payload["challenge"]}

    # Handle slash commands
    if payload.get("type") == "slash_command":
        command = payload["command"]
        text = payload["text"]
        return JSONResponse(content={
            "response_type": "in_channel",
            "text": f"Running {command} {text}...",
        })

    return {"ok": True}

from fastapi import FastAPI, Request
from pydantic import BaseModel

app = FastAPI()

class Alert(BaseModel):
    status: str
    labels: dict[str, str]
    annotations: dict[str, str]
    startsAt: str
    generatorURL: str

class AlertmanagerPayload(BaseModel):
    status: str
    alerts: list[Alert]

@app.post("/webhooks/alertmanager")
async def alertmanager_webhook(payload: AlertmanagerPayload):
    for alert in payload.alerts:
        if alert.status == "firing":
            print(f"FIRING: {alert.labels.get('alertname')}")
            print(f"  Severity: {alert.labels.get('severity')}")
            print(f"  Summary: {alert.annotations.get('summary')}")
            print(f"  Runbook: {alert.annotations.get('runbook_url', 'N/A')}")

            # Route to appropriate handler based on severity
            severity = alert.labels.get("severity", "info")
            if severity == "critical":
                page_oncall(alert)
            elif severity == "warning":
                send_to_slack(alert)

        elif alert.status == "resolved":
            print(f"RESOLVED: {alert.labels.get('alertname')}")

    return {"status": "processed", "alert_count": len(payload.alerts)}

import hmac
import hashlib
from fastapi import HTTPException

def verify_signature(
    body: bytes,
    signature: str,
    secret: str,
    algorithm: str = "sha256",
) -> bool:
    """Verify HMAC signature for any webhook source."""
    if not signature:
        return False

    # Strip prefix (e.g., "sha256=")
    if "=" in signature:
        algorithm, signature = signature.split("=", 1)

    secret_bytes = secret.encode() if isinstance(secret, str) else secret
    computed = hmac.new(secret_bytes, body, getattr(hashlib, algorithm)).hexdigest()

    # Constant-time comparison prevents timing attacks
    return hmac.compare_digest(f"{algorithm}={computed}", f"{algorithm}={signature}")