Add verify_webhook_signature helper for webhook authenticity

fishjam/__init__.py

@@ -12,7 +12,11 @@
 from fishjam._openapi_client.models import PeerMetadata
 
 # API
-from fishjam._webhook_notifier import decode_server_notifications, receive_binary
+from fishjam._webhook_notifier import (
+    decode_server_notifications,
+    receive_binary,
+    verify_webhook_signature,
+)
 from fishjam._ws_notifier import FishjamNotifier
 from fishjam.api._fishjam_client import (
     AgentOptions,
@@ -34,6 +38,7 @@
     "FishjamNotifier",
     "decode_server_notifications",
     "receive_binary",
+    "verify_webhook_signature",
     "PeerMetadata",
     "PeerOptions",
     "PeerOptionsVapi",

fishjam/_webhook_notifier.py

@@ -1,5 +1,6 @@
 """Module for decoding received webhook notifications from Fishjam."""
 
+import hmac
 import warnings
 from typing import List, Union
 
@@ -69,6 +70,26 @@ def decode_server_notifications(binary: bytes) -> List[AllowedNotification]:
     return []
 
 
+def verify_webhook_signature(body: bytes, signature: str, secret: str) -> bool:
+    """Verify the `x-fishjam-signature-256` header of a raw webhook body.
+
+    Accepts the `sha256=<hex>` format sent by Fishjam (the prefix is
+    optional) and compares in constant time. Call this with the raw request
+    body before passing it to `decode_server_notifications`.
+
+    Args:
+        body: The raw binary body of the webhook request.
+        signature: The value of the `x-fishjam-signature-256` header.
+        secret: The webhook secret configured in Fishjam.
+
+    Returns:
+        bool: True when the signature matches the body, False otherwise.
+    """
+    expected = hmac.new(secret.encode(), body, "sha256").hexdigest()
+    provided = signature.strip().removeprefix("sha256=")
+    return hmac.compare_digest(provided, expected)
+
+
 def receive_binary(
     binary: bytes,
 ) -> Union[AllowedNotification, List[AllowedNotification], None]:

tests/test_webhook_notifier.py

@@ -1,6 +1,12 @@
+import hmac
+
 import pytest
 
-from fishjam import decode_server_notifications, receive_binary
+from fishjam import (
+    decode_server_notifications,
+    receive_binary,
+    verify_webhook_signature,
+)
 from fishjam.events import (
     ServerMessagePeerConnected,
     ServerMessageRoomCreated,
@@ -171,3 +177,28 @@ def test_decode_empty_batch_returns_empty_list():
     )
 
     assert decode_server_notifications(binary) == []
+
+
+BODY = bytes(ServerMessage(room_created=ServerMessageRoomCreated(room_id="r1")))
+SECRET = "webhook-secret"
+SIGNATURE = hmac.new(SECRET.encode(), BODY, "sha256").hexdigest()
+
+
+def test_verify_valid_signature_with_prefix():
+    assert verify_webhook_signature(BODY, f"sha256={SIGNATURE}", SECRET)
+
+
+def test_verify_valid_signature_without_prefix():
+    assert verify_webhook_signature(BODY, SIGNATURE, SECRET)
+
+
+def test_verify_wrong_secret_fails():
+    assert not verify_webhook_signature(BODY, f"sha256={SIGNATURE}", "other-secret")
+
+
+def test_verify_tampered_body_fails():
+    assert not verify_webhook_signature(BODY + b"x", f"sha256={SIGNATURE}", SECRET)
+
+
+def test_verify_garbage_signature_fails():
+    assert not verify_webhook_signature(BODY, "sha256=nothex", SECRET)