Webhook ของ AI Agent: ลายเซ็นผ่านแล้ว ยังต้องตรวจเวลาที่เซ็นไว้

บทเรียน Security · แบบจำลองเพื่อการเรียนรู้ — ตัวอย่าง Python และชุดทดสอบยัง NOT RUN; การทดสอบ UI จริงยัง UNKNOWN

คำตอบสั้น: ลายเซ็นผูกกับกุญแจ แต่ยังต้องจำกัดอายุคำขอ

Webhook ที่ลายเซ็นถูกต้องพิสูจน์ได้เพียงว่าเนื้อหาที่ตรวจตรงกับกุญแจที่ใช้ลงนาม ภายใต้รูปแบบของผู้ให้บริการนั้น ถ้าผู้โจมตีคัดลอกคำขอที่ถูกต้องไว้ เขาอาจส่งสำเนาเดิมซ้ำในภายหลังได้ ระบบรับจึงควรตรวจ เวลาของความพยายามส่ง ที่อยู่ในข้อมูลซึ่งลายเซ็นครอบคลุมด้วย แล้วรับเฉพาะเมื่อเวลานั้นอยู่ในช่วงที่ยอมรับได้เมื่อเทียบกับนาฬิกา UTC ที่เชื่อถือได้ของฝั่งรับ

ทักษะในบทนี้คือสร้างด่านรับเข้าแบบจำลองที่ตรวจไบต์เดิมของ body และ timestamp ที่เซ็นไว้ ก่อนอนุญาตให้ขั้นตอนถัดไปพิจารณางาน ชื่อผล ACCEPT_ENVELOPE จึงหมายถึง “ซองผ่านการตรวจขนส่งที่กำหนดไว้” เท่านั้น ไม่ได้หมายถึงอนุญาตการกระทำ ตรวจโครงสร้าง JSON กันการส่งซ้ำ บันทึกงานถาวร หรือเริ่ม Agent แล้ว ตัวอย่างนี้ใช้ HMAC รูปแบบที่เราแต่งขึ้นเพื่อเรียนรู้ ไม่เข้ากันกับ Standard Webhooks, Svix, GitHub หรือ Stripe

เมื่อจบบทนี้ คุณจะอธิบายผลรับเข้าได้โดยไม่ปนสามเรื่อง

ผลลัพธ์ที่ต้องทำได้คืออ่านตัวอย่างหนึ่งรายการแล้วตัดสินใจอย่างมีเหตุผลว่าจะ ACCEPT_ENVELOPE, REJECT หรือ HOLD พร้อมบอกเหตุผลที่ตรวจสอบย้อนหลังได้ คุณจะชี้ได้ว่า HMAC คำนวณจากข้อความอะไร เวลาใดจึงเป็นเวลาที่พิสูจน์แล้ว และส่วนใดของการรับ webhook ยังไม่ได้รับการพิสูจน์

Freshness คืออายุของ delivery attempt; event age คือเวลาของเหตุการณ์ทางธุรกิจ; idempotency คือการกันผลข้างเคียงซ้ำ ทั้งสามเกี่ยวข้องกันแต่ไม่ใช่สิ่งเดียวกัน Standard Webhooks 1.0.0 แยก timestamp ของ attempt และ ID สำหรับกันซ้ำไว้คนละบทบาท (อ้างอิงท้ายเรื่อง)

จากแชต LLM สู่ Agent: webhook เป็นทางเข้า ไม่ใช่คำสั่งให้ลงมือ

ในแชต LLM เรามักส่งข้อความเข้าโมเดลแล้วอ่านข้อความตอบ แต่ Agent ที่ทำงานต่อเนื่องมีทางเข้าอื่นด้วย เช่นระบบภายนอกแจ้งว่าเอกสารพร้อมแล้วหรือวิดีโอเรนเดอร์เสร็จ webhook คือ HTTP request ที่ผู้ส่งเรียกมายัง endpoint ของเราเพื่อแจ้งเหตุการณ์ หากเอา request นั้นไปใส่คิวงาน Agent ทันที โดยไม่ตรวจที่มาและอายุของความพยายามส่ง ผู้รับอาจยอมรับสำเนาที่ถูกบันทึกไว้ หรือเชื่อ payload ที่ถูกแก้ไขระหว่างทาง

ปัญหาที่ด่าน timestamp แก้ และปัญหาที่มันแก้ไม่ได้

สมมติผู้ให้บริการส่งข้อความลงนามว่าเหตุการณ์หนึ่งเกิดขึ้น ผู้รับตรวจลายเซ็นได้ แล้วนำไปทำงาน หากคำขอเดิมถูกดักหรือคัดลอก ผู้ส่งเดิมที่ครอบครองกุญแจยังเป็นผู้สร้างลายเซ็นนั้นจริง การตรวจ HMAC อย่างเดียวจึงไม่แยกคำขอที่เพิ่งส่งกับสำเนาเก่าได้ การผูก timestamp ของความพยายามส่งเข้าไปในการคำนวณลายเซ็น และตรวจว่ามันไม่เก่าเกินหรืออยู่ไกลในอนาคตเกินไป ทำให้สำเนาเดิมมีอายุเพิ่มขึ้นเมื่อเวลาผ่านไป

ด่านนี้ลดช่วงเวลาที่ envelope เก่าซึ่งยังมีลายเซ็นถูกต้องจะผ่านได้ แต่ไม่ทำให้ replay เป็นไปไม่ได้ทั้งหมด ภายในหน้าต่างที่อนุญาต สำเนาเดิมยังอาจผ่านซ้ำได้ และอาจผ่านได้หลายครั้งในแบบจำลอง เพราะเราไม่ได้เก็บรหัสที่รับแล้วไว้ในฐานข้อมูล การป้องกันผลข้างเคียงซ้ำจึงยังต้องมี idempotency key หรือ receipt ที่บันทึกถาวรในอีกชั้นหนึ่ง บทเรียนเรื่อง receipt-before-ACK อธิบายการบันทึกหลักฐานรับงานก่อนตอบรับคิวไว้ที่ รับ receipt ก่อน ACK เมื่อ ACK อาจสูญหาย

