Webhook সঠিকভাবে: HMAC signature যাচাই ও replay ঠেকান

Webhook হলো এমন request যা অন্য সেবা আপনাকে পাঠায়: "একটি order-এর টাকা এসেছে", "build শেষ হয়েছে"। আপনার endpoint প্রকাশ্য হতে হয় যাতে তারা পৌঁছাতে পারে, অর্থাৎ ইন্টারনেটের যে কেউ সেখানে POST করতে পারে। Body অন্ধভাবে বিশ্বাস করলে আক্রমণকারী ভুয়া "order.paid" পাঠিয়ে বিনামূল্যে পণ্য পেতে পারে। প্রতিরক্ষা হলো signature যাচাই। নিচের কোড ও ফল প্রকাশের আগে Python 3-এ চালানো হয়েছে।

Signing কীভাবে কাজ করে

১. আপনি ও প্রেরক একটি গোপন key (secret) ভাগ করেন (webhook নিবন্ধনের সময় পাওয়া যায়)। ২. প্রেরক secret দিয়ে request-এর HMAC-SHA256 হিসাব করে header-এ পাঠায়। ৩. আপনি যা পেয়েছেন তার ওপর একই HMAC হিসাব করেন। দুটি মিললে বোঝা যায় request এসেছে এমন কারও কাছ থেকে যে secret জানে, এবং পথে বদলায়নি।

HMAC হলো key-যুক্ত hash। Secret ছাড়া আক্রমণকারী পরিবর্তিত body-র জন্য বৈধ signature বানাতে পারে না।

একটি চালানো যায় এমন উদাহরণ

webhook.py নামে সংরক্ষণ করুন (প্রেরকের sign ও গ্রাহকের যাচাই একই ফাংশন ব্যবহার করে):

import hashlib
import hmac
import json
import time

TOLERANCE = 300  # seconds


def sign(secret: bytes, timestamp: int, body: bytes) -> str:
    """Signature over 'timestamp.body' so the time cannot be swapped."""
    message = str(timestamp).encode() + b"." + body
    return hmac.new(secret, message, hashlib.sha256).hexdigest()


def build_headers(secret: bytes, body: bytes, now=None) -> dict:
    ts = int(now if now is not None else time.time())
    return {"X-Timestamp": str(ts), "X-Signature": "sha256=" + sign(secret, ts, body)}


class Receiver:
    def __init__(self, secret: bytes, now=time.time):
        self.secret = secret
        self.now = now
        self.seen = set()  # processed event ids (use a database in real life)
        self.processed = []

    def handle(self, headers: dict, raw_body: bytes):
        try:
            ts = int(headers["X-Timestamp"])
            received = headers["X-Signature"].removeprefix("sha256=")
        except (KeyError, ValueError):
            return 400, "missing or bad headers"

        if abs(self.now() - ts) > TOLERANCE:
            return 400, "timestamp outside tolerance"

        expected = sign(self.secret, ts, raw_body)
        if not hmac.compare_digest(expected, received):
            return 401, "bad signature"

        event = json.loads(raw_body)  # parse only AFTER verification
        if event["id"] in self.seen:
            return 200, "duplicate ignored"
        self.seen.add(event["id"])
        self.processed.append(event["type"])
        return 200, "ok"


if __name__ == "__main__":
    secret = b"whsec_demo_only_not_real"
    clock = [1_800_000_000]
    rx = Receiver(secret, now=lambda: clock[0])
    body = json.dumps({"id": "evt_1", "type": "order.paid"}).encode()
    good = build_headers(secret, body, now=clock[0])

    print("valid            :", rx.handle(good, body))
    print("same event again :", rx.handle(good, body))
    tampered = body.replace(b"order.paid", b"order.free")
    print("tampered body    :", rx.handle(good, tampered))
    print("wrong secret     :", rx.handle(build_headers(b"other", body, now=clock[0]), body))
    clock[0] += 3600
    print("replayed 1h later:", rx.handle(good, body))
    print("no headers       :", rx.handle({}, body))
    print("processed events :", rx.processed)

