Add Python webhook verification helper

Author: Delega Bot

src/delega/__init__.py

@@ -9,6 +9,7 @@
     DelegaRateLimitError,
 )
 from .models import Agent, Comment, Project, Task
+from .webhooks import verify_webhook
 
 __version__ = "0.1.1"
 
@@ -24,6 +25,7 @@
     "DelegaRateLimitError",
     "Project",
     "Task",
+    "verify_webhook",
 ]
 
 

src/delega/webhooks.py

@@ -0,0 +1,64 @@
+"""Webhook verification helpers."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+from datetime import datetime, timezone
+
+
+def _parse_timestamp(value: str) -> datetime:
+    try:
+        parsed = datetime.fromisoformat(value.replace("Z", "+00:00"))
+    except ValueError as exc:
+        raise ValueError("invalid timestamp") from exc
+
+    if parsed.tzinfo is None:
+        return parsed.replace(tzinfo=timezone.utc)
+    return parsed.astimezone(timezone.utc)
+
+
+def verify_webhook(
+    payload: bytes,
+    signature: str,
+    timestamp: str,
+    secret: str,
+    tolerance_seconds: int = 300,
+) -> bool:
+    """Verify a Delega webhook signature.
+
+    Args:
+        payload: Raw request body bytes.
+        signature: Value of the X-Delega-Signature header.
+        timestamp: Value of the X-Delega-Timestamp header.
+        secret: Your webhook secret.
+        tolerance_seconds: Max age in seconds.
+
+    Returns:
+        True if the signature matches and the timestamp is within tolerance.
+
+    Raises:
+        ValueError: If the signature format is invalid, the timestamp is stale,
+            or the signature does not match.
+    """
+    if not signature.startswith("sha256="):
+        raise ValueError("bad signature format")
+
+    signature_hex = signature[len("sha256=") :]
+    if len(signature_hex) != 64 or any(ch not in "0123456789abcdefABCDEF" for ch in signature_hex):
+        raise ValueError("bad signature format")
+
+    received_at = _parse_timestamp(timestamp)
+    age_seconds = abs((datetime.now(timezone.utc) - received_at).total_seconds())
+    if age_seconds > tolerance_seconds:
+        raise ValueError("stale timestamp")
+
+    expected = "sha256=" + hmac.new(
+        secret.encode("utf-8"),
+        timestamp.encode("utf-8") + b"." + payload,
+        hashlib.sha256,
+    ).hexdigest()
+    if not hmac.compare_digest(expected, signature):
+        raise ValueError("signature mismatch")
+
+    return True

tests/test_webhooks.py

@@ -0,0 +1,68 @@
+"""Webhook verification tests."""
+
+from __future__ import annotations
+
+import hashlib
+import hmac
+import unittest
+from datetime import datetime, timedelta, timezone
+
+from delega import verify_webhook
+
+
+def _timestamp(delta: timedelta = timedelta()) -> str:
+    return (datetime.now(timezone.utc) + delta).isoformat().replace("+00:00", "Z")
+
+
+def _signature(payload: bytes, timestamp: str, secret: str) -> str:
+    digest = hmac.new(
+        secret.encode("utf-8"),
+        timestamp.encode("utf-8") + b"." + payload,
+        hashlib.sha256,
+    ).hexdigest()
+    return f"sha256={digest}"
+
+
+class TestVerifyWebhook(unittest.TestCase):
+    def setUp(self) -> None:
+        self.payload = b'{"event":"task.created","task":{"id":"abc123"}}'
+        self.secret = "whsec_test_secret"
+
+    def test_verifies_valid_signature(self) -> None:
+        timestamp = _timestamp()
+        signature = _signature(self.payload, timestamp, self.secret)
+
+        self.assertTrue(
+            verify_webhook(self.payload, signature, timestamp, self.secret)
+        )
+
+    def test_rejects_bad_signature_format(self) -> None:
+        timestamp = _timestamp()
+
+        with self.assertRaisesRegex(ValueError, "bad signature format"):
+            verify_webhook(self.payload, "not-a-sha256", timestamp, self.secret)
+
+    def test_rejects_stale_timestamp(self) -> None:
+        timestamp = _timestamp(timedelta(minutes=-6))
+        signature = _signature(self.payload, timestamp, self.secret)
+
+        with self.assertRaisesRegex(ValueError, "stale timestamp"):
+            verify_webhook(self.payload, signature, timestamp, self.secret)
+
+    def test_rejects_signature_mismatch(self) -> None:
+        timestamp = _timestamp()
+        signature = _signature(self.payload, timestamp, "wrong_secret")
+
+        with self.assertRaisesRegex(ValueError, "signature mismatch"):
+            verify_webhook(self.payload, signature, timestamp, self.secret)
+
+    def test_rejects_invalid_timestamp(self) -> None:
+        signature = _signature(self.payload, "not-a-timestamp", self.secret)
+
+        with self.assertRaisesRegex(ValueError, "invalid timestamp"):
+            verify_webhook(
+                self.payload,
+                signature,
+                "not-a-timestamp",
+                self.secret,
+            )