הגדרת מערכת Webhook עם חתימות HMAC

הגדרת מערכת Webhook עם חתימות HMAC

למה HMAC חיוני לאבטחת Webhook

אתה מקבל התראות webhook מ-Stripe, GitHub או מהמיקרוסרוויס שלך. ללא חתימה, כל נקודת קצה ציבורית חשופה לזיוף. תוקף יכול להתחזות לשולח ולגרום לתשלום מזויף או לשינוי סטטוס. הפתרון הוא HMAC (קוד אימות מסרים מבוסס-האש): מנגנון סימטרי שבו השולח והמקבל חולקים מפתח סודי משותף. יישמנו עשרות אינטגרציות עם חתימות HMAC לפרויקטים של פינטק ומסחר אלקטרוני — זהו תקן אבטחה שנבחן בקרב על יותר מ-50 פרויקטים.

בפרויקט אחד, לקוח נתקל במתקפה: webhook מיירט על אישור תשלום הושמע שוב שעה לאחר מכן. המערכת עיבדה אותו שוב, וחייבה את הלקוח פעמיים. לאחר יישום HMAC עם חותמת זמן ואידמפוטנטיות, תקריות כאלה פסקו. הפגיעות נסגרה — חוסכים עד 5 שעות ניפוי באגים בחודש, שמתורגם לכ-$2,500 חיסכון בשעות פיתוח בשנה.

יישום חתימות HMAC היה משנה משחק עבור האבטחה שלנו. לא הייתה לנו אף תקריה מאז.

— לקוח, סטארטאפ פינטק

בעיות שאנחנו פותרים

• זיוף Webhook — תוקף שולח שינוי תשלום או סטטוס מזויף. HMAC מבטיח את האותנטיות של השולח. לפי הסטטיסטיקות שלנו, 70% מאינטגרציות ה-webhook חסרות חתימה מלכתחילה.
• מתקפת השמעה חוזרת — בקשה שמיירטה יכולה להישלח שוב. חותמת זמן + אימות מבוסס-זמן (בדרך כלל 5 דקות) חוסמים מתקפות כאלה. בלעדיה, חלון הפגיעות הוא אינסופי.
• מתקפת תזמון — השוואת מחרוזות רגילה (==) לוקחת זמנים שונים בהתאם להתאמה. אנו משתמשים ב-hmac.compare_digest() — השוואה בזמן קבוע העמידה בפני מתקפות תזמון. בביקורת אחת, גילינו ש-80% מהפרויקטים לא משתמשים בהשוואה בזמן קבוע.

אנו גם מיישמים אידמפוטנטיות כדי למנוע עיבוד כפול במשלוח חוזר.

השוואת שיטות חתימת Webhook

HMAC מנצח במהירות ובפשטות: פעולת הצפנה אחת של SHA-256 מהירה פי 10 מחתימת RSA. עבור רוב התרחישים, זו הבחירה האופטימלית. מבחינת ביצועים, HMAC-SHA256 חותם ב-~0.02 אלפיות השנייה ומאמת ב-~0.02 אלפיות השנייה, ומייצר חתימה של 64 בתים בלבד.

כיצד ליישם אימות חתימת HMAC

הגנה מפני מתקפות השמעה חוזרת

הכלי המרכזי הוא חותמת זמן. כלול אותה במסר החתום, ובצד המקבל בדוק את ההפרש. אם הבקשה "ישנה" יותר מ-5 דקות, דחה אותה. גם אם תוקף מיירט את החתימה, הוא לא יכול לעשות בה שימוש חוזר לאחר שתוקף החלון פג. בנוסף, שמור את חותמת הזמן האחרונה וחסום הגשות חוזרות עם אותו ערך.

למה חייבים לחתום על גוף הבקשה הגולמי

גוף הבקשה נחתם לפני העיבוד — כבתים גולמיים. אי אפשר לנתח JSON לפני בדיקת החתימה, כי מפענחים שונים משנים את העיצוב (רווחים, סדר מפתחות). בדוגמה, אנו לוקחים request.get_data() — בתים גולמיים. אם אתה משתמש ב-request.get_json(), החתימה לא תתאים גם אם המפתח נכון. טעות זו נמצאת ב-80% מהפרויקטים שמגיעים אלינו לביקורת.

טעויות יישום נפוצות

• אימות לאחר ניתוח JSON — החתימה מחושבת מהנתונים הגולמיים. השתמש תמיד ב-request.get_data().
• ללא השוואה בזמן קבוע — == רגילה הופכת את המערכת לפגיעה למתקפות תזמון. רק hmac.compare_digest().
• התעלמות מהגנה מפני השמעה חוזרת — ללא חותמת זמן, החתימה היא סטטית; חבילה שמיירטה יכולה לשמש שוב ללא הגבלת זמן.
• מפתח סודי קצר מדי — השתמש בלפחות 32 בתים, שנוצר באמצעות secrets.
• ללא אידמפוטנטיות — בניסיון חוזר (פסק זמן, שגיאה) הבקשה עשויה לעובד פעמיים. יישם מפתח אידמפוטנטיות (לדוגמה, X-Webhook-ID).

קוד יצירת ואימות חתימה

יצירת חתימה בעת שליחת Webhook

import hmac
import hashlib
import json
import requests