python3 webhook.py-এর ফল:

valid            : (200, 'ok')
same event again : (200, 'duplicate ignored')
tampered body    : (401, 'bad signature')
wrong secret     : (401, 'bad signature')
replayed 1h later: (400, 'timestamp outside tolerance')
no headers       : (400, 'missing or bad headers')
processed events : ['order.paid']

প্রতিটি ক্ষেত্র কী দেখায়

• Valid: গৃহীত ও একবার প্রক্রিয়াজাত।
• একই event আবার: সেবাদাতারা retry করে, তাই ডুপ্লিকেট স্বাভাবিক। আমরা event id মনে রাখি এবং দুবার প্রক্রিয়া না করে 200 দিই (idempotency key-এর একই ধারণা)।
• Body বদলানো: একটি শব্দ বদলাতেই signature মেলে না, 401।
• ভুল secret: secret ছাড়া আক্রমণকারী signature করতে পারে না, 401।
• এক ঘণ্টা পরে replay: ধরা পড়া, সত্যিই স্বাক্ষরিত request সময়সীমার পর অকেজো, 400।
• Header নেই: কোনো কাজের আগেই প্রত্যাখ্যাত।

যে খুঁটিনাটি গুরুত্বপূর্ণ

Raw byte যাচাই করুন। যে হুবহু body byte পেয়েছেন, তার ওপর HMAC হিসাব করুন। Framework আগে JSON parse করে আপনি আবার serialise করলে spacing বা key-র ক্রম বদলে signature মিলবে না। Raw body পড়ুন, যাচাই করুন, তারপর parse করুন (কোডের মন্তব্য দেখুন)।

Constant-time তুলনা। == নয়, hmac.compare_digest ব্যবহার করুন। সাধারণ তুলনা প্রথম অমিল অক্ষরেই থামতে পারে, আর সময়ের পার্থক্য সঠিক signature সম্পর্কে তথ্য ফাঁস করতে পারে।

Timestamp-ও sign করুন। আমরা timestamp.body sign করি। Timestamp signature-এর বাইরে থাকলে আক্রমণকারী পুরোনো request নতুন timestamp দিয়ে replay করতে পারত। test_swapped_timestamp প্রমাণ করে এটি বদলালে signature ভেঙে যায়।

Timestamp সহনসীমা। কয়েক মিনিটের চেয়ে পুরোনো (বা ভবিষ্যতের) request প্রত্যাখ্যান করুন। এতে ধরা পড়া request কত সময় replay করা যাবে তা সীমিত হয়। এর জন্য আপনার server-এর ঘড়ি মোটামুটি সঠিক থাকতে হবে।

Event id দিয়ে ডুপ্লিকেট বাদ দিন। বৈধ signature প্রমাণ করে কে পাঠিয়েছে, এটি যে প্রথমবার তা নয়। প্রক্রিয়া করা id সংরক্ষণ করুন (database-এ, unique constraint সহ) এবং পুনরাবৃত্তি উপেক্ষা করুন। আমাদের in-memory set কেবল ডেমোর জন্য।

Secret গোপন রাখুন। এটি environment variable বা secrets manager-এ রাখুন, repository-তে কখনো নয়। Rotation সমর্থন করুন: বদলানোর সময় অল্প সময় পুরোনো ও নতুন দুই secret-এর signature গ্রহণ করুন।

Test

test_webhook.py নামে সংরক্ষণ করুন:

import json
import unittest

from webhook import Receiver, build_headers

SECRET = b"s3cret"
NOW = 1_800_000_000


def make(event_id="e1"):
    body = json.dumps({"id": event_id, "type": "t"}).encode()
    return body, build_headers(SECRET, body, now=NOW)


