اعمل مستقبِل Webhook آمن في FastAPI: تحقق HMAC ومنع إعادة الإرسال

المستوى: محترف — دليل تنفيذي لمن يبني ويشغّل خدمات في الإنتاج. لو انت لسه مبتدئ، تحت كل جزء تقني فيه شرح مبسّط بمثال قبل التفاصيل.

اعمل مستقبِل Webhook آمن في FastAPI: تحقق HMAC ومنع إعادة الإرسال

لو أي حد على الإنترنت يقدر يبعت طلب لـ /webhook بتاعك ويأكد إن "الدفع تم"، انت مش بتستقبل أحداث، انت بتنفّذ أوامر مجهولة. الدليل ده يقفل الباب في FastAPI بأربع طبقات: قراءة الجسم الخام، تحقق توقيع HMAC، نافذة زمنية، ومفتاح تكرار.

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

الـ Webhook هو إن خدمة تانية (Stripe أو GitHub أو Paymob) تنادي رابطًا عندك لمّا يحصل حدث. المشكلة إن الرابط ده عام. أي طلب POST يوصله، وطبيعته إنه بيغيّر حالة: يشحن محفظة، يفعّل اشتراكًا، يبعت إيميل. من غير تحقق، جملة JSON مزوّرة واحدة تكفي لاستنزاف حسابك.

الفكرة بمثال بسيط ثم علميًا

تخيّل إنك بتستلم ورقة مكتوب فيها "البنك حوّلك 5000". أي حد يقدر يكتب الورقة دي. لكن انت والبنك متفقين على ختم سري. البنك بيحسب رقمًا من (نص الرسالة + الختم) ويحطه آخر الورقة. لو اتغيّر حرف واحد، الرقم ما يطابقش، فتعرف إن الورقة اتلعب فيها. ده بالظبط اللي بيحصل في HMAC.

علميًا: HMAC-SHA256 دالة بتاخد مفتاحًا سريًا والرسالة، وتطلع بصمة طولها 256 بت. من غير المفتاح ما ينفعش تزوّر بصمة صحيحة، وأي تعديل في بايت واحد بيقلب البصمة كلها. المرجع الأصلي هو RFC 2104، وهي نفس الآلية اللي Stripe وGitHub بيوقّعوا بيها الـ webhooks.

الخطوات: بناء المستقبِل

1. اقرأ الجسم الخام قبل أي parsing. التوقيع محسوب على البايتات الأصلية بالظبط، لو عملت parse وأعدت التسلسل هتكسر التوقيع من غير ما تعرف.
2. احسب HMAC على نفس البايتات بالمفتاح السري المشترك.
3. قارن بمقارنة ثابتة الزمن compare_digest عشان تمنع timing attack.
4. تأكد إن الطابع الزمني داخل ± 300 ثانية، ده اللي بيمنع إعادة الإرسال.
5. خزّن معرّف الحدث وتجاهل المكرر، ده الـ idempotency.
6. رُد 200 بسرعة وادفع المعالجة للخلفية.

import hmac, hashlib, time, json
from fastapi import FastAPI, Request, HTTPException, BackgroundTasks

app = FastAPI()
WEBHOOK_SECRET = b"whsec_من_متغيرات_البيئة"   # مش داخل الكود
TOLERANCE = 300                               # ثانية (النافذة ضد الإعادة)
seen_ids: set[str] = set()                    # في الإنتاج: Redis مع TTL

def verify(sig_header: str, body: bytes):
    try:
        parts = dict(p.split("=", 1) for p in sig_header.split(","))
        ts = int(parts["t"]); received = parts["v1"]
    except (ValueError, KeyError):
        raise HTTPException(400, "bad signature header")
    if abs(time.time() - ts) > TOLERANCE:       # الطابع الزمني قديم؟
        raise HTTPException(401, "stale timestamp")
    signed = f"{ts}.".encode() + body           # نفس صيغة التوقيع عند المصدر
    expected = hmac.new(WEBHOOK_SECRET, signed, hashlib.sha256).hexdigest()
    if not hmac.compare_digest(expected, received):   # مقارنة ثابتة الزمن
        raise HTTPException(401, "signature mismatch")

@app.post("/webhook")
async def webhook(request: Request, bg: BackgroundTasks):
    body = await request.body()                 # الجسم الخام قبل أي parsing
    verify(request.headers.get("X-Signature", ""), body)
    event = json.loads(body)
    if event["id"] in seen_ids:                 # idempotency
        return {"status": "duplicate ignored"}
    seen_ids.add(event["id"])
    bg.add_task(process_event, event)           # رد سريع، المعالجة بالخلفية
    return {"status": "accepted"}