แผนผังรับเข้า: ตรวจไบต์ ตรวจลายเซ็น ตรวจเวลา แล้วค่อยส่งต่อ

แบบจำลองแบ่งหน้าที่ออกเป็นห้าช่วงต่อกัน ขั้นแรกอ่าน body เป็น bytes และจำกัดขนาดก่อนเก็บหรือประมวลผล ขั้นที่สองตรวจ syntax ของ id, timestamp และ signature ให้ตรงรูปแบบจำกัดที่เลือก ขั้นที่สามประกอบข้อความที่ลงนามจาก id, timestamp แบบข้อความเดิม และ raw body ขั้นที่สี่คำนวณ HMAC-SHA256 และเปรียบเทียบ digest ขั้นที่ห้าจึงแปลง timestamp ที่ผ่านการยืนยันแล้วเป็นจำนวนเต็ม เทียบกับเวลารับที่เชื่อถือได้ และคืน decision แบบ immutable

โครงข้อความของแบบจำลองคือ id ตามด้วยจุด ตามด้วย timestamp แบบข้อความ ตามด้วยจุด แล้วต่อด้วย raw body bytes ตัวคั่นจุดช่วยแบ่งสอง field แรก ส่วน id ที่กำหนดเป็นอักขระ ASCII กลุ่มเล็กและห้ามมีจุดเพื่อไม่ให้ขอบเขตคลุมเครือ timestamp เป็นเลขฐานสิบ ASCII แบบ canonical คือจำนวนบวก ไม่มีศูนย์นำหน้า และยาวไม่เกิน 12 หลัก body เป็น bytes ตรงตามที่รับและถูกจำกัดขนาด ส่วน signature เป็น hex ตัวพิมพ์เล็ก 64 ตัว

ทำไมต้องใช้ verifier ของผู้ให้บริการจริง

มาตรฐานและผู้ให้บริการไม่ได้ใช้ header, prefix, encoding หรือรายละเอียดการคำนวณเหมือนกันทุกแห่ง Standard Webhooks อธิบายรูปแบบการเซ็นที่รวมรหัสข้อความ, timestamp และ payload ส่วนเอกสาร Svix เน้นการตรวจด้วย raw payload ก่อนแปลง JSON GitHub อธิบาย HMAC ของ payload ตามรูปแบบของตนเอง การอ้างอิงเหล่านี้ช่วยยืนยันหลักการและความแตกต่าง ไม่ได้ทำให้โค้ดสาธิตชิ้นเดียวใช้แทน SDK ของทุกค่ายได้

เหตุใด raw body จึงต้องมาก่อน JSON

JSON object เดียวกันอาจเขียนเป็นไบต์ได้หลายแบบ เช่นเว้นวรรคคนละตำแหน่ง สลับลำดับ key หรือใช้ escape รูปแบบต่างกัน เมื่อ parse แล้ว serialize ใหม่ โปรแกรมอาจเปลี่ยนช่องว่าง ลำดับตัวอักษรหรือการ escape แม้ค่าที่มองเห็นใน object จะดูเหมือนเดิม แต่ HMAC คำนวณจาก bytes ไม่ได้คำนวณจากความหมายที่มนุษย์อ่าน ดังนั้นการคำนวณลายเซ็นบน JSON ที่ serialize ใหม่อาจได้ digest ไม่ตรงกับผู้ส่ง

อ่านโค้ดจำลองโดยเริ่มจากขอบเขต ไม่ใช่จากตัวเลข digest

ไฟล์ mock_webhook_gate.py ใช้เฉพาะ standard library ของ Python 3.12 ขึ้นไป ประเภท GateDecision เป็น dataclass แบบ frozen จึงเป็นผล immutable ที่มี decision, reason และ age ซึ่งจะมีค่าเมื่อ signature ถูกต้องแล้ว ฟังก์ชันรับ key, id, timestamp, body, signature, เวลา UTC ที่ผู้เรียกฉีดเข้ามา, tolerance และธงบอกว่านาฬิกาเชื่อถือได้หรือไม่ ไม่มีการเปิด socket อ่าน environment เรียก database หรือสั่ง Agent

จากนั้นคำนวณ HMAC ด้วย hmac.new(key, signed_message, hashlib.sha256).hexdigest() และใช้ hmac.compare_digest เทียบ digest ที่มีชนิดเดียวกัน เอกสาร Python ระบุการใช้ฟังก์ชันนี้เพื่อช่วยลดการวิเคราะห์เวลาตามเนื้อหาของการเปรียบเทียบอินพุตแบบเดียวกัน แต่เราไม่กล่าวว่าทั้งโปรแกรมหรือบริการมี timing คงที่ การ parse timestamp เป็นจำนวนเต็มเกิดหลังลายเซ็นผ่าน เพื่อไม่ปฏิบัติต่อเวลาที่ยังไม่ผ่านการยืนยันเหมือนข้อเท็จจริงของผู้ส่ง

mock_webhook_gate.py

โค้ดและคำสั่งด้านล่างเป็นฉบับเขียนเพื่อการเรียนรู้ NOT RUN ผลลัพธ์ที่ระบุเป็นผลคาดหมาย ไม่ใช่บันทึกการรัน

"""Educational custom webhook envelope gate; not a provider implementation.

This module performs bounded syntax checks, HMAC verification over the exact
received bytes, and a signed-attempt timestamp freshness decision. It does not
parse JSON, authorize actions, deduplicate deliveries, persist work, or start an
agent. The key used by the authored tests is synthetic and public.
"""

from dataclasses import dataclass
import hashlib
import hmac
import re


MAX_BODY_BYTES = 65_536
MAX_ID_CHARS = 64
MAX_TIMESTAMP_CHARS = 12
MAX_TOLERANCE_SECONDS = 300
_ID_RE = re.compile(rb"[A-Za-z0-9_-]{1,64}\Z")
_TIMESTAMP_RE = re.compile(rb"[1-9][0-9]{0,11}\Z")
_SIGNATURE_RE = re.compile(r"[0-9a-f]{64}\Z")