def send_webhook(url: str, payload: dict, secret: str):
    body = json.dumps(payload, separators=(',', ':'))
    timestamp = int(time.time())
    # Подпись включает timestamp для защиты от replay attacks
    message = f"{timestamp}.{body}"
    signature = hmac.new(
        secret.encode(),
        message.encode(),
        hashlib.sha256
    ).hexdigest()
    response = requests.post(
        url,
        data=body,
        headers={
            'Content-Type': 'application/json',
            'X-Webhook-Timestamp': str(timestamp),
            'X-Webhook-Signature': f"sha256={signature}",
            'X-Webhook-ID': str(uuid.uuid4()),
        },
        timeout=10
    )
    return response

אימות חתימה בצד המקבל

import hmac
import hashlib
import time

def verify_webhook_signature(request) -> bool:
    secret = os.environ['WEBHOOK_SECRET']
    # Извлечь из заголовков
    timestamp = request.headers.get('X-Webhook-Timestamp')
    received_sig = request.headers.get('X-Webhook-Signature', '')
    if not timestamp or not received_sig:
        return False
    # Защита от replay attack: не принимать события старше 5 минут
    if abs(time.time() - int(timestamp)) > 300:
        return False
    # Вычислить ожидаемую подпись
    body = request.get_data()  # raw bytes, до парсинга!
    message = f"{timestamp}.{body.decode()}".encode()
    expected_sig = "sha256=" + hmac.new(
        secret.encode(),
        message,
        hashlib.sha256
    ).hexdigest()
    # Constant-time comparison для защиты от timing attack
    return hmac.compare_digest(expected_sig, received_sig)

@app.route('/webhooks/payments', methods=['POST'])
def payment_webhook():
    if not verify_webhook_signature(request):
        return jsonify({'error': 'Invalid signature'}), 401
    # Безопасно обрабатывать payload
    event = request.get_json()
    process_payment_event(event)
    return jsonify({'status': 'ok'})

לוגיקת ניסיונות חוזרים ואידמפוטנטיות

class WebhookDelivery:
    MAX_ATTEMPTS = 5
    RETRY_DELAYS = [10, 30, 120, 600, 3600]  # секунды между попытками

    def deliver_with_retry(self, webhook_id: str, url: str, payload: dict, secret: str):
        for attempt, delay in enumerate(self.RETRY_DELAYS):
            try:
                response = send_webhook(url, payload, secret)
                if response.status_code < 300:
                    db.mark_delivered(webhook_id)
                    return True
                db.log_attempt(webhook_id, attempt + 1, response.status_code)
            except requests.exceptions.Timeout:
                db.log_attempt(webhook_id, attempt + 1, error='timeout')
            if attempt < len(self.RETRY_DELAYS) - 1:
                time.sleep(delay)
        db.mark_failed(webhook_id)
        return False


def handle_webhook_idempotent(webhook_id: str, handler_fn):
    """Предотвратить двойную обработку при retry"""
    if db.is_processed(webhook_id):
        return  # Уже обработано
    with db.transaction():
        db.mark_processing(webhook_id)
        handler_fn()
        db.mark_processed(webhook_id)

תהליך האינטגרציה שלנו עם Webhook והתוצרים

איך אנחנו עושים את זה

1. ניתוח — דנים בתרחישים: אילו מערכות חיצוניות שולחות webhooks, אילו נתונים מועברים, האם יש צורך בלוגיקת ניסיונות חוזרים.
2. עיצוב — בוחרים פורמט כותרת (כמו Stripe, GitHub או מותאם אישית), מגדירים חלון זמן מקובל, מחליטים על אחסון מפתח אידמפוטנטיות (Redis, PostgreSQL).
3. יישום — כתיבת middleware לאימות, מטפל בשידור חוזר (משלוח מחדש), מטפל אירועים אידמפוטנטי.
4. בדיקות — בדיקות אינטגרציה עם בקשות מזויפות (חתימות תקינות, לא תקינות, חותמות זמן שפגו, הגשות חוזרות).
5. פריסה וניטור — הגדרת התראות על שגיאות אימות, רישום כל webhook עם מידע מטא.

מה כלול בעבודה

• פיתוח middleware לאימות חתימת HMAC
• עיצוב תוכנית הגנה מפני מתקפות השמעה חוזרת עם חותמת זמן
• יישום אידמפוטנטיות באמצעות idempotency-key
• אינטגרציה עם שירותים קיימים (Stripe, GitHub, שערי תשלום)
• הגדרת לוגיקת ניסיונות חוזרים וניטור
• תיעוד על הליך החלפת מפתחות ופרמטרים
• הדרכה לצוות שלך (עד 3 שעות)

התוצרים שלנו כוללים תיעוד מקיף, גישה מאובטחת למאגר, הדרכת צוות ותמיכה לאחר פריסה למשך שבועיים.

לוחות זמנים ועלות הפרויקט

יישום מערכת חתימת HMAC מקצה לקצה עם מנגנוני ניסיונות חוזרים ואידמפוטנטיות אורך 1 עד 3 ימי עבודה בהתאם למורכבות האינטגרציה. עלויות טיפוסיות מתחילות מ-$500 להגדרה בסיסית, עם אינטגרציות מתקדמות שנעות עד $3,000. קבל ייעוץ — צור קשר, ונעריך את הפרויקט שלך. ההשקעה מחזירה את עצמה לעתים קרובות תוך חודשים על ידי מניעת תקריות אבטחה יקרות.