المستوى: محترف — دليل تنفيذي لمن يبني ويشغّل خدمات في الإنتاج. لو انت لسه مبتدئ، تحت كل جزء تقني فيه شرح مبسّط بمثال قبل التفاصيل.
اعمل مستقبِل Webhook آمن في FastAPI: تحقق HMAC ومنع إعادة الإرسال
لو أي حد على الإنترنت يقدر يبعت طلب لـ /webhook بتاعك ويأكد إن "الدفع تم"، انت مش بتستقبل أحداث، انت بتنفّذ أوامر مجهولة. الدليل ده يقفل الباب في FastAPI بأربع طبقات: قراءة الجسم الخام، تحقق توقيع HMAC، نافذة زمنية، ومفتاح تكرار.
المشكلة باختصار
الـ Webhook هو إن خدمة تانية (Stripe أو GitHub أو Paymob) تنادي رابطًا عندك لمّا يحصل حدث. المشكلة إن الرابط ده عام. أي طلب POST يوصله، وطبيعته إنه بيغيّر حالة: يشحن محفظة، يفعّل اشتراكًا، يبعت إيميل. من غير تحقق، جملة JSON مزوّرة واحدة تكفي لاستنزاف حسابك.
الفكرة بمثال بسيط ثم علميًا
تخيّل إنك بتستلم ورقة مكتوب فيها "البنك حوّلك 5000". أي حد يقدر يكتب الورقة دي. لكن انت والبنك متفقين على ختم سري. البنك بيحسب رقمًا من (نص الرسالة + الختم) ويحطه آخر الورقة. لو اتغيّر حرف واحد، الرقم ما يطابقش، فتعرف إن الورقة اتلعب فيها. ده بالظبط اللي بيحصل في HMAC.
علميًا: HMAC-SHA256 دالة بتاخد مفتاحًا سريًا والرسالة، وتطلع بصمة طولها 256 بت. من غير المفتاح ما ينفعش تزوّر بصمة صحيحة، وأي تعديل في بايت واحد بيقلب البصمة كلها. المرجع الأصلي هو RFC 2104، وهي نفس الآلية اللي Stripe وGitHub بيوقّعوا بيها الـ webhooks.
الخطوات: بناء المستقبِل
- اقرأ الجسم الخام قبل أي parsing. التوقيع محسوب على البايتات الأصلية بالظبط، لو عملت parse وأعدت التسلسل هتكسر التوقيع من غير ما تعرف.
- احسب HMAC على نفس البايتات بالمفتاح السري المشترك.
- قارن بمقارنة ثابتة الزمن
compare_digestعشان تمنع timing attack. - تأكد إن الطابع الزمني داخل ± 300 ثانية، ده اللي بيمنع إعادة الإرسال.
- خزّن معرّف الحدث وتجاهل المكرر، ده الـ idempotency.
- رُد 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"}