@dataclass(frozen=True)
class GateDecision:
    """Immutable result; age_seconds is populated only after valid HMAC."""

    decision: str
    reason: str
    age_seconds: int | None = None


def sign_synthetic_fixture(key: bytes, envelope_id: str, timestamp: str, raw_body: bytes) -> str:
    """Create a fixture signature for local examples/tests only; never a live signer."""
    if not isinstance(key, bytes) or not key:
        raise TypeError("key must be non-empty bytes")
    if not isinstance(envelope_id, str) or not envelope_id.isascii():
        raise TypeError("envelope_id must be ASCII text")
    if not isinstance(timestamp, str) or not timestamp.isascii():
        raise TypeError("timestamp must be ASCII text")
    if not isinstance(raw_body, bytes):
        raise TypeError("raw_body must be bytes")
    message = envelope_id.encode("ascii") + b"." + timestamp.encode("ascii") + b"." + raw_body
    return hmac.new(key, message, hashlib.sha256).hexdigest()


def verify_envelope(
    *,
    key: bytes,
    envelope_id: str,
    timestamp: str,
    raw_body: bytes,
    signature: str,
    receiver_now: int,
    tolerance_seconds: int,
    clock_trusted: bool,
) -> GateDecision:
    """Return ACCEPT_ENVELOPE, REJECT, or HOLD using an injected UTC clock.

    The custom signed message is ASCII id + b"." + canonical timestamp text +
    b"." + untouched body. Limits and ±300-second tolerance are local teaching
    policies, not provider or standards requirements.
    """
    if not isinstance(clock_trusted, bool):
        return GateDecision("REJECT", "INVALID_CLOCK_TRUST_FLAG")
    if not isinstance(key, bytes) or not key:
        return GateDecision("REJECT", "INVALID_KEY_INPUT")
    if not isinstance(envelope_id, str) or not envelope_id.isascii():
        return GateDecision("REJECT", "INVALID_ID")
    id_bytes = envelope_id.encode("ascii")
    if not _ID_RE.fullmatch(id_bytes):
        return GateDecision("REJECT", "INVALID_ID")
    if not isinstance(timestamp, str) or not timestamp.isascii():
        return GateDecision("REJECT", "INVALID_TIMESTAMP")
    timestamp_bytes = timestamp.encode("ascii")
    if not _TIMESTAMP_RE.fullmatch(timestamp_bytes):
        return GateDecision("REJECT", "INVALID_TIMESTAMP")
    if not isinstance(raw_body, bytes):
        return GateDecision("REJECT", "INVALID_BODY_TYPE")
    if len(raw_body) > MAX_BODY_BYTES:
        return GateDecision("REJECT", "BODY_TOO_LARGE")
    if not isinstance(signature, str) or not _SIGNATURE_RE.fullmatch(signature):
        return GateDecision("REJECT", "INVALID_SIGNATURE_FORMAT")
    if type(receiver_now) is not int or receiver_now <= 0:
        return GateDecision("REJECT", "INVALID_RECEIVER_TIME")
    if type(tolerance_seconds) is not int or not 1 <= tolerance_seconds <= MAX_TOLERANCE_SECONDS:
        return GateDecision("REJECT", "INVALID_TOLERANCE")

    signed_message = id_bytes + b"." + timestamp_bytes + b"." + raw_body
    expected = hmac.new(key, signed_message, hashlib.sha256).hexdigest()
    # compare_digest is used for this digest comparison; it is not a promise
    # that the complete verifier or surrounding service has constant timing.
    if not hmac.compare_digest(expected, signature):
        return GateDecision("REJECT", "BAD_SIGNATURE")

    # Authenticate the envelope before consulting its timestamp. If the
    # receiver clock is untrusted, hold only a signature-verified envelope.
    if not clock_trusted:
        return GateDecision("HOLD", "CLOCK_UNTRUSTED")

    # Only now treat the signed timestamp as an authenticated attempt time.
    signed_attempt_seconds = int(timestamp)
    age = receiver_now - signed_attempt_seconds
    if age > tolerance_seconds:
        return GateDecision("REJECT", "TOO_OLD", age)
    if age < -tolerance_seconds:
        return GateDecision("REJECT", "TOO_FUTURE", age)
    return GateDecision("ACCEPT_ENVELOPE", "SIGNATURE_AND_FRESHNESS_OK", age)

test_mock_webhook_gate.py

โค้ดและคำสั่งด้านล่างเป็นฉบับเขียนเพื่อการเรียนรู้ NOT RUN ผลลัพธ์ที่ระบุเป็นผลคาดหมาย ไม่ใช่บันทึกการรัน

"""Authored standard-library tests. Status: NOT RUN by assignment scope."""

import unittest

from mock_webhook_gate import sign_synthetic_fixture, verify_envelope


# Public, deterministic synthetic fixture: exactly 32 bytes, never a credential.
KEY = bytes(range(32))
NOW = 2_000_000_000
TOLERANCE = 300
BODY = b'{"event":"video.render.ready","created_at":1999990000}'


def signed(timestamp: str, body: bytes = BODY, envelope_id: str = "evt_demo_01") -> str:
    return sign_synthetic_fixture(KEY, envelope_id, timestamp, body)


def verify(timestamp: str, *, body: bytes = BODY, envelope_id: str = "evt_demo_01",
           now: int = NOW, tolerance: int = TOLERANCE, trusted: bool = True,
           signature: str | None = None):
    return verify_envelope(
        key=KEY,
        envelope_id=envelope_id,
        timestamp=timestamp,
        raw_body=body,
        signature=signature if signature is not None else signed(timestamp, body, envelope_id),
        receiver_now=now,
        tolerance_seconds=tolerance,
        clock_trusted=trusted,
    )


