{{#each realm_embedded_bots}}
{{#each (object_entries config) as |entry|}}
diff --git a/web/templates/settings/edit_bot_form.hbs b/web/templates/settings/edit_bot_form.hbs
index 4d71e501168b8..9b460fc21eb4c 100644
--- a/web/templates/settings/edit_bot_form.hbs
+++ b/web/templates/settings/edit_bot_form.hbs
@@ -39,8 +39,15 @@
widget_name="edit_bot_owner"
label=(t 'Owner')}}
-
+
+
+ {{#if is_incoming_webhook_bot}}
+
+
+
+ {{/if}}
+
{{!-- Shows the current avatar --}}
@@ -109,6 +116,18 @@
}}
{{/if}}
+
+ {{#if is_incoming_webhook_bot}}
+
+ {{> ../components/action_button
+ label=(t "Delete secret")
+ variant="subtle"
+ intent="danger"
+ id="clear_webhook_secret_button"
+ }}
+
+ {{/if}}
+
{{#if is_active}}
{{> ../components/action_button
diff --git a/zerver/actions/users.py b/zerver/actions/users.py
index 3be5a98f257a4..0dd1a9e0f0c2a 100644
--- a/zerver/actions/users.py
+++ b/zerver/actions/users.py
@@ -827,7 +827,7 @@ def do_update_outgoing_webhook_service(
def do_update_bot_config_data(bot_profile: UserProfile, config_data: dict[str, str]) -> None:
for key, value in config_data.items():
set_bot_config(bot_profile, key, value)
- updated_config_data = get_bot_config(bot_profile)
+ service_dicts = get_service_dicts_for_bot(bot_profile.id)
send_event_on_commit(
bot_profile.realm,
dict(
@@ -835,7 +835,7 @@ def do_update_bot_config_data(bot_profile: UserProfile, config_data: dict[str, s
op="update",
bot=dict(
user_id=bot_profile.id,
- services=[dict(config_data=updated_config_data)],
+ services=service_dicts,
),
),
bot_owner_user_ids(bot_profile),
diff --git a/zerver/lib/test_classes.py b/zerver/lib/test_classes.py
index 6f3c061f8644e..e2e15fa2fcfef 100644
--- a/zerver/lib/test_classes.py
+++ b/zerver/lib/test_classes.py
@@ -1,5 +1,7 @@
import asyncio
import base64
+import hashlib
+import hmac
import os
import re
import shutil
@@ -35,6 +37,7 @@
from django.test.testcases import SerializeMixin
from django.urls import resolve
from django.utils import translation
+from django.utils.encoding import force_bytes
from django.utils.module_loading import import_string
from django.utils.timezone import now as timezone_now
from fakeldap import MockLDAP
@@ -54,6 +57,7 @@
from zerver.actions.user_settings import do_change_full_name, do_change_user_setting
from zerver.actions.users import do_change_user_role
from zerver.decorator import do_two_factor_login
+from zerver.lib.bot_config import set_bot_config
from zerver.lib.cache import bounce_key_prefix_for_testing
from zerver.lib.email_notifications import MissedMessageData, handle_missedmessage_emails
from zerver.lib.initial_password import initial_password
@@ -2544,6 +2548,8 @@ class WebhookTestCase(ZulipTestCase):
DEFAULT_URL_TEMPLATE: str = (
"/api/v1/external/{webhook_dir_name}?stream={stream}&api_key={api_key}"
)
+ WEBHOOK_SIGNATURE_HEADER: str | None = None
+ WEBHOOK_TEST_SECRET: str | None = None
def get_webhook_dir_name(self) -> str:
module_parts = self.__module__.split(".")
@@ -2650,16 +2656,37 @@ def check_webhook(
"""
self.subscribe(self.test_user, self.channel_name)
+ url = getattr(self, "url", None)
+ if url is None:
+ url = self.build_webhook_url() # nocoverage
+
+ webhook_secret = getattr(self, "WEBHOOK_TEST_SECRET", None)
+ if webhook_secret is not None:
+ set_bot_config(self.test_user, "webhook_secret", webhook_secret)
+
payload = self.get_payload(fixture_name)
if content_type is not None:
extra["content_type"] = content_type
+
+ signature_header_name = getattr(self, "WEBHOOK_SIGNATURE_HEADER", None)
+ if signature_header_name is not None:
+ try:
+ raw_payload = self.get_body(fixture_name)
+ except FileNotFoundError: # nocoverage
+ raw_payload = ""
+
+ signature_value = self.get_webhook_signature(force_bytes(raw_payload))
+ if signature_value is not None:
+ django_header = "HTTP_" + signature_header_name.upper().replace("-", "_")
+ extra[django_header] = signature_value
+
headers = call_fixture_to_headers(self.webhook_dir_name, fixture_name)
headers = standardize_headers(headers)
extra.update(headers)
try:
msg = self.send_webhook_payload(
self.test_user,
- self.url,
+ url,
payload,
**extra,
)
@@ -2699,6 +2726,18 @@ def assert_channel_message(
self.assertEqual(message.topic_name(), topic_name)
self.assertEqual(message.content, content)
+ def get_webhook_signature(self, raw_payload: bytes) -> str | None:
+ """
+ Generate the signature header value for a given payload.
+ Override this method in child classes if the integration uses different signature format.
+ """
+ secret = getattr(self, "WEBHOOK_TEST_SECRET", None)
+ if secret is None:
+ return None # nocoverage
+
+ # Default implementation matches the current GitHub standard format
+ return "sha256=" + hmac.new(force_bytes(secret), raw_payload, hashlib.sha256).hexdigest()
+
def send_and_test_private_message(
self,
fixture_name: str,
@@ -2715,6 +2754,11 @@ def send_and_test_private_message(
Most webhooks send to streams, and you will want to look at
check_webhook.
"""
+
+ webhook_secret = getattr(self, "WEBHOOK_TEST_SECRET", None)
+ if webhook_secret is not None:
+ set_bot_config(self.test_user, "webhook_secret", webhook_secret) # nocoverage
+
payload = self.get_payload(fixture_name)
extra["content_type"] = content_type
diff --git a/zerver/lib/webhooks/common.py b/zerver/lib/webhooks/common.py
index f2bfa6f7bc2bf..4589b9696c268 100644
--- a/zerver/lib/webhooks/common.py
+++ b/zerver/lib/webhooks/common.py
@@ -25,6 +25,7 @@
check_send_stream_message_by_id,
send_rate_limited_pm_notification_to_bot_owner,
)
+from zerver.lib.bot_config import ConfigError, get_bot_config
from zerver.lib.exceptions import (
AnomalousWebhookPayloadError,
ErrorCode,
@@ -320,8 +321,45 @@ def parse_multipart_string(body: str) -> dict[str, str]:
return data
+def validate_webhook_delivery(
+ request: HttpRequest, signature_header_name: str, algorithm: str = "sha256"
+) -> None:
+ assert request.user.is_authenticated
+ user_profile = request.user
+ assert isinstance(user_profile, UserProfile)
+
+ try:
+ config = get_bot_config(user_profile)
+ webhook_secret = config.get("webhook_secret", "")
+ except ConfigError:
+ webhook_secret = ""
+
+ if not webhook_secret:
+ return
+
+ signature_header = request.headers.get(signature_header_name, "")
+ signature = signature_header.split("=")[-1] if "=" in signature_header else signature_header
+
+ payload = request.body.decode("utf-8")
+
+ try:
+ validate_webhook_signature(
+ payload=payload,
+ signature=signature,
+ secret=webhook_secret,
+ algorithm=algorithm,
+ )
+ except JsonableError:
+ raise
+ except Exception as err: # nocoverage
+ raise JsonableError(str(err))
+
+
def validate_webhook_signature(
- request: HttpRequest, payload: str, signature: str, algorithm: str = "sha256"
+ payload: str,
+ signature: str,
+ secret: str,
+ algorithm: str = "sha256",
) -> None:
if not settings.VERIFY_WEBHOOK_SIGNATURES: # nocoverage
return
@@ -331,14 +369,10 @@ def validate_webhook_signature(
_("The algorithm '{algorithm}' is not supported.").format(algorithm=algorithm)
)
- webhook_secret: str | None = request.GET.get("webhook_secret")
- if webhook_secret is None:
- raise JsonableError(
- _(
- "The webhook secret is missing. Please set the webhook_secret while generating the URL."
- )
- )
- webhook_secret_bytes = force_bytes(webhook_secret)
+ if not secret:
+ raise JsonableError(_("Webhook secret is not configured for this bot."))
+
+ webhook_secret_bytes = force_bytes(secret)
payload_bytes = force_bytes(payload)
signed_payload = hmac.new(
diff --git a/zerver/tests/test_bots.py b/zerver/tests/test_bots.py
index de5feb4f992a1..dc92dd2666072 100644
--- a/zerver/tests/test_bots.py
+++ b/zerver/tests/test_bots.py
@@ -2305,6 +2305,85 @@ def test_create_incoming_webhook_bot_with_incorrect_service_name(self) -> None:
with self.assertRaises(UserProfile.DoesNotExist):
UserProfile.objects.get(full_name="My Stripe Bot")
+ def test_create_incoming_webhook_bot_with_secret(self) -> None:
+ self.login("hamlet")
+ self.assert_num_bots_equal(0)
+ self.create_bot(
+ bot_type=UserProfile.INCOMING_WEBHOOK_BOT,
+ config_data=orjson.dumps({"webhook_secret": "test-secret-key-123"}).decode(),
+ )
+ self.assert_num_bots_equal(1)
+
+ new_bot = UserProfile.objects.get(full_name="The Bot of Hamlet")
+ config_data = get_bot_config(new_bot)
+ self.assertEqual(config_data["webhook_secret"], "test-secret-key-123")
+
+ def test_create_incoming_webhook_bot_without_secret(self) -> None:
+ self.login("hamlet")
+ self.assert_num_bots_equal(0)
+ self.create_bot(bot_type=UserProfile.INCOMING_WEBHOOK_BOT)
+ self.assert_num_bots_equal(1)
+
+ new_bot = UserProfile.objects.get(full_name="The Bot of Hamlet")
+
+ with self.assertRaisesMessage(ConfigError, "No config data available."):
+ get_bot_config(new_bot)
+
+ def test_patch_incoming_webhook_bot_add_secret(self) -> None:
+ self.login("hamlet")
+ self.create_bot(bot_type=UserProfile.INCOMING_WEBHOOK_BOT)
+
+ bot_email = "hambot-bot@zulip.testserver"
+ bot_id = self.get_bot_user(bot_email).id
+
+ bot_update = {
+ "config_data": orjson.dumps({"webhook_secret": "test-secret-key-123"}).decode()
+ }
+ result = self.client_patch(f"/json/bots/{bot_id}", bot_update)
+ self.assert_json_success(result)
+
+ bot = self.get_bot_user(bot_email)
+ config_data = get_bot_config(bot)
+ self.assertEqual(config_data["webhook_secret"], "test-secret-key-123")
+
+ def test_patch_incoming_webhook_bot_update_secret(self) -> None:
+ self.login("hamlet")
+ self.create_bot(
+ bot_type=UserProfile.INCOMING_WEBHOOK_BOT,
+ config_data=orjson.dumps({"webhook_secret": "old-test-secret-123"}).decode(),
+ )
+
+ bot_email = "hambot-bot@zulip.testserver"
+ bot_id = self.get_bot_user(bot_email).id
+
+ bot_update = {
+ "config_data": orjson.dumps({"webhook_secret": "new-test-secret-123"}).decode()
+ }
+ result = self.client_patch(f"/json/bots/{bot_id}", bot_update)
+ self.assert_json_success(result)
+
+ bot = self.get_bot_user(bot_email)
+ config_data = get_bot_config(bot)
+ self.assertEqual(config_data["webhook_secret"], "new-test-secret-123")
+
+ def test_patch_incoming_webhook_bot_clear_secret(self) -> None:
+ self.login("hamlet")
+ self.create_bot(
+ bot_type=UserProfile.INCOMING_WEBHOOK_BOT,
+ config_data=orjson.dumps({"webhook_secret": "test-secret-key-123"}).decode(),
+ )
+
+ bot_email = "hambot-bot@zulip.testserver"
+ bot_id = self.get_bot_user(bot_email).id
+
+ bot_update = {"config_data": orjson.dumps({"webhook_secret": ""}).decode()}
+ result = self.client_patch(f"/json/bots/{bot_id}", bot_update)
+ self.assert_json_success(result)
+
+ bot = self.get_bot_user(bot_email)
+ config_data = get_bot_config(bot)
+ self.assertEqual(config_data["webhook_secret"], "")
+
def test_get_bot_api_key(self) -> None:
self.login("hamlet")
self.create_bot()
diff --git a/zerver/tests/test_integrations_dev_panel.py b/zerver/tests/test_integrations_dev_panel.py
index ad46910f00f3a..7188affc10bb2 100644
--- a/zerver/tests/test_integrations_dev_panel.py
+++ b/zerver/tests/test_integrations_dev_panel.py
@@ -1,3 +1,5 @@
+import hashlib
+import hmac
from unittest.mock import MagicMock, patch
import orjson
@@ -339,3 +341,155 @@ def test_send_all_webhook_fixture_messages_for_missing_fixtures(
}
self.assertEqual(response.status_code, 404)
self.assertEqual(orjson.loads(response.content), expected_response)
+
+ def test_recalculate_signature_method_not_allowed(self) -> None:
+ target_url = "/devtools/integrations/recalculate_signature"
+ # The endpoint expects a POST request. GET should fail with 405.
+ response = self.client_get(target_url)
+ self.assertEqual(response.status_code, 405)
+ self.assertEqual(orjson.loads(response.content), {"error": "Method not allowed"})
+
+ def test_recalculate_signature_unsupported_integration(self) -> None:
+ target_url = "/devtools/integrations/recalculate_signature"
+ data = {
+ "secret": "my_secret",
+ "payload": '{"event": "ping"}',
+ "integration_name": "unsupported_platform",
+ }
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ expected_response = {
+ "supported": False,
+ "msg": "No signature rules configured for this platform.",
+ }
+ self.assertEqual(orjson.loads(response.content), expected_response)
+
+ def test_recalculate_signature_empty_secret_triggers_clear(self) -> None:
+ target_url = "/devtools/integrations/recalculate_signature"
+ data = {
+ "secret": "",
+ "payload": '{"event": "ping"}',
+ "integration_name": "github",
+ }
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ expected_response = {"supported": True, "clear_signature": True}
+ self.assertEqual(orjson.loads(response.content), expected_response)
+
+ def test_recalculate_signature_success_with_json_payload(self) -> None:
+ target_url = "/devtools/integrations/recalculate_signature"
+ secret = "github_webhook_secret"
+
+ payload = '{\n "zen": "Non-blocking is better than blocking."\n}'
+
+ data = {
+ "secret": secret,
+ "payload": payload,
+ "integration_name": "github ", # Tests trimming behavior
+ }
+
+ # Manually compute the expected HMAC hash of minified JSON
+ minified_payload_bytes = orjson.dumps(orjson.loads(payload))
+ expected_hash = hmac.new(
+ secret.encode(), minified_payload_bytes, hashlib.sha256
+ ).hexdigest()
+
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ expected_response = {
+ "supported": True,
+ "clear_signature": False,
+ "header_key": "X_HUB_SIGNATURE_256",
+ "signature": f"sha256={expected_hash}",
+ }
+ self.assertEqual(orjson.loads(response.content), expected_response)
+
+ def test_recalculate_signature_success_with_non_json_payload(self) -> None:
+ target_url = "/devtools/integrations/recalculate_signature"
+ secret = "github_webhook_secret"
+ payload = "plain-text-payload-string"
+
+ data = {
+ "secret": secret,
+ "payload": payload,
+ "integration_name": "GITHUB",
+ }
+
+ # Falls back to plain text bytes computation upon JSON extraction failure
+ expected_hash = hmac.new(secret.encode(), payload.encode(), hashlib.sha256).hexdigest()
+
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ expected_response = {
+ "supported": True,
+ "clear_signature": False,
+ "header_key": "X_HUB_SIGNATURE_256",
+ "signature": f"sha256={expected_hash}",
+ }
+ self.assertEqual(orjson.loads(response.content), expected_response)
+
+ def test_recalculate_signature_exception_handling(self) -> None:
+ target_url = "/devtools/integrations/recalculate_signature"
+
+ # Sending a malformed request context (e.g. string payload instead of valid json object)
+ # to force the parsing logic down the general exception handling path.
+ response = self.client_post(
+ target_url, "invalid_json_body", content_type="application/json"
+ )
+ self.assertEqual(response.status_code, 400)
+
+ response_data = orjson.loads(response.content)
+ self.assertIn("error", response_data)
+
+ def test_sync_signature_headers_endpoint_success(self) -> None:
+ """Tests the backend counterpart of sync_signature_headers for a valid integration."""
+ target_url = "/devtools/integrations/recalculate_signature"
+ data = {
+ "secret": "my_webhook_secret",
+ "payload": '{"event": "ping"}',
+ "integration_name": "github",
+ }
+
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ response_data = orjson.loads(response.content)
+ self.assertTrue(response_data["supported"])
+ self.assertFalse(response_data["clear_signature"])
+ self.assertEqual(response_data["header_key"], "X_HUB_SIGNATURE_256")
+ self.assertTrue(response_data["signature"].startswith("sha256="))
+
+ def test_sync_signature_headers_endpoint_empty_secret(self) -> None:
+ """Tests that passing an empty secret returns clear_signature=True to clear the UI instantly."""
+ target_url = "/devtools/integrations/recalculate_signature"
+ data = {
+ "secret": "",
+ "payload": '{"event": "ping"}',
+ "integration_name": "github",
+ }
+
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ response_data = orjson.loads(response.content)
+ self.assertTrue(response_data["supported"])
+ self.assertTrue(response_data["clear_signature"])
+
+ def test_sync_signature_headers_endpoint_unsupported(self) -> None:
+ """Tests that an unregistered integration name returns supported=False to drop headers."""
+ target_url = "/devtools/integrations/recalculate_signature"
+ data = {
+ "secret": "secret",
+ "payload": "{}",
+ "integration_name": "some_random_platform",
+ }
+
+ response = self.client_post(target_url, data, content_type="application/json")
+ self.assertEqual(response.status_code, 200)
+
+ response_data = orjson.loads(response.content)
+ self.assertFalse(response_data["supported"])
diff --git a/zerver/tests/test_webhooks_common.py b/zerver/tests/test_webhooks_common.py
index 02db0bdd1ce09..48ea22a889aaf 100644
--- a/zerver/tests/test_webhooks_common.py
+++ b/zerver/tests/test_webhooks_common.py
@@ -14,6 +14,7 @@
from zerver.actions.custom_profile_fields import try_add_realm_custom_profile_field
from zerver.actions.streams import do_rename_stream
from zerver.decorator import webhook_view
+from zerver.lib.bot_config import set_bot_config
from zerver.lib.exceptions import InvalidJSONError, JsonableError
from zerver.lib.request import RequestNotes
from zerver.lib.send_email import FromAddress
@@ -29,6 +30,7 @@
get_service_api_data,
guess_zulip_user_from_external_account,
standardize_headers,
+ validate_webhook_delivery,
validate_webhook_signature,
)
from zerver.models import Client, CustomProfileField, Message, UserProfile
@@ -152,18 +154,14 @@ def test_standardize_headers(self) -> None:
@override_settings(VERIFY_WEBHOOK_SIGNATURES=True)
def test_validate_webhook_signature(self) -> None:
- request = HostRequestMock()
- request.GET = QueryDict("", mutable=True)
-
- # Valid signature
webhook_secret = "test_secret"
payload = '{"key": "value"}'
signature = hmac.new(
force_bytes(webhook_secret), force_bytes(payload), hashlib.sha256
).hexdigest()
- request.GET.update({"webhook_secret": webhook_secret})
- validate_webhook_signature(request, payload, signature)
+ # Valid signature
+ validate_webhook_signature(payload, signature, webhook_secret)
# Invalid signature
invalid_signature = "invalid_signature"
@@ -171,15 +169,47 @@ def test_validate_webhook_signature(self) -> None:
JsonableError,
"Webhook signature verification failed.",
):
- validate_webhook_signature(request, payload, invalid_signature)
+ validate_webhook_signature(payload, invalid_signature, webhook_secret)
- # No webhook_secret parameter
- request.GET.clear()
+ # Missing or empty secret
with self.assertRaisesRegex(
JsonableError,
- "The webhook secret is missing. Please set the webhook_secret while generating the URL.",
+ "Webhook secret is not configured for this bot.",
):
- validate_webhook_signature(request, payload, signature)
+ validate_webhook_signature(payload, signature, secret="")
+
+ @override_settings(VERIFY_WEBHOOK_SIGNATURES=True)
+ def test_validate_webhook_delivery(self) -> None:
+ webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip"))
+ webhook_secret = "test_secret"
+ payload = '{"key": "value"}'
+ signature = hmac.new(
+ force_bytes(webhook_secret), force_bytes(payload), hashlib.sha256
+ ).hexdigest()
+
+ set_bot_config(webhook_bot, "webhook_secret", webhook_secret)
+ request = HostRequestMock(meta_data={"HTTP_X_HUB_SIGNATURE_256": f"sha256={signature}"})
+ request.user = webhook_bot
+ request.GET = QueryDict("", mutable=True)
+ request._body = force_bytes(payload)
+
+ # Valid signature
+ validate_webhook_delivery(request, "X_HUB_Signature_256")
+
+ # Invalid signature
+ request.META["HTTP_X_HUB_SIGNATURE_256"] = "sha256=invalid_signature"
+ del request.headers
+ with self.assertRaisesRegex(
+ JsonableError,
+ "Webhook signature verification failed.",
+ ):
+ validate_webhook_delivery(request, "X_HUB_Signature_256")
+
+ # No webhook_secret configured for this bot skips validation
+ set_bot_config(webhook_bot, "webhook_secret", "")
+ request.META["HTTP_X_HUB_SIGNATURE_256"] = f"sha256={signature}"
+ del request.headers
+ validate_webhook_delivery(request, "X_HUB_Signature_256")
def test_check_send_webhook_message_returns_id(self) -> None:
webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip"))
diff --git a/zerver/views/development/integrations.py b/zerver/views/development/integrations.py
index a3ec1ce71221c..6c524999d13f7 100644
--- a/zerver/views/development/integrations.py
+++ b/zerver/views/development/integrations.py
@@ -1,12 +1,17 @@
+import hashlib
+import hmac
import os
+from collections.abc import Callable
from contextlib import suppress
from typing import TYPE_CHECKING, Any
import orjson
-from django.http import HttpRequest, HttpResponse
+from django.http import HttpRequest, HttpResponse, JsonResponse
from django.http.response import HttpResponseBase
from django.shortcuts import render
from django.test import Client
+from django.utils.encoding import force_bytes
+from django.views.decorators.csrf import csrf_exempt
from pydantic import Json
from zerver.lib.exceptions import JsonableError, ResourceNotFoundError
@@ -156,3 +161,63 @@ def send_all_webhook_fixture_messages(
}
)
return json_success(request, data={"responses": responses})
+
+
+def format_github_signature(secret_bytes: bytes, payload_bytes: bytes) -> tuple[str, str]:
+ """Formats signature header following X-Hub-Signature-256 standard."""
+ signed_payload = hmac.new(secret_bytes, payload_bytes, hashlib.sha256).hexdigest()
+ return "X_HUB_SIGNATURE_256", f"sha256={signed_payload}"
+
+
+SIGNATURE_REGISTRY: dict[str, Callable[[bytes, bytes], tuple[str, str]]] = {
+ "github": format_github_signature
+}
+
+
+@csrf_exempt
+def recalculate_signature(request: HttpRequest) -> JsonResponse:
+ """
+ Unified endpoint invoked by the frontend UI dev panel to dynamically compute
+ and format signature header blocks based on the integration.
+ """
+ if request.method != "POST":
+ return JsonResponse({"error": "Method not allowed"}, status=405)
+
+ try:
+ data = orjson.loads(request.body)
+ secret = data.get("secret", "")
+ payload_string = data.get("payload", "")
+ integration_name = data.get("integration_name", "").lower().strip()
+
+ # Check if the integration has signature management registered
+ if integration_name not in SIGNATURE_REGISTRY:
+ return JsonResponse(
+ {"supported": False, "msg": "No signature rules configured for this platform."}
+ )
+
+ if not secret:
+ return JsonResponse({"supported": True, "clear_signature": True})
+
+ # Normalize and minify JSON formats for crypto verification stability
+ try:
+ payload_bytes = orjson.dumps(orjson.loads(payload_string))
+ except Exception:
+ payload_bytes = force_bytes(payload_string)
+
+ webhook_secret_bytes = force_bytes(secret)
+
+ # Execute the registered structural format strategy
+ formatter = SIGNATURE_REGISTRY[integration_name]
+ header_key, header_value = formatter(webhook_secret_bytes, payload_bytes)
+
+ return JsonResponse(
+ {
+ "supported": True,
+ "clear_signature": False,
+ "header_key": header_key,
+ "signature": header_value,
+ }
+ )
+
+ except Exception:
+ return JsonResponse({"error": "Invalid request payload."}, status=400)
diff --git a/zerver/webhooks/github/tests.py b/zerver/webhooks/github/tests.py
index 5f5909e4419b0..5677385b672f6 100644
--- a/zerver/webhooks/github/tests.py
+++ b/zerver/webhooks/github/tests.py
@@ -1,7 +1,9 @@
from unittest.mock import patch
import orjson
+from django.test import override_settings
+from zerver.lib.bot_config import set_bot_config
from zerver.lib.message import truncate_topic
from zerver.lib.test_classes import WebhookTestCase
from zerver.lib.webhooks.git import COMMITS_LIMIT
@@ -22,6 +24,9 @@
class GitHubWebhookTest(WebhookTestCase):
+ WEBHOOK_SIGNATURE_HEADER = "X_HUB_Signature_256"
+ WEBHOOK_TEST_SECRET = "testingthis"
+
def test_ping_event(self) -> None:
expected_message = "GitHub webhook has been successfully configured by TomaszKolek."
self.check_webhook("ping", TOPIC_REPO, expected_message)
@@ -853,6 +858,49 @@ def test_issue_comment_silent_mention_with_multiple_matches(self) -> None:
expected_message = "baxterthehacker [commented](https://github.com/baxterthehacker/public-repo/issues/2#issuecomment-99262140) on [issue #2](https://github.com/baxterthehacker/public-repo/issues/2):\n\n``` quote\nYou are totally right! I'll get this fixed right away.\n```"
self.check_webhook("issue_comment", TOPIC_ISSUE, expected_message)
+ def test_github_webhook_bad_signature(self) -> None:
+ with override_settings(VERIFY_WEBHOOK_SIGNATURES=True):
+ url = self.build_webhook_url()
+ set_bot_config(self.test_user, "webhook_secret", self.WEBHOOK_TEST_SECRET)
+
+ result = self.client_post(
+ url,
+ self.get_payload("ping"),
+ content_type="application/json",
+ HTTP_X_HUB_SIGNATURE_256="sha256=completely_invalid_hash_value",
+ )
+ self.assert_json_error(result, "Webhook signature verification failed.")
+
+ def test_github_webhook_signature_disabled_skips_validation(self) -> None:
+ """Verifies that when VERIFY_WEBHOOK_SIGNATURES is explicitly disabled,
+ requests pass through even if the signature value is completely bogus.
+ """
+ with override_settings(VERIFY_WEBHOOK_SIGNATURES=False):
+ expected_message = "GitHub webhook has been successfully configured by TomaszKolek."
+ self.check_webhook(
+ "ping",
+ TOPIC_REPO,
+ expected_message,
+ HTTP_X_HUB_SIGNATURE_256="sha256=invalid_hash",
+ )
+
+ def test_github_webhook_valid_signature_success(self) -> None:
+ """Verifies that a mathematically correct HMAC signature passes
+ cleanly when verification enforcement is active."""
+ expected_message = "GitHub webhook has been successfully configured by TomaszKolek."
+
+ with override_settings(VERIFY_WEBHOOK_SIGNATURES=True):
+ self.check_webhook("ping", TOPIC_REPO, expected_message)
+
+ def test_github_webhook_missing_secret(self) -> None:
+ """Verifies that if no webhook secret is configured for the bot,
+ the request is processed normally without requiring signature verification."""
+
+ with override_settings(VERIFY_WEBHOOK_SIGNATURES=True):
+ set_bot_config(self.test_user, "webhook_secret", "")
+ expected_message = "GitHub webhook has been successfully configured by TomaszKolek."
+ self.check_webhook("ping", TOPIC_REPO, expected_message)
+
class GitHubSponsorsHookTests(WebhookTestCase):
URL_TEMPLATE = "/api/v1/external/githubsponsors?stream={stream}&api_key={api_key}"
diff --git a/zerver/webhooks/github/view.py b/zerver/webhooks/github/view.py
index c1a13b5ad0daf..b276df37fd713 100644
--- a/zerver/webhooks/github/view.py
+++ b/zerver/webhooks/github/view.py
@@ -22,6 +22,7 @@
get_event_header,
get_setup_webhook_message,
guess_zulip_user_from_external_account,
+ validate_webhook_delivery,
)
from zerver.lib.webhooks.git import (
CONTENT_MESSAGE_TEMPLATE,
@@ -1215,6 +1216,8 @@ def api_github_webhook(
directly to the X-GitHub-Event header's event, but we sometimes
refine it based on the payload.
"""
+ validate_webhook_delivery(request, "X_HUB_Signature_256", "sha256")
+
header_event = get_event_header(request, "X-GitHub-Event", "GitHub")
# Ignore events from private repositories if the URL option is set
diff --git a/zproject/dev_urls.py b/zproject/dev_urls.py
index c57717f9fe2a1..d637ca8e497c5 100644
--- a/zproject/dev_urls.py
+++ b/zproject/dev_urls.py
@@ -24,6 +24,7 @@
check_send_webhook_fixture_message,
dev_panel,
get_fixtures,
+ recalculate_signature,
send_all_webhook_fixture_messages,
)
from zerver.views.development.registration import (
@@ -98,6 +99,10 @@
"devtools/integrations/send_all_webhook_fixture_messages", send_all_webhook_fixture_messages
),
path("devtools/integrations//fixtures", get_fixtures),
+ path(
+ "devtools/integrations/recalculate_signature",
+ recalculate_signature,
+ ),
path("config-error/", config_error, name="config_error"),
# Special endpoint to remove all the server-side caches.
path("flush_caches", remove_caches),