اعمل Webhook Receiver آمن بـ FastAPI وتحقق HMAC

اعمل Webhook Receiver آمن بـ FastAPI وتحقق HMAC

هتطلع من المقال ده بـ endpoint حقيقي يستقبل Webhooks، يرفض الطلبات المزيفة قبل ما تلمس منطقك الداخلي، ويشتغل محليًا خلال أقل من 10 دقائق.

مستوى القارئ: متوسط

المشكلة باختصار

الـ Webhook شكله بسيط: خدمة خارجية تبعت `POST` على endpoint عندك. المشكلة إن أي حد يقدر يبعت `POST` لنفس الرابط لو عرفه. الطريقة الشائعة الغلط إنك تقرأ JSON وتبدأ تنفذ المنطق فورًا. الطريقة دي بتفشل لو حد بعت payload مزيف يعمل deploy، يفتح ticket، أو يشغّل job مكلف.

الافتراض إن عندك خدمة صغيرة تستقبل GitHub Webhooks لتشغيل CI job أو تحديث حالة داخلية. عند 1000 طلب في الدقيقة، لو 880 طلب منهم مزيفين، فأنت مش محتاج queue أكبر. أنت محتاج ترفضهم قبل المعالجة.

الفكرة بمثال واضح

ركز في التشبيه العملي: عندك مكتب يستقبل طرود. الموظف مش بيفتح الطرد ويبدأ يفرزه قبل ما يتأكد إن الختم صحيح. لو الختم غلط، الطرد يترفض عند الباب. الـ HMAC هو الختم. مزود الخدمة يستخدم secret مشترك عشان يعمل توقيع للـ payload. السيرفر عندك يعيد حساب نفس التوقيع من الجسم الخام للطلب. لو النتيجة مطابقة، الطلب غالبًا جاي من المصدر الصحيح ولم يتغير في الطريق.

علميًا، GitHub يرسل التوقيع في header اسمه `X-Hub-Signature-256`، والقيمة تبدأ بـ `sha256=`. التوقيع معمول بـ HMAC-SHA256 على محتوى الطلب نفسه. لذلك لازم تستخدم raw bytes من `request.body()` قبل أي تعديل أو parsing. لو قرأت JSON وعدلت المسافات أو encoding، التوقيع ممكن يفشل حتى لو الطلب صحيح.

الخطوات العملية

1. اعمل مجلد جديد وثبت FastAPI وUvicorn.
2. خزّن الـ secret في environment variable، مش داخل الكود.
3. اقرأ جسم الطلب الخام كـ bytes.
4. احسب HMAC-SHA256 بنفس الـ secret.
5. قارن باستخدام `hmac.compare_digest` بدل `==`.
6. بعد التحقق فقط، ابدأ parsing للـ JSON أو ادخل الطلب في queue.

mkdir secure-webhook
cd secure-webhook
python -m venv .venv
.venv\Scripts\activate
pip install fastapi uvicorn
set WEBHOOK_SECRET=change-this-long-random-secret

الكود الكامل

import hashlib
import hmac
import json
import os
from fastapi import FastAPI, Header, HTTPException, Request

app = FastAPI()
SECRET = os.environ.get("WEBHOOK_SECRET", "")

def verify_signature(payload: bytes, signature_header: str | None) -> None:
    if not SECRET:
        raise HTTPException(status_code=500, detail="WEBHOOK_SECRET is not configured")

    if not signature_header:
        raise HTTPException(status_code=403, detail="missing signature")

    expected = "sha256=" + hmac.new(
        SECRET.encode("utf-8"),
        msg=payload,
        digestmod=hashlib.sha256,
    ).hexdigest()

    if not hmac.compare_digest(expected, signature_header):
        raise HTTPException(status_code=403, detail="invalid signature")

@app.post("/webhooks/github")
async def github_webhook(
    request: Request,
    x_hub_signature_256: str | None = Header(default=None),
):
    payload = await request.body()
    verify_signature(payload, x_hub_signature_256)

    event = request.headers.get("x-github-event", "unknown")
    data = json.loads(payload.decode("utf-8"))

    return {
        "ok": True,
        "event": event,
        "repository": data.get("repository", {}).get("full_name"),
    }

شغّله كده:

uvicorn main:app --reload --port 8000

اختبار محلي قبل ربط GitHub

أفضل طريقة للاختبار إنك تولّد توقيع بنفس الـ secret ثم تبعته مع curl. ده يمنعك من ربط GitHub بدري ثم تكتشف إن المشكلة في header أو encoding.

# sign.py
import hashlib, hmac, os
payload = b'{"repository":{"full_name":"ahmed/demo"}}'
secret = os.environ["WEBHOOK_SECRET"].encode("utf-8")
print("sha256=" + hmac.new(secret, payload, hashlib.sha256).hexdigest())

set SIG=sha256=PUT_THE_OUTPUT_HERE
curl -X POST http://127.0.0.1:8000/webhooks/github ^
  -H "Content-Type: application/json" ^
  -H "X-GitHub-Event: push" ^
  -H "X-Hub-Signature-256: %SIG%" ^
  --data "{\"repository\":{\"full_name\":\"ahmed/demo\"}}"

النتيجة المتوقعة: `ok: true`. جرّب تغيّر حرف واحد في body مع نفس التوقيع. لازم يرجع 403. لو رجع 200، يبقى التحقق مش مربوط بالـ raw body الحقيقي.

القياس والـ trade-off

في سيناريو واقعي: endpoint عام يستقبل 1000 request/minute. بدون تحقق مبكر، كل الطلبات تدخل JSON parsing وربما queue أو database. بعد التحقق، الطلبات غير الموقعة تترفض في خطوة CPU صغيرة. في اختبار تقديري، لو 880 طلب مزيف من أصل 1000، التحقق المبكر يقلل الطلبات التي تصل للمنطق الداخلي من 1000 إلى 120. الرقم مش وعد أداء ثابت، لكنه يوضح أين يحدث التوفير.

الـ trade-off هنا إنك زوّدت خطوة أمنية ولازم تدير secret rotation. المكسب: رفض مبكر، حماية من payload tampering، وتقليل الضغط على jobs الداخلية. الخسارة: لو secret اتغير في GitHub وماتغيرش عندك، كل الطلبات الصحيحة هتفشل. لذلك خليه في secret manager أو environment مضبوط، وسجّل عدد 403 في monitoring.

متى لا تستخدم هذه الطريقة

لا تستخدمها لو المصدر لا يدعم توقيع HMAC أو لا يرسل raw payload ثابت. في الحالة دي استخدم API token، mTLS، IP allowlist، أو endpoint خلف API Gateway. كذلك لا تعتمد على HMAC وحده لو عندك خطر replay attacks عالي. أضف timestamp أو nonce لو مزود الخدمة يدعمهم.

مصادر اعتمدت عليها

• GitHub Docs: Validating webhook deliveries
• Python docs: hmac و compare_digest
• FastAPI docs: Request Body

الخطوة التالية

الخطوة التالية: اعمل endpoint `/webhooks/github` محليًا، وخلّي أول اختبار عندك هو طلب بتوقيع غلط. لو لم يرجع 403، لا تربطه بأي automation حقيقي لسه.