class VerifyEnvelopeTests(unittest.TestCase):
    def test_current_attempt_is_accepted(self):
        self.assertEqual(verify(str(NOW)).decision, "ACCEPT_ENVELOPE")

    def test_exact_old_boundary_is_inclusive(self):
        self.assertEqual(verify(str(NOW - TOLERANCE)).decision, "ACCEPT_ENVELOPE")

    def test_one_second_too_old_is_rejected(self):
        self.assertEqual(verify(str(NOW - TOLERANCE - 1)).reason, "TOO_OLD")

    def test_exact_future_boundary_is_inclusive(self):
        self.assertEqual(verify(str(NOW + TOLERANCE)).decision, "ACCEPT_ENVELOPE")

    def test_one_second_too_future_is_rejected(self):
        self.assertEqual(verify(str(NOW + TOLERANCE + 1)).reason, "TOO_FUTURE")

    def test_tampered_timestamp_breaks_signature(self):
        original = signed(str(NOW))
        result = verify(str(NOW - 1), signature=original)
        self.assertEqual(result.reason, "BAD_SIGNATURE")
        self.assertIsNone(result.age_seconds)

    def test_changed_identifier_breaks_signature(self):
        original = signed(str(NOW), envelope_id="evt_demo_01")
        result = verify(str(NOW), envelope_id="evt_demo_02", signature=original)
        self.assertEqual(result.reason, "BAD_SIGNATURE")

    def test_one_hex_digit_changed_in_signature_is_bad_signature(self):
        original = signed(str(NOW))
        replacement = "0" if original[0] != "0" else "1"
        changed = replacement + original[1:]
        result = verify(str(NOW), signature=changed)
        self.assertEqual(result.reason, "BAD_SIGNATURE")

    def test_changed_body_byte_breaks_signature(self):
        original = signed(str(NOW), BODY)
        result = verify(str(NOW), body=BODY + b" ", signature=original)
        self.assertEqual(result.reason, "BAD_SIGNATURE")
        self.assertIsNone(result.age_seconds)

    def test_clock_untrusted_holds(self):
        result = verify(str(NOW), trusted=False)
        self.assertEqual((result.decision, result.reason), ("HOLD", "CLOCK_UNTRUSTED"))
        self.assertIsNone(result.age_seconds)

    def test_untrusted_clock_does_not_mask_invalid_signature(self):
        original = signed(str(NOW))
        replacement = "0" if original[0] != "0" else "1"
        invalid = replacement + original[1:]
        result = verify(str(NOW), trusted=False, signature=invalid)
        self.assertEqual((result.decision, result.reason), ("REJECT", "BAD_SIGNATURE"))
        self.assertIsNone(result.age_seconds)

    def test_old_business_event_with_fresh_attempt_passes_gate(self):
        old_event = b'{"created_at":1000000000,"event":"video.render.ready"}'
        self.assertEqual(verify(str(NOW), body=old_event).decision, "ACCEPT_ENVELOPE")

    def test_same_fresh_delivery_replayed_twice_still_passes(self):
        sig = signed(str(NOW))
        first = verify(str(NOW), signature=sig)
        second = verify(str(NOW), signature=sig)
        self.assertEqual(first.decision, "ACCEPT_ENVELOPE")
        self.assertEqual(second.decision, "ACCEPT_ENVELOPE")

    def test_noncanonical_timestamp_is_rejected(self):
        self.assertEqual(verify("02000000000").reason, "INVALID_TIMESTAMP")

    def test_zero_timestamp_is_rejected(self):
        self.assertEqual(verify("0").reason, "INVALID_TIMESTAMP")

    def test_twelve_digit_positive_timestamp_is_accepted_when_fresh(self):
        timestamp = "999999999999"
        result = verify(timestamp, now=999999999999)
        self.assertEqual(result.decision, "ACCEPT_ENVELOPE")

    def test_timestamp_over_twelve_digits_is_rejected(self):
        self.assertEqual(verify("1000000000000").reason, "INVALID_TIMESTAMP")

    def test_identifier_rejects_dot_separator_injection(self):
        self.assertEqual(verify(str(NOW), envelope_id="event.with.dot").reason, "INVALID_ID")

    def test_non_ascii_identifier_is_rejected(self):
        self.assertEqual(verify(str(NOW), envelope_id="งาน").reason, "INVALID_ID")

    def test_uppercase_hex_signature_is_rejected_by_envelope_policy(self):
        self.assertEqual(verify(str(NOW), signature=signed(str(NOW)).upper()).reason,
                         "INVALID_SIGNATURE_FORMAT")

    def test_body_limit_is_enforced(self):
        too_large = b"x" * 65_537
        self.assertEqual(verify(str(NOW), body=too_large).reason, "BODY_TOO_LARGE")

    def test_exact_body_limit_is_allowed(self):
        max_body = b"x" * 65_536
        result = verify(str(NOW), body=max_body)
        self.assertEqual(result.decision, "ACCEPT_ENVELOPE")

    def test_receiver_time_rejects_bool(self):
        self.assertEqual(verify(str(NOW), now=True).reason, "INVALID_RECEIVER_TIME")

    def test_receiver_time_rejects_float(self):
        self.assertEqual(verify(str(NOW), now=float(NOW)).reason, "INVALID_RECEIVER_TIME")

    def test_tolerance_rejects_bool(self):
        self.assertEqual(verify(str(NOW), tolerance=True).reason, "INVALID_TOLERANCE")

    def test_tolerance_must_be_between_one_and_three_hundred(self):
        self.assertEqual(verify(str(NOW), tolerance=0).reason, "INVALID_TOLERANCE")
        self.assertEqual(verify(str(NOW), tolerance=301).reason, "INVALID_TOLERANCE")

    def test_result_is_immutable(self):
        result = verify(str(NOW))
        with self.assertRaises((AttributeError, TypeError)):
            result.reason = "changed"


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

README.md

โค้ดและคำสั่งด้านล่างเป็นฉบับเขียนเพื่อการเรียนรู้ NOT RUN ผลลัพธ์ที่ระบุเป็นผลคาดหมาย ไม่ใช่บันทึกการรัน

# Mock webhook freshness gate (educational)

