MCP Events 接入实战

"""完全离线教学模型；不是平台 callback、HTTP 服务或生产 SDK。"""

import base64
import binascii
from dataclasses import dataclass
from datetime import datetime, timezone
import hashlib
import hmac
import json
import re
import sqlite3
import time
from urllib.parse import urlsplit


MAX_BODY = 256 * 1024  # 复现当前 MCP Events 集成的完整 body 上限
TOLERANCE = 300  # 投递时间戳允许 ±300 秒；不限制事件发生时间
ID = re.compile(r"[A-Za-z0-9_-]{1,200}\Z")  # 本例额外收紧 ID 格式
ISO_TIME = re.compile(
    r"\d{4}-\d{2}-\d{2}T\d{2}:\d{2}:\d{2}(?:\.\d{1,6})?"
    r"(?:Z|[+-](?:[01]\d|2[0-3]):[0-5]\d)\Z"
)


class Rejected(Exception):
    def __init__(self, status, code):
        self.status, self.code = status, code
        super().__init__(code)


@dataclass(frozen=True)
class Response:
    status: int
    body: dict


@dataclass(frozen=True)
class Subscription:
    secrets: tuple  # 仅来自可信配置；轮换期可配置同一订阅的多个 key
    document_id: str = "demo_doc"
    event_name: str = "comment.created"


def decode_secret(secret):
    """支持标准 base64（包括省略末尾 padding）；拒绝不合规配置。"""
    if not isinstance(secret, str) or not secret.startswith("whsec_"):
        raise ValueError("configured secret must start with whsec_")
    encoded = secret[6:]
    try:
        key = base64.b64decode(encoded + "=" * (-len(encoded) % 4), validate=True)
    except (ValueError, binascii.Error) as exc:
        raise ValueError("invalid configured base64 secret") from exc
    if not 24 <= len(key) <= 64:
        raise ValueError("configured key must contain 24..64 bytes")
    return key


def signature(raw_body, message_id, timestamp, key):
    """只为本地 fixture 签名；绝不生成、请求或替代平台 secret。"""
    message = (message_id + "." + timestamp + ".").encode("utf-8") + raw_body
    digest = hmac.new(key, message, hashlib.sha256).digest()
    return "v1," + base64.b64encode(digest).decode("ascii")


def normalized_headers(headers):
    normalized = {}
    if not isinstance(headers, dict) or len(headers) > 64:
        raise Rejected(400, "invalid_headers")
    for name, value in headers.items():
        if not isinstance(name, str) or not isinstance(value, str):
            raise Rejected(400, "invalid_headers")
        lower = name.lower()
        if lower in normalized or len(value) > 8192 or "\r" in value or "\n" in value:
            raise Rejected(400, "invalid_headers")
        normalized[lower] = value
    for name in ("webhook-id", "webhook-timestamp", "webhook-signature", "x-mcp-subscription-id"):
        if not normalized.get(name):
            raise Rejected(400, "missing_header")
    return normalized


def verify(raw_body, headers, keys, now):
    """精确原始字节验签在 JSON 解析之前；这里只实现 v1 HMAC-SHA256。"""
    message_id, timestamp = headers["webhook-id"], headers["webhook-timestamp"]
    if not ID.fullmatch(message_id) or not re.fullmatch(r"[0-9]{1,12}", timestamp):
        raise Rejected(400, "invalid_signed_metadata")
    supplied = []
    for item in headers["webhook-signature"].split(" "):
        version, sep, value = item.partition(",")
        if version != "v1" or not sep:
            continue
        try:
            digest = base64.b64decode(value, validate=True)
        except (ValueError, binascii.Error):
            continue
        if len(digest) == 32:
            supplied.append(digest)
    message = (message_id + "." + timestamp + ".").encode("utf-8") + raw_body
    valid = False
    for key in keys:
        expected = hmac.new(key, message, hashlib.sha256).digest()
        for digest in supplied:
            valid |= hmac.compare_digest(expected, digest)
    if not valid:
        raise Rejected(401, "invalid_signature")
    if abs(now - int(timestamp)) > TOLERANCE:
        raise Rejected(401, "timestamp_outside_window")


