fix(webhook): fail loudly on unverifiable signatures, add timestamp freshness and replay protection (fixes #11)

lark_channel/card/action_handler.py

@@ -1,4 +1,3 @@
-import hashlib
 import hmac
 import json
 import logging
@@ -19,6 +18,10 @@
     build_error_response_content,
     should_record_security_audit,
 )
+from lark_channel.core.webhook_signature import (
+    ReplayGuard,
+    verify_webhook_signature,
+)
 from .model import Card
 
 if TYPE_CHECKING:
@@ -32,6 +35,7 @@ def __init__(self, security: Optional["SecurityConfig"] = None) -> None:
         self._verification_token: Optional[str] = None
         self._processor: Optional[Callable[[Card], Any]] = None
         self._security = security or _default_security_config()
+        self._replay_guard_instance: Optional[ReplayGuard] = None
 
     def do(self, req: RawRequest) -> RawResponse:
         if logger.isEnabledFor(logging.DEBUG):
@@ -204,15 +208,25 @@ def _record_security_audit(
         )
 
     def _verify_sign(self, request: RawRequest) -> None:
-        if self._verification_token is None or self._verification_token == "":
-            return
-        timestamp = request.headers.get(LARK_REQUEST_TIMESTAMP)
-        nonce = request.headers.get(LARK_REQUEST_NONCE)
-        signature = request.headers.get(LARK_REQUEST_SIGNATURE)
-        bs = (timestamp + nonce + self._verification_token).encode(UTF_8) + request.body
-        h = hashlib.sha1(bs)
-        if signature != h.hexdigest():
-            raise AccessDeniedException("signature verification failed")
+        verify_webhook_signature(
+            request,
+            secret=self._verification_token,
+            algorithm="sha1",
+            security=self._security,
+            record_audit=lambda reason, action: self._record_security_audit(
+                reason, action=action, request=request
+            ),
+            warn=lambda msg: logger.warning("%s", msg),
+            replay_guard=self._replay_guard(),
+        )
+
+    def _replay_guard(self) -> Optional[ReplayGuard]:
+        ttl = self._security.replay_protection_seconds
+        if ttl is None:
+            return None
+        if self._replay_guard_instance is None:
+            self._replay_guard_instance = ReplayGuard(ttl)
+        return self._replay_guard_instance
 
     @staticmethod
     def builder(

lark_channel/channel/config.py

@@ -416,6 +416,12 @@ class SecurityConfig:
     max_ws_fragment_bytes: Optional[int] = None
     max_concurrent_ws_handlers: Optional[int] = None
     resource_overflow_policy: ResourceOverflowPolicy = "audit"
+    # Webhook signature hardening (issue #11). Both are opt-in: when unset,
+    # legacy behaviour is preserved (timestamps are not checked, no replay
+    # dedup). When set, violations are audited (and warned) in compat/audit
+    # mode and rejected in strict mode.
+    max_timestamp_skew_seconds: Optional[int] = None
+    replay_protection_seconds: Optional[int] = None
 
     def __post_init__(self) -> None:
         if self.mode not in ("compat", "audit", "strict"):
@@ -441,6 +447,8 @@ def __post_init__(self) -> None:
             "max_ws_fragment_parts",
             "max_ws_fragment_bytes",
             "max_concurrent_ws_handlers",
+            "max_timestamp_skew_seconds",
+            "replay_protection_seconds",
         ):
             value = getattr(self, field_name)
             if value is not None and (

lark_channel/channel/tests/test_webhook_signature_hardening.py

@@ -0,0 +1,310 @@
+"""Webhook signature hardening (issue #11).
+
+Regression tests: a request carrying signature headers must never be accepted
+silently when verification is impossible (no secret configured), and the
+opt-in timestamp-freshness and replay-protection checks must reject stale or
+replayed requests in strict mode while keeping the legacy accepting behaviour
+(plus audit records and warnings) in compat/audit mode.
+"""
+
+import hashlib
+import json
+import time
+
+import pytest
+
+from lark_channel.card.action_handler import CardActionHandler
+from lark_channel.channel.config import SecurityConfig
+from lark_channel.core.const import (
+    LARK_REQUEST_NONCE,
+    LARK_REQUEST_SIGNATURE,
+    LARK_REQUEST_TIMESTAMP,
+)
+from lark_channel.core.model import RawRequest
+from lark_channel.core.webhook_signature import (
+    REASON_WEBHOOK_REPLAY_DETECTED,
+    REASON_WEBHOOK_SIGNATURE_UNVERIFIABLE,
+    REASON_WEBHOOK_TIMESTAMP_STALE,
+)
+from lark_channel.event.dispatcher_handler import EventDispatcherHandler
+from lark_channel.event.security import InMemorySecurityAuditRecorder
+
+
+def _request(body, headers=None):
+    req = RawRequest()
+    req.uri = "https://example.com/open-apis/bot/v2/hook"
+    req.headers = headers or {}
+    req.body = body if isinstance(body, bytes) else json.dumps(body).encode("utf-8")
+    return req
+
+
+def _signed_headers(body, secret, *, algorithm="sha256", timestamp=None, nonce=None):
+    timestamp = timestamp or str(int(time.time()))
+    nonce = nonce or "nonce-1"
+    data = (timestamp + nonce + secret).encode("utf-8") + body
+    digest = (
+        hashlib.sha256(data).hexdigest()
+        if algorithm == "sha256"
+        else hashlib.sha1(data).hexdigest()
+    )
+    return {
+        LARK_REQUEST_SIGNATURE: digest,
+        LARK_REQUEST_TIMESTAMP: timestamp,
+        LARK_REQUEST_NONCE: nonce,
+    }
+
+
+def _plain_event():
+    return {
+        "schema": "2.0",
+        "header": {"event_type": "example.event", "token": "verification-token"},
+        "event": {"value": "ok"},
+    }
+
+
+def _plain_card():
+    return {"type": "card.action.trigger", "action": {"value": {"k": "v"}}}
+
+
+def _reasons(recorder):
+    return [e.reason for e in recorder.events]
+
+
+# ---------------------------------------------------------------------------
+# No secret configured -> must not silently no-op
+# ---------------------------------------------------------------------------
+
+
+def test_compat_unverifiable_signature_is_audited_not_blocked():
+    seen = []
+    recorder = InMemorySecurityAuditRecorder()
+    handler = (
+        EventDispatcherHandler.builder(
+            "",
+            "verification-token",
+            security=SecurityConfig(audit_recorder=recorder),
+        )
+        .register_p2_customized_event("example.event", lambda event: seen.append(event))
+        .build()
+    )
+
+    resp = handler.do(_request(_plain_event(), _signed_headers(b"", "some-key")))
+
+    assert resp.status_code == 200
+    assert len(seen) == 1
+    assert REASON_WEBHOOK_SIGNATURE_UNVERIFIABLE in _reasons(recorder)
+
+
+def test_strict_unverifiable_signature_rejects():
+    recorder = InMemorySecurityAuditRecorder()
+    handler = (
+        EventDispatcherHandler.builder(
+            "",
+            "verification-token",
+            security=SecurityConfig(mode="strict", audit_recorder=recorder),
+        )
+        .register_p2_customized_event("example.event", lambda event: None)
+        .build()
+    )
+
+    resp = handler.do(_request(_plain_event(), _signed_headers(b"", "some-key")))
+
+    assert resp.status_code == 500
+    assert REASON_WEBHOOK_SIGNATURE_UNVERIFIABLE in _reasons(recorder)
+
+
+def test_compat_card_unverifiable_signature_is_audited_not_blocked():
+    seen = []
+    recorder = InMemorySecurityAuditRecorder()
+    handler = (
+        CardActionHandler.builder(
+            "",
+            "",
+            security=SecurityConfig(audit_recorder=recorder),
+        )
+        .register(lambda card: seen.append(card))
+        .build()
+    )
+
+    resp = handler.do(_request(_plain_card(), _signed_headers(b"", "some-key", algorithm="sha1")))
+
+    assert resp.status_code == 200
+    assert len(seen) == 1
+    assert REASON_WEBHOOK_SIGNATURE_UNVERIFIABLE in _reasons(recorder)
+
+
+# ---------------------------------------------------------------------------
+# Timestamp freshness (opt-in)
+# ---------------------------------------------------------------------------
+
+
+def test_strict_stale_timestamp_rejects():
+    recorder = InMemorySecurityAuditRecorder()
+    body = json.dumps(_plain_event()).encode("utf-8")
+    headers = _signed_headers(body, "encrypt-key", timestamp="1500000000")
+    handler = (
+        EventDispatcherHandler.builder(
+            "encrypt-key",
+            "verification-token",
+            security=SecurityConfig(
+                mode="strict",
+                audit_recorder=recorder,
+                max_timestamp_skew_seconds=60,
+            ),
+        )
+        .register_p2_customized_event("example.event", lambda event: None)
+        .build()
+    )
+
+    resp = handler.do(_request(body, headers))
+
+    assert resp.status_code == 500
+    assert REASON_WEBHOOK_TIMESTAMP_STALE in _reasons(recorder)
+
+
+def test_compat_stale_timestamp_is_audited_not_blocked():
+    seen = []
+    recorder = InMemorySecurityAuditRecorder()
+    body = json.dumps(_plain_event()).encode("utf-8")
+    headers = _signed_headers(body, "encrypt-key", timestamp="1500000000")
+    handler = (
+        EventDispatcherHandler.builder(
+            "encrypt-key",
+            "verification-token",
+            security=SecurityConfig(
+                audit_recorder=recorder,
+                max_timestamp_skew_seconds=60,
+            ),
+        )
+        .register_p2_customized_event("example.event", lambda event: seen.append(event))
+        .build()
+    )
+
+    resp = handler.do(_request(body, headers))
+
+    assert resp.status_code == 200
+    assert len(seen) == 1
+    assert REASON_WEBHOOK_TIMESTAMP_STALE in _reasons(recorder)
+
+
+def test_fresh_timestamp_passes_with_skew_enabled():
+    seen = []
+    recorder = InMemorySecurityAuditRecorder()
+    body = json.dumps(_plain_event()).encode("utf-8")
+    headers = _signed_headers(body, "encrypt-key")
+    handler = (
+        EventDispatcherHandler.builder(
+            "encrypt-key",
+            "verification-token",
+            security=SecurityConfig(
+                mode="strict",
+                audit_recorder=recorder,
+                max_timestamp_skew_seconds=60,
+            ),
+        )
+        .register_p2_customized_event("example.event", lambda event: seen.append(event))
+        .build()
+    )
+
+    resp = handler.do(_request(body, headers))
+
+    assert resp.status_code == 200
+    assert len(seen) == 1
+    assert REASON_WEBHOOK_TIMESTAMP_STALE not in _reasons(recorder)
+
+
+def test_lowercase_signature_headers_are_verified():
+    """ASGI servers lowercase header names (issue #12): the hardened verifier
+    must find the signature headers case-insensitively, not crash with a
+    TypeError from None + None + secret."""
+    seen = []
+    body = json.dumps(_plain_event()).encode("utf-8")
+    headers = _signed_headers(body, "encrypt-key")
+    lowercase = {k.lower(): v for k, v in headers.items()}
+    handler = (
+        EventDispatcherHandler.builder(
+            "encrypt-key",
+            "verification-token",
+            security=SecurityConfig(mode="strict"),
+        )
+        .register_p2_customized_event("example.event", lambda event: seen.append(event))
+        .build()
+    )
+
+    resp = handler.do(_request(body, lowercase))
+
+    assert resp.status_code == 200
+    assert len(seen) == 1
+
+
+# ---------------------------------------------------------------------------
+# Replay protection (opt-in)
+# ---------------------------------------------------------------------------
+
+
+def test_strict_replayed_request_rejects():
+    recorder = InMemorySecurityAuditRecorder()
+    body = json.dumps(_plain_event()).encode("utf-8")
+    headers = _signed_headers(body, "encrypt-key")
+    handler = (
+        EventDispatcherHandler.builder(
+            "encrypt-key",
+            "verification-token",
+            security=SecurityConfig(
+                mode="strict",
+                audit_recorder=recorder,
+                replay_protection_seconds=60,
+            ),
+        )
+        .register_p2_customized_event("example.event", lambda event: None)
+        .build()
+    )
+
+    first = handler.do(_request(body, headers))
+    assert first.status_code == 200
+
+    replay = handler.do(_request(body, headers))
+    assert replay.status_code == 500
+    assert REASON_WEBHOOK_REPLAY_DETECTED in _reasons(recorder)
+
+
+def test_strict_replayed_card_rejects():
+    recorder = InMemorySecurityAuditRecorder()
+    body = json.dumps(_plain_card()).encode("utf-8")
+    headers = _signed_headers(body, "verification-token", algorithm="sha1")
+    handler = (
+        CardActionHandler.builder(
+            "",
+            "verification-token",
+            security=SecurityConfig(
+                mode="strict",
+                audit_recorder=recorder,
+                replay_protection_seconds=60,
+            ),
+        )
+        .register(lambda card: None)
+        .build()
+    )
+
+    first = handler.do(_request(body, headers))
+    assert first.status_code == 200
+
+    replay = handler.do(_request(body, headers))
+    assert replay.status_code == 500
+    assert REASON_WEBHOOK_REPLAY_DETECTED in _reasons(recorder)
+
+
+# ---------------------------------------------------------------------------
+# Configuration validation
+# ---------------------------------------------------------------------------
+
+
+def test_security_config_validates_new_fields():
+    with pytest.raises(ValueError):
+        SecurityConfig(max_timestamp_skew_seconds=0)
+    with pytest.raises(ValueError):
+        SecurityConfig(replay_protection_seconds=-1)
+    with pytest.raises(TypeError):
+        SecurityConfig(max_timestamp_skew_seconds=True)  # bool is not an int
+    assert SecurityConfig(max_timestamp_skew_seconds=300).max_timestamp_skew_seconds == 300
+    assert SecurityConfig(replay_protection_seconds=60).replay_protection_seconds == 60