This Python 3.12+ standard-library example checks a **custom, educational HMAC-SHA256 hex envelope** before job admission. It is **not compatible with Standard Webhooks, Svix, GitHub, Stripe, or any provider**. The sample does not open a server, call a network/API/LLM/GPU, parse JSON, authorize an action, deduplicate, persist work, or execute an Agent.

The only skill demonstrated is verifying the original request bytes and the signed delivery-attempt timestamp before admitting an envelope. `ACCEPT_ENVELOPE` means transport checks passed under the demonstration policy; it does not authorize execution.

## Requirements and permissions

- Python 3.12 or newer; standard library only; no virtual environment, package install, environment variable, or credential required.
- The fixture key is synthetic and public. Never replace it with a live secret or copy this custom format into a production verifier.
- The example has no external service or API cost. Local CPU and electricity use are not measured.
- Tests are authored but **NOT RUN** in this assignment. The outputs below are expected results from the written contract, not observed execution results.

## Inputs and policy

The verifier receives the exact `raw_body: bytes`, an ASCII ID matching `[A-Za-z0-9_-]{1,64}`, canonical positive ASCII decimal timestamp text (at most 12 digits, no leading zero), one lowercase 64-character hex signature, an injected trusted receiver UTC timestamp, a trusted-clock boolean, and an integer tolerance from 1 through 300 seconds. The custom signed message is the ID, a dot, timestamp text, a dot, and the untouched body bytes. The body limit is 65,536 bytes. Those limits and the ±300-second inclusive window are local teaching choices, not specification mandates.

The verifier returns an immutable decision and reason. It uses `hmac.compare_digest` to compare the digest; that does not establish constant timing for the entire function or service. It rejects a bad signature before consulting the signed timestamp or clock-trust state. After a valid signature, an untrusted clock returns `HOLD` without calculating an age; only a valid signature and trusted clock allow freshness evaluation.

## Run commands (when separately authorized)

From this directory, a reader may run the authored tests with:

```text
python -B -m unittest -v
```

To view the module's public API without invoking a live integration, inspect `mock_webhook_gate.py`. The files intentionally contain no runnable server or live signing flow. This assignment did not execute the module or tests.

## Expected decision examples (not observed output)

With `receiver_now = 2000000000`, `tolerance_seconds = 300`, trusted clock, a matching synthetic key, and a correctly signed envelope:

| Signed attempt timestamp | Expected age | Expected result |
|---:|---:|---|
| 2000000000 | 0 | `ACCEPT_ENVELOPE` |
| 1999999700 | +300 | `ACCEPT_ENVELOPE` (inclusive old boundary) |
| 1999999699 | +301 | `REJECT / TOO_OLD` |
| 2000000300 | -300 | `ACCEPT_ENVELOPE` (inclusive future boundary) |
| 2000000301 | -301 | `REJECT / TOO_FUTURE` |

Changing the timestamp or even one body byte while retaining the previous signature is expected to produce `REJECT / BAD_SIGNATURE`, with `age_seconds` unset. An untrusted clock is expected to produce `HOLD / CLOCK_UNTRUSTED`. A fresh attempt carrying an old business event can pass this freshness gate; business-age policy remains a separate decision. Replaying the same fresh envelope twice is expected to pass twice because no deduplication state exists.

## Authored test coverage and limitations

`test_mock_webhook_gate.py` contains 27 authored `unittest` cases for the window edges, ID/body/timestamp/signature tampering, clock state and signature-before-clock ordering, business event age, repeated fresh delivery, malformed fields, size bounds, strict integer inputs, tolerance bounds, and immutable output. This is a test plan in code, **not evidence of runtime pass**. There is no provider SDK, HTTP stack, TLS, persistent receipt, database, queue, integration/load/timing/UI test, key rotation, or production readiness claim.

ตัวอย่างทำงาน: event เก่าอาจมากับ attempt ใหม่

สมมตินาฬิกาฝั่งรับที่เชื่อถือได้อ่านค่า 2,000,000,000 และนโยบายการสาธิตตั้ง tolerance ไว้ 300 วินาที เราใช้ id evt_demo_01 และ body ที่ระบุ event video.render.ready พร้อมเวลาเกิดเหตุการณ์เก่ากว่าเวลาฝั่งรับมาก คีย์ที่ใช้เป็น fixture สังเคราะห์และเปิดเผยได้สำหรับการสอนเท่านั้น เมื่อสร้างลายเซ็นที่ถูกต้องจาก id, timestamp ของความพยายามส่ง และ body เดิม ผลต่อไปนี้เป็นผลคาดหมายจากสัญญาและตัวอย่าง ไม่ใช่ output ที่ได้จากการรันโปรแกรม

ถ้า timestamp ของ attempt เท่ากับ 2,000,000,000 อายุเป็น 0 และผลคาดหมายคือ ACCEPT_ENVELOPE ถ้าเป็น 1,999,999,700 อายุ 300 ซึ่งตรงขอบเก่าพอดี ก็ยังรับ ถ้าเป็น 1,999,999,699 อายุ 301 ให้ปฏิเสธว่าเก่าเกินไป หาก timestamp อยู่ที่ 2,000,000,300 อายุ -300 จึงยังรับตามขอบอนาคตที่รวมไว้ แต่ 2,000,000,301 ให้ปฏิเสธว่าอยู่ในอนาคตเกินนโยบาย เนื้อหาที่มี event time เก่าแต่อยู่ใน attempt สดผ่าน freshness ได้ แล้วให้ business policy ตรวจต่อ; fresh replay ยังผ่านซ้ำเพราะไม่มี dedup

ตารางผลตัดสินที่ตรวจย้อนกลับได้

กรณี id, timestamp หรือ signature ผิดรูปแบบให้ REJECT ด้วย reason กลุ่ม INVALID โดยไม่ใช้เวลาไปตัดสินหรือส่งต่อ กรณี digest ของ id, timestamp และ raw body ไม่ตรงให้ REJECT / BAD_SIGNATURE และห้ามถือ timestamp เป็นข้อมูลยืนยันแล้ว หาก clock_trusted เป็น false ให้ HOLD / CLOCK_UNTRUSTED จนมีเวลาที่เชื่อถือได้ เมื่อ signature ผ่านและ age อยู่ในช่วง ±300 วินาทีรวมขอบ ให้ ACCEPT_ENVELOPE แล้วส่งต่อไปตรวจ JSON, schema, สิทธิ์และ duplicate แยกกัน เมื่อ age มากกว่า 300 ให้ REJECT / TOO_OLD; เมื่อ age ต่ำกว่า -300 ให้ REJECT / TOO_FUTURE