def unique_object(pairs):
    result = {}
    for key, value in pairs:
        if key in result:
            raise ValueError("duplicate JSON key")
        result[key] = value
    return result


def reject_constant(value):
    raise ValueError("non-finite JSON number: " + value)


def parse_body(raw_body):
    try:
        body = json.loads(raw_body.decode("utf-8"), object_pairs_hook=unique_object,
                          parse_constant=reject_constant)
        # 也拒绝转义的孤立 surrogate，避免后续 UTF-8 存储异常。
        json.dumps(body, ensure_ascii=False, allow_nan=False).encode("utf-8")
    except (ValueError, UnicodeError, RecursionError) as exc:
        raise Rejected(400, "invalid_json") from exc
    if not isinstance(body, dict):
        raise Rejected(400, "invalid_schema")
    return body


def require_fields(body, required, optional=()):
    if not isinstance(body, dict) or not set(required) <= body.keys():
        raise Rejected(400, "invalid_schema")
    if body.keys() - set(required) - set(optional):
        raise Rejected(400, "invalid_schema")
    if "_meta" in body and not isinstance(body["_meta"], dict):
        raise Rejected(400, "invalid_schema")


def text(value, maximum=4096, empty=False):
    if not isinstance(value, str) or len(value) > maximum or (not empty and not value):
        raise Rejected(400, "invalid_schema")
    return value


def canonical_event(body, headers, subscription):
    """手写的示例 schema 检查，不是通用 JSON Schema validator。"""
    require_fields(body, ("name", "eventId", "timestamp", "data"), ("cursor", "_meta"))
    for field in ("name", "eventId", "timestamp"):
        text(body[field], 200)
    if body["eventId"] != headers["webhook-id"]:
        raise Rejected(400, "event_id_mismatch")
    if body["name"] != subscription.event_name:
        raise Rejected(422, "event_name_mismatch")
    if body.get("cursor") is not None:
        text(body["cursor"], 4096, empty=True)
    stamp = body["timestamp"]
    try:
        if not ISO_TIME.fullmatch(stamp):
            raise ValueError("expected timezone-aware ISO timestamp")
        event_time = datetime.fromisoformat(stamp.replace("Z", "+00:00"))
        event_time = event_time.astimezone(timezone.utc).isoformat()
    except (ValueError, OverflowError) as exc:
        raise Rejected(400, "invalid_event_timestamp") from exc
    data = body["data"]
    require_fields(data, ("document_id", "comment_id", "text", "url"))
    text(data["document_id"], 200)
    text(data["comment_id"], 200)
    text(data["text"], 10000, empty=True)
    text(data["url"], 4096)
    try:
        url = urlsplit(data["url"])
        if (url.scheme != "https" or not url.hostname
                or url.username is not None or url.password is not None):
            raise ValueError("expected HTTPS display URL")
    except ValueError as exc:
        raise Rejected(400, "invalid_schema") from exc
    if data["document_id"] != subscription.document_id:
        raise Rejected(422, "document_filter_mismatch")
    # 本地策略：比较不可变业务字段；重投可更换 cursor、_meta、JSON 排版。
    core = {"name": body["name"], "eventId": body["eventId"],
            "timestamp": event_time, "data": data}
    return json.dumps(core, sort_keys=True, ensure_ascii=False, separators=(",", ":"))