class WebhookTests(unittest.TestCase):
    def setUp(self):
        self.rx = Receiver(SECRET, now=lambda: NOW)

    def test_valid(self):
        body, h = make()
        self.assertEqual(self.rx.handle(h, body)[0], 200)
        self.assertEqual(self.rx.processed, ["t"])

    def test_tampered_body(self):
        body, h = make()
        self.assertEqual(self.rx.handle(h, body + b" ")[0], 401)
        self.assertEqual(self.rx.processed, [])

    def test_swapped_timestamp(self):
        body, h = make()
        h["X-Timestamp"] = str(NOW + 10)
        self.assertEqual(self.rx.handle(h, body)[0], 401)

    def test_old_timestamp(self):
        body = b'{"id":"e","type":"t"}'
        h = build_headers(SECRET, body, now=NOW - 1000)
        self.assertEqual(self.rx.handle(h, body)[0], 400)

    def test_duplicate_event(self):
        body, h = make()
        self.rx.handle(h, body)
        self.assertEqual(self.rx.handle(h, body), (200, "duplicate ignored"))
        self.assertEqual(self.rx.processed, ["t"])


if __name__ == "__main__":
    unittest.main()

$ python3 -m unittest -v
Ran 5 tests in 0.002s

OK

বাস্তবে webhook গ্রহণ

• দ্রুত সাড়া দিন। যাচাই করুন, event সংরক্ষণ করুন, 2xx ফেরত দিন, আর ধীর কাজ background job-এ করুন। প্রেরকরা প্রায়ই কয়েক সেকেন্ড পর timeout করে retry করে।
• সঠিক status code দিন। খারাপ request-এ 401 বা 400। 5xx প্রেরককে পরে আবার চেষ্টা করতে বলে, নিজের সাময়িক ব্যর্থতায় যা চান।
• ক্রমহীন event সামলান। পৌঁছানোর ক্রম সাধারণত নিশ্চিত নয়। "created" আগে আসবে ধরে না নিয়ে object-এর বর্তমান অবস্থা দেখুন।
• ডিবাগের জন্য ব্যর্থতা log করুন (secret বা পুরো signature log না করে)।
• HTTPS ব্যবহার করুন যাতে পথে body পড়া না যায়। Signature অখণ্ডতা ও উৎস প্রমাণ করে, গোপনীয়তা নয়।

সাধারণ ভুল

• "URL গোপন" ভেবে যাচাই বাদ দেওয়া। Log ও টুলে URL ফাঁস হয়।
• == দিয়ে signature তুলনা।
• Raw body-র বদলে আবার serialise করা body যাচাই।
• Timestamp না দেখা, যা replay-এর সুযোগ দেয়।
• যাচাইয়ের আগে প্রক্রিয়া করা, বা যাচাইয়ের আগে অবিশ্বস্ত JSON parse করা।
• সাড়া দেওয়ার আগে ধীর কাজ করা, যা timeout ও ডুপ্লিকেট delivery ঘটায়।

এই উদাহরণের সীমা

প্রতিটি সেবাদাতার নিজস্ব header নাম ও signing ফরম্যাট আছে (কেউ শুধু body sign করে, কেউ URL যোগ করে, কেউ ভিন্ন encoding ব্যবহার করে)। সবসময় সেবাদাতার ডকুমেন্টেশন অনুসরণ করুন এবং থাকলে তাদের অফিসিয়াল যাচাই লাইব্রেরি ব্যবহার করুন। এই উদাহরণ সাধারণ প্যাটার্ন দেখায়, কোনো নির্দিষ্ট সেবাদাতার হুবহু স্কিম নয়।

মূল কথা: প্রতিটি webhook-কে অবিশ্বস্ত ধরুন যতক্ষণ না raw body ও timestamp-এর ওপর হিসাব করা HMAC signature constant-time তুলনায় মেলে; তারপর timestamp সীমা দেখুন এবং event id দিয়ে ডুপ্লিকেট বাদ দিন।

---

Backend সিরিজ, পর্ব ৩ (নির্ভরযোগ্য API): 2 নম্বর (মোট 5)। আগের লেখা: Idempotency key: API retry নিরাপদ করুন, দুবার চার্জ রুখুন.