การมีผล ACCEPT ไม่ใช่คำสั่ง execute Agent สัญญารับเข้าในบทนี้จงใจสิ้นสุดที่การคืนผล immutable เท่านั้น ผู้เรียกต้องมีจุดเชื่อมต่อชัดเจนว่าเมื่อใดจึงสร้างงานต่อ และควรใช้ transaction หรือ durable queue เมื่อจำเป็นต้องรับประกันการกู้คืน ตัวอย่าง outbox สำหรับการส่งผลต่อเนื่องมีใน Transactional Outbox สำหรับ Agent ซึ่งครอบคลุมความทนทานของการเขียนและส่งต่ออีกปัญหาหนึ่ง ไม่ใช่ตัวแทนการตรวจ webhook freshness

จุดล้มเหลวและขอบเขตความปลอดภัย

นาฬิกาคลาดเคลื่อน: tolerance แคบเกินไปอาจปฏิเสธ request ที่ถูกต้องเมื่อระบบมี clock skew หรือคิวส่งต่อใช้เวลานานเกินคาด ส่วน tolerance กว้างเกินไปก็ยืดช่วงที่สำเนาเดิมมีโอกาสผ่าน ตัวอย่าง 300 วินาทีเป็นค่าที่เราเลือกเพื่อให้เห็นขอบ ไม่ใช่คำแนะนำการตั้ง production ต้องตรวจแหล่งเวลา, monitoring/NTP, การกระจายความคลาดเคลื่อนจริง และสัญญา provider ก่อนตั้งค่า หาก clock health ผิดปกติ ให้หยุด admission หรือพักงานตามนโยบายแทนการใช้เวลาที่ไม่น่าเชื่อถือ

กุญแจและการหมุนเวียน: ตัวอย่างมี key เดียวและไม่มี key ID, rotation, overlap, revocation หรือ secret storage จึงไม่ครอบคลุมวงจรชีวิตกุญแจจริง ระบบจริงต้องรู้ว่ากุญแจใดผูกกับ endpoint/tenant ใด วิธีเปลี่ยนกุญแจที่ไม่ทำให้ request ระหว่างทางพัง และวิธีเพิกถอนเมื่อสงสัยว่ารั่ว จัดเก็บ secret ในระบบจัดการความลับ ห้ามใส่กุญแจจริงใน source, test fixture, log หรือบทความ

สิทธิ์และข้อมูลไม่น่าเชื่อถือ: ลายเซ็นที่ถูกต้องไม่ได้แปลว่า event มีสิทธิ์แก้ resource ที่ระบุ ตรวจ tenant, audience, allowed event types และ action authorization หลัง freshness gate และก่อนเรียก tool หรือโมเดล ป้องกัน prompt injection ใน payload โดยถือเนื้อหาเป็น input data ไม่ใช่ policy เก็บเพียงข้อมูล audit ที่จำเป็น จำกัดผู้เข้าถึง และกำหนดระยะเวลาเก็บ payload ตามกฎข้อมูลขององค์กร

ชุดกรณีทดสอบที่เขียนไว้ และสถานะที่ต้องรายงาน

ไฟล์ test_mock_webhook_gate.py เขียนกรณีทดสอบด้วย unittest ของ Python ครอบคลุมขอบเวลาแบบรวมและเกินหนึ่งวินาทีทั้งด้านเก่าและอนาคต การแก้ timestamp หรือ body โดยใช้ signature เดิม clock ที่ไม่น่าเชื่อถือ event timestamp เก่าแต่ attempt สด replay สดซ้ำสองครั้ง รูปแบบ id/timestamp/signature ขนาด body ขอบเขต 12 หลัก receiver time ที่เป็น bool หรือ float tolerance ที่ไม่ถูกต้อง และความ immutable ของผลลัพธ์

จำนวนกรณีที่เขียนเป็นแผนตรวจ ไม่ใช่ผลการทดสอบ บทนี้ไม่ได้รัน import, ตัวอย่าง, module หรือ test ตามขอบเขตงาน ผล code tests จึงเป็น NOT RUN และไม่มีข้อความ PASS การตรวจเชิงข้อความพบว่าชื่อ test สอดคล้องกับ acceptance criteria ที่ระบุ แต่ไม่ได้พิสูจน์ว่า Python interpreter จะยอมรับ syntax หรือผลรันตรงคาด ผู้ใช้ที่ต้องการทดลองใน checkout ที่ปลอดภัยสามารถใช้คำสั่งใน README และตรวจผลด้วยตนเอง

แผนภาพและ Labs: แยกสิ่งที่จำลองออกจากระบบจริง

แบบจำลอง 3D ของบทนี้วางด่านรับเข้าเป็นชั้น เลือกสถานการณ์สด เก่า อนาคต ไบต์เปลี่ยน หรือนาฬิกาไม่น่าเชื่อถือ แล้วอ่านว่าการตัดสินหยุดที่ชั้นใด ส่วน Interactive แยกอีกชุดให้เปลี่ยนอายุ attempt ตั้งแต่ −360 ถึง +360 วินาที โดยค่าบวกหมายถึงผ่านมาแล้ว ค่าลบหมายถึงเวลาอนาคต พร้อมเลือกผลลายเซ็นและความน่าเชื่อถือของนาฬิกาที่สมมติไว้