class Receiver:
    def __init__(self, database, subscriptions, clock=time.time):
        self.clock, self.subscriptions, self.keys = clock, dict(subscriptions), {}
        owners = {}
        for sid, sub in self.subscriptions.items():
            if not ID.fullmatch(sid) or not sub.secrets:
                raise ValueError("invalid local subscription configuration")
            self.keys[sid] = tuple(decode_secret(secret) for secret in sub.secrets)
            for key in self.keys[sid]:
                if key in owners and owners[key] != sid:
                    raise ValueError("never reuse a key across subscriptions")
                owners[key] = sid
        self.db = sqlite3.connect(database, isolation_level=None)
        self.db.execute("PRAGMA foreign_keys = ON")
        self.db.executescript("""
            CREATE TABLE IF NOT EXISTS inbox (
                subscription_id TEXT NOT NULL, event_id TEXT NOT NULL,
                canonical_core TEXT NOT NULL, raw_body BLOB NOT NULL,
                accepted_at REAL NOT NULL,
                PRIMARY KEY (subscription_id, event_id));
            CREATE TABLE IF NOT EXISTS local_summary_jobs (
                subscription_id TEXT NOT NULL, event_id TEXT NOT NULL,
                state TEXT NOT NULL DEFAULT 'pending_local_only',
                PRIMARY KEY (subscription_id, event_id),
                FOREIGN KEY (subscription_id, event_id)
                    REFERENCES inbox(subscription_id, event_id));
        """)

    def close(self):
        self.db.close()

    def receive(self, raw_body, headers):
        try:
            if not isinstance(raw_body, bytes):
                raise Rejected(400, "body_must_be_bytes")
            if len(raw_body) > MAX_BODY:
                raise Rejected(413, "body_too_large")
            headers = normalized_headers(headers)
            sid = headers["x-mcp-subscription-id"]
            if sid not in self.subscriptions:
                raise Rejected(404, "unknown_subscription")
            # 路由头本身不在签名输入中；不同订阅必须使用不同可信 key。
            verify(raw_body, headers, self.keys[sid], self.clock())
            body = parse_body(raw_body)
            if body.get("type") == "verification":
                return self._verification(body)
            core = canonical_event(body, headers, self.subscriptions[sid])
            return self._record(sid, body["eventId"], core, raw_body)
        except Rejected as exc:
            return Response(exc.status, {"error": exc.code})
        except sqlite3.Error:
            return Response(503, {"error": "local_storage_unavailable"})

    def _verification(self, body):
        require_fields(body, ("type", "challenge"))
        text(body["challenge"], 4096)
        return Response(200, {"challenge": body["challenge"]})

    def _record(self, sid, event_id, core, raw_body):
        self.db.execute("BEGIN IMMEDIATE")
        try:
            row = self.db.execute(
                "SELECT canonical_core FROM inbox WHERE subscription_id=? AND event_id=?",
                (sid, event_id)).fetchone()
            if row:
                if row[0] != core:
                    raise Rejected(409, "event_content_conflict")
                result = Response(200, {"status": "duplicate"})
            else:
                self.db.execute("INSERT INTO inbox VALUES (?, ?, ?, ?, ?)",
                                (sid, event_id, core, raw_body, self.clock()))
                self.db.execute(
                    "INSERT INTO local_summary_jobs (subscription_id, event_id) VALUES (?, ?)",
                    (sid, event_id))
                result = Response(202, {"status": "accepted_local_only"})
            self.db.execute("COMMIT")
            return result
        except Exception:
            self.db.execute("ROLLBACK")
            raise

"""公开的本地测试夹具。以下 secret 没有保密性，绝不能用于实际接入。"""

import base64
import json

from receiver import Subscription, decode_secret, signature


NOW = 1_800_000_000
SUBSCRIPTION_ID = "sub_local_demo"
OTHER_SUBSCRIPTION_ID = "sub_local_other"
# 32 字节可预测 ASCII：教学测试值，不是平台 secret。
LOCAL_TEST_SECRET = "whsec_" + base64.b64encode(b"PUBLIC_LOCAL_TEST_ONLY_KEY_000001").decode()
ROTATED_TEST_SECRET = "whsec_" + base64.b64encode(b"PUBLIC_LOCAL_TEST_ONLY_KEY_000002").decode()
OTHER_TEST_SECRET = "whsec_" + base64.b64encode(b"PUBLIC_LOCAL_TEST_ONLY_KEY_000003").decode()


def subscriptions():
    return {
        SUBSCRIPTION_ID: Subscription((LOCAL_TEST_SECRET,)),
        OTHER_SUBSCRIPTION_ID: Subscription((OTHER_TEST_SECRET,)),
    }


