Guide: GitHub events → PartyFlow через прокси#

[GitHub] ──POST + X-Hub-Signature-256──▶ [Your proxy] ──POST + X-PartyFlow-Signature──▶ [PartyFlow]
          (raw GitHub payload)                                (Slack-формат message)

"""
GitHub → PartyFlow webhook proxy.
 
Accepts GitHub webhooks, verifies X-Hub-Signature-256, converts a handful
of event types into Slack-compatible payloads, signs with PartyFlow
native HMAC and forwards to the incoming webhook URL.
"""
import hashlib
import hmac
import json
import os
import time
 
import httpx
from fastapi import FastAPI, Header, HTTPException, Request
 
# ── Config ────────────────────────────────────────────────────────────
GITHUB_SECRET = os.environ["GITHUB_WEBHOOK_SECRET"].encode()
PARTYFLOW_URL = os.environ["PARTYFLOW_WEBHOOK_URL"]         # https://.../api/v1/webhooks/incoming/whk_...
PARTYFLOW_SECRET = os.environ["PARTYFLOW_SIGNING_SECRET"].encode()
 
app = FastAPI()
 
 
# ── GitHub signature verification ─────────────────────────────────────
def verify_github_signature(secret: bytes, body: bytes, header: str) -> bool:
    if not header or not header.startswith("sha256="):
        return False
    expected = "sha256=" + hmac.new(secret, body, hashlib.sha256).hexdigest()
    return hmac.compare_digest(expected, header)
 
 
# ── Event converters: GitHub → Slack-format ───────────────────────────
def convert_pull_request(event: dict) -> dict | None:
    action = event.get("action")
    pr = event.get("pull_request", {})
    repo = event.get("repository", {}).get("full_name", "?")
    author = pr.get("user", {}).get("login", "?")
    title = pr.get("title", "")
    url = pr.get("html_url", "")
    number = pr.get("number", 0)
 
    if action == "opened":
        return {
            "text": f":inbox_tray: New PR #{number} in `{repo}`",
            "attachments": [{
                "color": "#36a64f",
                "title": title,
                "title_link": url,
                "author_name": author,
                "text": (pr.get("body") or "")[:500],
            }],
        }
 
    if action == "closed" and pr.get("merged"):
        return {
            "text": f":white_check_mark: PR #{number} merged to `{repo}`",
            "attachments": [{
                "color": "#6f42c1",
                "title": title,
                "title_link": url,
                "author_name": author,
            }],
        }
 
    # Other PR actions — skip.
    return None
 
 
def convert_push(event: dict) -> dict | None:
    # Skip branch-delete events and pushes with no commits.
    commits = event.get("commits") or []
    if not commits:
        return None
 
    repo = event.get("repository", {}).get("full_name", "?")
    ref = event.get("ref", "").replace("refs/heads/", "")
    pusher = event.get("pusher", {}).get("name", "?")
    compare_url = event.get("compare", "")
 
    lines = [f"- `{c['id'][:7]}` {c['message'].splitlines()[0]}" for c in commits[:5]]
    more = f"\n… and {len(commits) - 5} more commits" if len(commits) > 5 else ""
 
    return {
        "text": f":arrow_up: {pusher} pushed {len(commits)} commit(s) to `{repo}/{ref}`",
        "attachments": [{
            "color": "#0366d6",
            "title": "Compare diff",
            "title_link": compare_url,
            "text": "\n".join(lines) + more,
        }],
    }
 
 
# ── PartyFlow forwarding ──────────────────────────────────────────────
def sign_and_forward(payload: dict) -> int:
    body = json.dumps(payload, separators=(",", ":")).encode()
    ts = str(int(time.time()))
    signing_string = ts.encode() + b"." + body
    signature = "sha256=" + hmac.new(PARTYFLOW_SECRET, signing_string, hashlib.sha256).hexdigest()
 
    with httpx.Client(timeout=10.0) as client:
        resp = client.post(
            PARTYFLOW_URL,
            content=body,
            headers={
                "Content-Type": "application/json",
                "X-PartyFlow-Timestamp": ts,
                "X-PartyFlow-Signature": signature,
            },
        )
    resp.raise_for_status()
    return resp.status_code
 
 
# ── HTTP endpoint ─────────────────────────────────────────────────────
@app.post("/gh/partyflow")
async def github_proxy(
    request: Request,
    x_hub_signature_256: str = Header(default=""),
    x_github_event: str = Header(default=""),
):
    body = await request.body()
 
    if not verify_github_signature(GITHUB_SECRET, body, x_hub_signature_256):
        raise HTTPException(401, "invalid GitHub signature")
 
    event = json.loads(body)
 
    if x_github_event == "pull_request":
        converted = convert_pull_request(event)
    elif x_github_event == "push":
        converted = convert_push(event)
    else:
        converted = None
 
    # Not every event is interesting — skip quietly with 204.
    if converted is None:
        return {"skipped": True, "event": x_github_event}
 
    status = sign_and_forward(converted)
    return {"forwarded": True, "partyflow_status": status}

pip install fastapi uvicorn httpx
 
export GITHUB_WEBHOOK_SECRET='my-github-secret'
export PARTYFLOW_WEBHOOK_URL='https://api.partyflow.ru/api/v1/webhooks/incoming/whk_...'
export PARTYFLOW_SIGNING_SECRET='whsec_...'
 
uvicorn proxy:app --host 0.0.0.0 --port 8080

FROM python:3.12-slim
WORKDIR /app
COPY proxy.py /app/
RUN pip install --no-cache-dir fastapi uvicorn httpx
CMD ["uvicorn", "proxy:app", "--host", "0.0.0.0", "--port", "8080"]

[Unit]
Description=GitHub→PartyFlow proxy
After=network.target
 
[Service]
EnvironmentFile=/etc/gh-partyflow/env
ExecStart=/usr/local/bin/uvicorn proxy:app --host 0.0.0.0 --port 8080
Restart=on-failure
 
[Install]
WantedBy=multi-user.target