ทดลองค่าที่ −301, −300, 300 และ 301 เพื่อเห็นขอบแบบรวม จากนั้นกดส่งซองเดิมซ้ำขณะยังสด แล้วสังเกตว่ากฎ freshness อาจให้ผลเดิมอีกครั้ง ตัวเลือก business event เก่าจงใจไม่เปลี่ยนผลเวลา attempt ทั้งหมดเป็นแบบจำลองเชิงกฎซึ่งรับผลลายเซ็นเป็นสมมติฐาน ไม่คำนวณ HMAC จริง ไม่เชื่อม endpoint และไม่เริ่ม Agent การมี label, reduced motion และ fallback ในโค้ดเป็นงานออกแบบ ผล UI จริงยัง UNKNOWN

เชื่อมด่านรับเข้ากับ Video Lab อย่างมีขอบเขต

ใช้ event สังเคราะห์ video.render.ready เป็นบริบทสำหรับงานวิดีโอที่อาจส่งต่อหลังการตรวจ ingress และเงื่อนไขปลายทางครบแล้ว Native Video Agent Lab รุ่น active1.4.0 เป็นเครื่องมือจำลอง workflow วิดีโอ ไม่ตรวจ webhook, HMAC, timestamp หรือ freshness ให้เพิ่มเงื่อนไขเหล่านี้ลง brief ที่อ่านข้อความได้ข้าง Lab และตรวจ prompt ที่สร้างร่วมกับ brief ห้ามตีความว่าการกดสร้าง prompt คือผ่านด่านรับเข้าแล้ว

ค่าเวลาและต้นทุนที่ปรับใน Lab เป็นสมมติฐานฝึกวางแผน ไม่ใช่ผลวัดงานจริง การฝัง shortcode และอ่าน server rendering ยืนยันเพียงเนื้อหาที่บันทึก การทำงานของปุ่ม การคัดลอก และหน้าจอจริงยังไม่ได้ทดสอบและคง UNKNOWN

เป้าหมาย: วางแผน workflow วิดีโอสั้นเพื่ออธิบายว่า “ผ่าน freshness gate แล้ว จึงค่อยส่งงานไปขั้น workflow ภายหลัง”
ข้อมูลและเครื่องมือ: ใช้เฉพาะ event สังเคราะห์ video.render.ready และตัวเลือกใน Hybrid Mind Video Agent Lab; ไม่ใช้ข้อมูลจริงหรือ credential
ขอบเขต: สมมติว่า ingress แยกตรวจ raw bytes, ลายเซ็น และอายุ delivery attempt แล้ว; ห้ามอ้างว่า Video Lab ทำการตรวจเหล่านี้
สิ่งส่งมอบ: prompt/แผนงานวิดีโอที่ระบุผู้ชม ลำดับภาพ คำบรรยาย และผลตรวจที่มนุษย์อ่านได้
เกณฑ์ตรวจรับ: งานแสดงลำดับ “fresh envelope ผ่านก่อน → ค่อยพิจารณา workflow”; แยกเวลาของ business event ออกจากเวลาส่ง; ไม่อ้างการป้องกัน replay หรือการทำงานจริง
จุดหยุด: หยุดทันทีเมื่อ brief ขอ secret, ข้อมูลจริง, การเรียก endpoint, การอนุมัติ action หรือหลักฐาน ingress ที่ไม่มีให้

ก่อนใช้แนวคิดนี้กับ endpoint จริง

เริ่มจากระบุ provider, โปรโตคอล, header และ bytes ที่ผู้ให้บริการลงนามจริงให้ชัด จากนั้นใช้ verifier หรือ SDK ทางการที่ตรงรุ่น ตรวจ raw body โดยไม่ผ่าน middleware ที่เปลี่ยนข้อมูล และทดสอบกรณี body ที่ถูกปรับ whitespace หรือลำดับ key ตรวจว่ามี signed attempt timestamp หรือไม่ อย่าสร้าง freshness check จาก timestamp ที่ผู้ให้บริการไม่ได้ลงนาม

แบบฝึกหัด: เติมตารางตัดสินใจให้ครบ

ใช้ receiver time 2,000,000,000 และ tolerance 300 วินาที เติมผล decision/reason สำหรับกรณีต่อไปนี้: (ก) timestamp 1,999,999,700 พร้อม signature ถูกต้อง (ข) timestamp 1,999,999,699 พร้อม signature ถูกต้อง (ค) timestamp 2,000,000,301 พร้อม signature ถูกต้อง (ง) timestamp ถูกแก้หนึ่งหลักแต่ยังใช้ signature เก่า (จ) clock_trusted เป็น false (ฉ) attempt สดและถูกต้อง แต่ body มี created_at เก่ามาก (ช) replay envelope สดชุดเดิมสองครั้ง

เกณฑ์ตรวจ: ต้องระบุว่าขอบ ±300 รวมอยู่ในช่วง จึงตอบ ACCEPT_ENVELOPE สำหรับ (ก); (ข) เก่าเกินด้วย TOO_OLD; (ค) อนาคตเกินด้วย TOO_FUTURE; (ง) BAD_SIGNATURE และไม่มี age ที่ยืนยันแล้ว; (จ) HOLD/CLOCK_UNTRUSTED; (ฉ) freshness ผ่านได้แต่ต้องให้กฎธุรกิจตรวจ event age ต่อ; (ช) verifier รับได้สองครั้งเพราะยังไม่มี dedup การตอบเพียงว่า “ลายเซ็นถูกต้อง” ไม่ผ่านเกณฑ์ เพราะไม่แยก freshness, business age และ replay/idempotency

Brief หกส่วนที่คัดลอกได้