def event(event_id="evt_local_1"):
    return {
        "name": "comment.created",
        "eventId": event_id,
        "timestamp": "2027-01-15T08:00:00Z",
        "data": {"document_id": "demo_doc", "comment_id": "comment_local_1",
                 "text": "请核对第三段的引用。", "url": "https://example.invalid/comments/1"},
        "cursor": "opaque_local_cursor_1",
    }


def raw(body):
    return json.dumps(body, ensure_ascii=False, separators=(",", ":")).encode("utf-8")


def headers(raw_body, message_id="evt_local_1", timestamp=NOW,
            subscription_id=SUBSCRIPTION_ID, secret=LOCAL_TEST_SECRET):
    timestamp = str(timestamp)
    return {
        "webhook-id": message_id,
        "webhook-timestamp": timestamp,
        "webhook-signature": signature(raw_body, message_id, timestamp, decode_secret(secret)),
        "X-MCP-Subscription-Id": subscription_id,
    }

"""无网络、无依赖包、无监听端口；所有行为均在临时 SQLite 中断言。"""

import base64
from concurrent.futures import ThreadPoolExecutor
import json
from pathlib import Path
import tempfile
import threading
import unittest
from unittest.mock import patch

from fixtures import (LOCAL_TEST_SECRET, NOW, OTHER_SUBSCRIPTION_ID, OTHER_TEST_SECRET,
                      ROTATED_TEST_SECRET, SUBSCRIPTION_ID, event, headers, raw,
                      subscriptions)
from receiver import (MAX_BODY, Receiver, Subscription, decode_secret,
                      signature, verify)


class SignatureTests(unittest.TestCase):
    def test_official_known_vector(self):
        # 官方 Python tests/test_webhooks.py:test_sign_function 的公开向量。
        # Official public vector: Standard Webhooks Python tests (see README.md).
        key = decode_secret("whsec_MfKQ9r8GKYqrTwjUPD8ILPZIo2LaLaSw")
        payload = b'{"test": 2432232314}'
        message_id, stamp = "msg_p5jXN8AQM9LWM0D4loKWxJek", "1614265330"
        expected = "v1,g0hM9SsE+OTPJTGt/tmIKtSyZlE3uFJELVlNIOLJ1OE="
        self.assertEqual(signature(payload, message_id, stamp, key), expected)
        verify(payload, {"webhook-id": message_id, "webhook-timestamp": stamp,
                         "webhook-signature": expected}, (key,), int(stamp))

    def test_secret_format_and_size(self):
        for size in (24, 32, 64):
            value = "whsec_" + base64.b64encode(b"x" * size).decode()
            self.assertEqual(decode_secret(value), b"x" * size)
            self.assertEqual(decode_secret(value.rstrip("=")), b"x" * size)
        bad = ("whsec_", "not_a_secret", "whsec_@@@", "whsec_" + "eA==",
               "whsec_" + base64.b64encode(b"x" * 65).decode())
        for value in bad:
            with self.subTest(value=value), self.assertRaises(ValueError):
                decode_secret(value)


