Dashboard + FastAPI
وبهوک ها: راه اندازی
ساخت endpoint در داشبورد و نمونه کامل دریافت وبهوک در FastAPI.
داشبورد
- صفحه وبهوک ها از داشبورد باز می شود.
- URL عمومی HTTPS در فرم وبهوک ثبت می شود.
- رویدادهای مورد نیاز انتخاب می شوند؛ حالت همه رویدادها هم قابل نگه داشتن است.
- secret نمایش داده شده در تنظیمات امن محصول ذخیره می شود.
- دکمه تست، event آزمایشی
test.pingرا ارسال می کند.
جای تصویر: ساخت endpoint وبهوک و نمایش secret
نمونه FastAPI
import hashlib
import hmac
import time
from fastapi import FastAPI, Header, Request, HTTPException
app = FastAPI()
WEBHOOK_SECRET = "whsec_..."
def parse_signature(header: str):
timestamp = None
signatures = []
for part in header.split(","):
if "=" not in part:
continue
key, value = part.strip().split("=", 1)
if key == "t":
timestamp = value
elif key == "v1":
signatures.append(value)
return timestamp, signatures
@app.post("/webhooks/navidaa")
async def navidaa_webhook(request: Request, navidaa_signature: str = Header(...)):
raw_body = await request.body()
timestamp, signatures = parse_signature(navidaa_signature)
if not timestamp or not signatures:
raise HTTPException(status_code=400, detail="bad signature")
if abs(int(time.time()) - int(timestamp)) > 300:
raise HTTPException(status_code=400, detail="stale signature")
signed_payload = f"{timestamp}.".encode() + raw_body
expected = hmac.new(
WEBHOOK_SECRET.encode(),
signed_payload,
hashlib.sha256,
).hexdigest()
if not any(hmac.compare_digest(expected, sig) for sig in signatures):
raise HTTPException(status_code=400, detail="invalid signature")
payload = await request.json()
event_id = payload["event_id"]
event_type = payload["event"]
# Store event_id to process repeated deliveries idempotently.
return {"ok": True}
تست
برای تست، endpoint در داشبورد ساخته می شود و دکمه تست یک payload آزمایشی ارسال می کند. اگر پاسخ ۲xx برنگردد، رکورد delivery در داشبورد قابل مشاهده است و تلاش دوباره طبق مقررات انجام می شود.