เป้าหมาย: ออกแบบด่านรับซอง webhook ที่ตรวจไบต์เดิม HMAC และ signed attempt timestamp ก่อนส่งต่อผล ACCEPT_ENVELOPE, REJECT หรือ HOLD ไม่เริ่ม Agent เอง
ข้อมูลและเครื่องมือ: ใช้ Python3.12+ standard library และ fixture สังเคราะห์32bytes; รับ id ASCII [A-Za-z0-9_-]{1,64}, timestamp บวกแบบ canonical ≤12หลัก, raw bytes ≤65536, signature lowercasehex64, receiverUTCเวลา int บวก, clock flag bool และ tolerance int1..300 ซึ่งเป็นนโยบายตัวอย่าง
ขอบเขต: Custom Mock เพื่อเรียนรู้เท่านั้น ไม่เข้ากับ provider ใด ห้ามใช้ secretจริง, network/API/LLM/GPU/server/database/queue, ติดตั้ง package, parseJSONในverifier, deploy หรือแตะข้อมูลจริง ไม่ทำ authorization/dedup/exactly-once
สิ่งส่งมอบ: Module, decisiontable, README วิธีใช้ และ authoredtests พร้อม reason/ageที่ยืนยันแล้ว แหล่งอ้างอิงและข้อจำกัด ผลทดสอบที่ยังไม่รันต้องระบุ NOT RUN ไม่บันทึก key/signature/body ลง log
เกณฑ์ตรวจรับ: ±300รวมขอบ; ±301ปฏิเสธ; แก้ id/time/body/signature แล้วต้องไม่ผ่าน; ไม่ใช้ ageจากลายเซ็นผิด; MACถูกแต่นาฬิกาไม่น่าเชื่อถือให้HOLD; eventเก่า attemptสดและ freshreplayยังอาจผ่าน; ผล immutable ไม่สั่งaction
จุดหยุด: หยุดเมื่อเอกสาร/โค้ด/ตาราง/ชุดทดสอบที่เขียนครบตามสัญญา หรือข้อมูลรับเข้าผิด/เกินขนาด/MACผิด/เวลาเก่าหรืออนาคตเกินนโยบาย; ถ้าไม่มี clockเชื่อถือได้ให้HOLD ห้ามขยายสิทธิ์ เปิดendpoint หรืออ้างผลรันที่ไม่มี

ต้นทุน ข้อจำกัด และสิ่งที่ยังไม่ทราบ

ตัวอย่างนี้ใช้ Python standard library และไม่มีการเรียก API, LLM, GPU, network หรือระบบภายนอก ดังนั้นต้นทุน API/GPU/external service ของการสาธิตตามที่เขียนไว้คือศูนย์ ค่าใช้จ่ายของการรัน Python บนเครื่องผู้เรียนขึ้นกับอุปกรณ์และพลังงาน จึงไม่ได้วัดหรืออ้างตัวเลข latency/cost ในบทนี้ การคำนวณ HMAC เป็นงาน CPU ที่มีขนาดสัมพันธ์กับ body แต่เราไม่ได้ทำ benchmark

ไม่ทราบว่า tolerance ใดเหมาะกับผู้ให้บริการหรือ deployment ของผู้อ่าน ไม่ทราบค่าคลาดเคลื่อน clock จริง ไม่ทราบรูปแบบ retry, key management, payload schema, rate limit หรือการรับประกัน delivery ของ endpoint ใด ไม่มีการยืนยันว่า framework ของ production เก็บ raw bytes ตามที่ verifier ต้องการ และไม่มี runtime evidence ว่าตัวอย่างผ่าน interpreter เพราะไม่ได้รัน

คำถามที่พบบ่อย

ลายเซ็น HMAC ถูกต้องแล้ว ทำไมต้องตรวจ timestamp อีก?

HMAC ตรวจว่า bytes ตรงกับข้อความที่ลงนามด้วย key แต่สำเนาคำขอเดิมยังมีลายเซ็นถูกต้องได้ การตรวจ timestamp ของ attempt ที่ถูกเซ็นช่วยจำกัดช่วงที่ envelope เก่าผ่านได้ อย่างไรก็ตาม replay สดภายในช่วงยังอาจผ่าน จึงต้องใช้ dedup หรือ idempotency แยกต่างหาก

ใช้เวลาที่อยู่ใน JSON body แทนได้ไหม?

ได้เฉพาะเมื่อ provider ระบุ field นั้นเป็น timestamp ของ attempt และลายเซ็นครอบคลุมตามโปรโตคอลที่ตรวจสอบแล้ว เวลาธุรกิจอย่าง created_at อาจเก่าแม้เพิ่ง retry ส่งมา และ timestamp ที่ผู้ส่งไม่ได้ลงนามไม่ควรใช้พิสูจน์ freshness

ทำไมต้องตรวจ raw body ก่อน parse JSON?

HMAC คำนวณจาก bytes การ parse แล้ว serialize ใหม่อาจเปลี่ยนช่องว่าง ลำดับ key หรือ escape ทำให้ bytes ไม่ตรงกับที่ผู้ส่งลงนาม จึงตรวจ raw payload ตามวิธีของ provider ก่อนเปลี่ยนรูปข้อมูล

ถ้า clock ของ server ไม่น่าเชื่อถือควรทำอย่างไร?

อย่าตัดสิน freshness ด้วยเวลาที่ผู้ส่งควบคุมได้ เมื่อซองผ่าน HMAC แล้วแต่ clock_trusted เป็น false แบบจำลองคืน HOLD ระบบจริงต้องกำหนดนโยบายพักหรือกักงาน ตรวจแหล่งเวลาและ clock health แล้วกลับมารับงานเมื่อเงื่อนไขเชื่อถือได้

ACCEPT_ENVELOPE หมายถึงเรียก Agent ได้เลยหรือไม่?

ไม่ หมายถึงผ่านเฉพาะ syntax, HMAC และ freshness ตามสัญญาจำลอง ยังต้อง parse/schema validation, authorization, tenant checks, dedup และการบันทึกงานตามสถาปัตยกรรมก่อนอนุญาตให้ Agent ทำ action

แหล่งอ้างอิงและอ่านต่อ

Standard Webhooks 1.0.0 อธิบาย signed ID, timestamp และ payload พร้อมแยก delivery attempt จาก idempotency: specification. Svix อธิบายการตรวจ raw payload ก่อนแปลงตาม verifier ของตน: คู่มือตรวจ payload. Python 3.12 อธิบาย HMAC และ compare_digest: เอกสาร hmac. GitHub อธิบายการตรวจตามรูปแบบ webhook ของตน: คู่มือตรวจ delivery. อ่านต่อเรื่อง receipt-before-ACK และ outbox ในลิงก์ภายในด้านบน.