class ReceiverTests(unittest.TestCase):
    def setUp(self):
        self.directory = tempfile.TemporaryDirectory()
        self.path = str(Path(self.directory.name) / "inbox.sqlite")
        self.receiver = Receiver(self.path, subscriptions(), clock=lambda: NOW)

    def tearDown(self):
        self.receiver.close()
        self.directory.cleanup()

    def send(self, body=None, **header_options):
        body = event() if body is None else body
        payload = raw(body)
        return self.receiver.receive(payload, headers(payload, **header_options))

    def counts(self):
        return tuple(self.receiver.db.execute("SELECT COUNT(*) FROM " + name).fetchone()[0]
                     for name in ("inbox", "local_summary_jobs"))

    def test_valid_event_is_only_durably_queued(self):
        response = self.send()
        self.assertEqual((response.status, response.body),
                         (202, {"status": "accepted_local_only"}))
        self.assertEqual(self.counts(), (1, 1))
        state = self.receiver.db.execute("SELECT state FROM local_summary_jobs").fetchone()[0]
        self.assertEqual(state, "pending_local_only")

    def test_body_whitespace_change_breaks_old_signature(self):
        payload = raw(event())
        response = self.receiver.receive(payload + b" ", headers(payload))
        self.assertEqual(response.status, 401)
        self.assertEqual(self.counts(), (0, 0))

    def test_tampered_signed_headers(self):
        payload = raw(event())
        for name, value in (("webhook-id", "evt_other"), ("webhook-timestamp", str(NOW + 1))):
            tampered = headers(payload)
            tampered[name] = value
            with self.subTest(name=name):
                self.assertEqual(self.receiver.receive(payload, tampered).status, 401)

    def test_expired_and_future_delivery_timestamps(self):
        for offset in (-301, 301):
            with self.subTest(offset=offset):
                response = self.send(timestamp=NOW + offset)
                self.assertEqual(response.body, {"error": "timestamp_outside_window"})

    def test_time_window_boundary_is_inclusive(self):
        for offset in (-300, 300):
            with self.subTest(offset=offset):
                self.assertIn(self.send(timestamp=NOW + offset).status, (200, 202))

    def test_unknown_subscription(self):
        self.assertEqual(self.send(subscription_id="sub_unknown").status, 404)

    def test_unsigned_routing_header_cannot_switch_subscription(self):
        payload = raw(event())
        changed = headers(payload)
        changed["X-MCP-Subscription-Id"] = OTHER_SUBSCRIPTION_ID
        self.assertEqual(self.receiver.receive(payload, changed).status, 401)

    def test_configuration_rejects_cross_subscription_key_reuse(self):
        config = subscriptions()
        config[OTHER_SUBSCRIPTION_ID] = Subscription((LOCAL_TEST_SECRET,))
        with self.assertRaises(ValueError):
            Receiver(":memory:", config)

    def test_body_event_id_must_match_signed_header(self):
        self.assertEqual(self.send(event("evt_other")).body, {"error": "event_id_mismatch"})

    def test_wrong_event_name_and_document_filter(self):
        wrong_name = event()
        wrong_name["name"] = "comment.deleted"
        wrong_document = event()
        wrong_document["data"]["document_id"] = "some_other_doc"
        for body in (wrong_name, wrong_document):
            with self.subTest(body=body):
                self.assertEqual(self.send(body).status, 422)

    def test_invalid_schema_and_timestamp(self):
        invalid = []
        body = event(); del body["data"]["comment_id"]; invalid.append(body)
        body = event(); body["data"]["text"] = 42; invalid.append(body)
        body = event(); body["data"]["url"] = "file:///etc/passwd"; invalid.append(body)
        body = event(); body["data"]["extra"] = True; invalid.append(body)
        body = event(); body["timestamp"] = "2027-01-15T08:00:00"; invalid.append(body)
        body = event(); body["timestamp"] = "2027-02-30T08:00:00Z"; invalid.append(body)
        body = event(); body["timestamp"] = "2027-01-15T08:00:00+00:60"; invalid.append(body)
        body = event(); body["timestamp"] = "0001-01-01T00:00:00+01:00"; invalid.append(body)
        body = event(); body["timestamp"] = "9999-12-31T23:59:59-01:00"; invalid.append(body)
        body = event(); body["cursor"] = 4; invalid.append(body)
        body = event(); body["_meta"] = []; invalid.append(body)
        body = event(); body["secret"] = LOCAL_TEST_SECRET; invalid.append(body)
        invalid.extend(([], None, 4, "event"))
        for body in invalid:
            payload = raw(body)
            with self.subTest(body=body):
                self.assertEqual(self.receiver.receive(payload, headers(payload)).status, 400)
        self.assertEqual(self.counts(), (0, 0))

    def test_duplicate_json_keys_nonfinite_invalid_utf8_and_surrogate(self):
        bad_payloads = (b'{"name":"comment.created","name":"comment.deleted"}',
                        b'{"data":{"text":"a","text":"b"}}',
                        b'{"data":NaN}', b'{"data":Infinity}', b'{"data":1e400}',
                        b'\xff', b'{"x":"\\ud800"}', b'{')
        for payload in bad_payloads:
            with self.subTest(payload=payload):
                response = self.receiver.receive(payload, headers(payload))
                self.assertEqual(response.body, {"error": "invalid_json"})

    def test_invalid_signature_is_checked_before_json(self):
        payload = b'{invalid'
        bad_headers = headers(payload)
        bad_headers["webhook-signature"] = "v1,not_base64"
        self.assertEqual(self.receiver.receive(payload, bad_headers).body,
                         {"error": "invalid_signature"})

    def test_dual_signature_rotation_and_unknown_versions(self):
        payload = raw(event())
        old_headers = headers(payload)
        new_headers = headers(payload, secret=ROTATED_TEST_SECRET)
        combined = dict(old_headers)
        combined["webhook-signature"] = ("v2,ignored v1,malformed " +
            new_headers["webhook-signature"] + " " + old_headers["webhook-signature"])
        self.assertEqual(self.receiver.receive(payload, combined).status, 202)
        self.receiver.close()
        config = subscriptions()
        config[SUBSCRIPTION_ID] = Subscription((ROTATED_TEST_SECRET,))
        self.receiver = Receiver(self.path, config, clock=lambda: NOW)
        self.assertEqual(self.receiver.receive(payload, combined).status, 200)
        self.assertEqual(self.receiver.receive(payload, old_headers).status, 401)

    def test_duplicate_survives_restart(self):
        self.assertEqual(self.send().status, 202)
        self.assertEqual(self.send().body, {"status": "duplicate"})
        self.receiver.close()
        self.receiver = Receiver(self.path, subscriptions(), clock=lambda: NOW)
        self.assertEqual(self.send().body, {"status": "duplicate"})
        self.assertEqual(self.counts(), (1, 1))

    def test_concurrent_connections_create_only_one_job(self):
        barrier = threading.Barrier(2)
        payload = raw(event())
        def deliver():
            receiver = Receiver(self.path, subscriptions(), clock=lambda: NOW)
            try:
                barrier.wait(timeout=5)
                return receiver.receive(payload, headers(payload)).status
            finally:
                receiver.close()
        with ThreadPoolExecutor(max_workers=2) as pool:
            responses = list(pool.map(lambda _: deliver(), range(2)))
        self.assertEqual(sorted(responses), [200, 202])
        self.assertEqual(self.counts(), (1, 1))

    def test_cursor_meta_json_layout_and_equivalent_timezone_can_change(self):
        self.send()
        body = event()
        body["cursor"] = "different_opaque_cursor"
        body["_meta"] = {"attempt": 2}
        body["timestamp"] = "2027-01-15T09:00:00+01:00"
        payload = json.dumps(body, sort_keys=True, indent=2).encode("utf-8")
        response = self.receiver.receive(payload, headers(payload, timestamp=NOW + 1))
        self.assertEqual(response.status, 200)
        self.assertEqual(self.counts(), (1, 1))

    def test_cursor_is_optional(self):
        body = event()
        del body["cursor"]
        self.assertEqual(self.send(body).status, 202)
        body["cursor"] = None
        self.assertEqual(self.send(body).body, {"status": "duplicate"})

    def test_same_event_id_different_core_content_is_conflict(self):
        self.send()
        for field, value in (("text", "changed"), ("comment_id", "another"),
                             ("url", "https://example.invalid/comments/2")):
            body = event()
            body["data"][field] = value
            with self.subTest(field=field):
                self.assertEqual(self.send(body).body, {"error": "event_content_conflict"})
        body = event(); body["timestamp"] = "2027-01-15T08:00:01Z"
        self.assertEqual(self.send(body).status, 409)
        self.assertEqual(self.counts(), (1, 1))

    def test_same_event_id_is_scoped_to_subscription(self):
        self.send()
        self.assertEqual(self.send(subscription_id=OTHER_SUBSCRIPTION_ID,
                                   secret=OTHER_TEST_SECRET).status, 202)
        self.assertEqual(self.counts(), (2, 2))

    def test_job_failure_rolls_back_inbox_and_retry_succeeds(self):
        self.receiver.db.execute("""CREATE TRIGGER fail_job BEFORE INSERT ON local_summary_jobs
                                 BEGIN SELECT RAISE(ABORT, 'injected local test failure'); END""")
        self.assertEqual(self.send().status, 503)
        self.assertEqual(self.counts(), (0, 0))
        self.receiver.db.execute("DROP TRIGGER fail_job")
        self.assertEqual(self.send().status, 202)
        self.assertEqual(self.counts(), (1, 1))

    def test_out_of_order_events_are_accepted_without_order_claim(self):
        newer, older = event("evt_newer"), event("evt_older")
        newer["timestamp"], older["timestamp"] = "2027-01-15T08:00:10Z", "2027-01-15T08:00:00Z"
        self.assertEqual(self.send(newer, message_id="evt_newer").status, 202)
        self.assertEqual(self.send(older, message_id="evt_older").status, 202)
        self.assertEqual(self.counts(), (2, 2))

    def verification(self):
        return {"type": "verification", "challenge": "local_challenge_123"}

    def test_verification_echo_after_signature_without_business_job(self):
        body = self.verification()
        response = self.send(body, message_id="verification_local_1")
        self.assertEqual(
            (response.status, response.body),
            (200, {"challenge": "local_challenge_123"}),
        )
        self.assertEqual(self.counts(), (0, 0))

    def test_verification_wrong_signature_does_not_echo(self):
        payload = raw(self.verification())
        response = self.receiver.receive(payload, headers(payload, secret=OTHER_TEST_SECRET))
        self.assertEqual(response.body, {"error": "invalid_signature"})

    def test_verification_invalid_challenge(self):
        for challenge in (None, "", "x" * 4097):
            body = self.verification(); body["challenge"] = challenge
            with self.subTest(challenge_type=type(challenge)):
                self.assertEqual(self.send(body).status, 400)

    def test_oversize_body_rejected_before_parse_or_storage(self):
        payload = b"x" * (MAX_BODY + 1)
        self.assertEqual(self.receiver.receive(payload, headers(payload)).status, 413)
        self.assertEqual(self.counts(), (0, 0))

    def test_body_at_exact_size_limit_is_accepted(self):
        payload = raw(event())
        payload += b" " * (MAX_BODY - len(payload))
        self.assertEqual(self.receiver.receive(payload, headers(payload)).status, 202)

    def test_no_network_operation_is_needed(self):
        with patch("socket.socket", side_effect=AssertionError("network forbidden")), \
             patch("socket.create_connection", side_effect=AssertionError("network forbidden")), \
             patch("socket.getaddrinfo", side_effect=AssertionError("DNS forbidden")):
            self.assertEqual(self.send().status, 202)
            self.assertEqual(self.send(self.verification()).status, 200)

    def test_header_names_are_case_insensitive(self):
        payload = raw(event())
        mixed = {name.upper(): value for name, value in headers(payload).items()}
        self.assertEqual(self.receiver.receive(payload, mixed).status, 202)

    def test_invalid_or_ambiguous_headers_and_non_bytes(self):
        payload = raw(event())
        cases = []
        bad = headers(payload); del bad["webhook-id"]; cases.append(bad)
        bad = headers(payload); bad["Webhook-Id"] = bad["webhook-id"]; cases.append(bad)
        bad = headers(payload); bad["webhook-id"] = "evt.with.dot"; cases.append(bad)
        bad = headers(payload); bad["webhook-timestamp"] = "1.2"; cases.append(bad)
        bad = headers(payload); bad["webhook-id"] = "x\r\nInjected: y"; cases.append(bad)
        for bad in cases:
            with self.subTest(headers=bad):
                self.assertEqual(self.receiver.receive(payload, bad).status, 400)
        self.assertEqual(self.receiver.receive(payload.decode(), headers(payload)).status, 400)


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