ليه compare_digest مش == عادية؟

المقارنة العادية بتوقف عند أول حرف مختلف. الفرق البسيط ده في الزمن (نانوثواني) بيسرّب معلومة تخلّي المهاجم يخمّن التوقيع الصحيح حرفًا حرفًا. compare_digest بتقارن في زمن ثابت مهما كان موضع الاختلاف. الافتراض هنا إن المهاجم يقدر يقيس زمن ردك بدقة، وده وارد فعلاً في خدمة عامة تحت ضغط.

منع إعادة الإرسال والتكرار

التوقيع الصحيح لوحده مش كفاية. لو مهاجم سجّل طلبًا صحيحًا وأعاد إرساله بعد ساعة، التوقيع لسه صحيح. الحل طبقتين فوق التوقيع: نافذة زمنية ومفتاح تكرار.

سيناريو واقعي: مزود الدفع بعت charge.succeeded، الشبكة قطعت الرد، فالمزود أعاد الإرسال 3 مرات. من غير idempotency هتشحن المحفظة 3 مرات وتخصم من نفسك فرق حقيقي. مع مفتاح تكرار، أول طلب بيتعالج والباقي بيترجعله 200 من غير أي تنفيذ.

خُد بالك: Stripe مثلاً بيعيد المحاولة حتى 3 أيام لو خدمتك ما ردّتش 2xx. يعني الـ TTL بتاع مخزن المعرّفات لازم يغطّي المدة دي، مش دقيقة ولا اتنين.

الـ trade-offs

تكلفة التحقق: HMAC-SHA256 على جسم بحجم 2KB بياخد عشرات الميكروثانية، مهملة عمليًا. المقارنة ثابتة الزمن والنافذة الزمنية تكلفتهم صفر تقريبًا. مفتاح التكرار بيكلّف قراءة Redis حوالي 1ms وشوية تخزين. الـ trade-off هنا واضح: كل ما زوّدت الـ TTL، أمان أعلى ضد الإعادة المتأخرة، مقابل تخزين أكبر. الافتراض إن معدل الأحداث عندك ≤ بضع آلاف/دقيقة؛ فوق كده استخدم Redis مع سياسة انتهاء بدل الـ set في الذاكرة.

التحقق من أنه يعمل

1. ابعت طلبًا بتوقيع صحيح وطابع زمني حديث، لازم يرجّع 200 وaccepted.
2. غيّر بايت واحد في الجسم، لازم يرجّع 401.
3. ابعت طابعًا زمنيًا قديمًا (t-900s)، لازم يرجّع 401.
4. أعد إرسال نفس المعرّف، لازم يرجّع 200 مع duplicate من غير تنفيذ مزدوج.

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

لو الاتصال داخلي بين خدماتك على شبكة خاصة فيها mTLS شغّال، توقيع HMAC ممكن يبقى تكرار بلا فائدة. ولو المزود بيوفّر توقيعًا جاهزًا بـ OAuth أو mTLS مخصص، استخدم آليته بدل ما تخترع واحدة. ولأحداث لا تغيّر حالة (قراءة فقط)، مخزن التكرار زيادة مش محتاجها.

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

افتح معالج الـ webhook عندك دلوقتي وشوف: بتتحقق على الجسم الخام request.body() ولا على JSON بعد ما عملتله parse وإعادة تسلسل؟ لو الثانية، توقيعك بيتكسر بصمت وانت فاكر نفسك محمي. صلّح دي الأول، وبعدها ضيف compare_digest ثم نافذة الوقت.

المصادر

• Stripe — Verify webhook signatures: https://stripe.com/docs/webhooks/signatures
• GitHub — Validating webhook deliveries: https://docs.github.com/en/webhooks/using-webhooks/validating-webhook-deliveries
• RFC 2104 — HMAC: Keyed-Hashing for Message Authentication: https://www.rfc-editor.org/rfc/rfc2104
• Python docs — hmac.compare_digest: https://docs.python.org/3/library/hmac.html
• FastAPI — Request body و Background Tasks: https://fastapi.tiangolo.com/
• OWASP Cheat Sheet Series — أمن الـ Webhooks وSSRF: https://cheatsheetseries.owasp.org/