From cae52255970630b087426544d6b21de0edc33f0f Mon Sep 17 00:00:00 2001 From: Srinandha Murugesan Date: Fri, 14 Aug 2026 06:06:49 +0000 Subject: [PATCH 1/2] webhook: Add generic HMAC signature verification utilities. Modify signature validation function to support HMAC verification for incoming webhooks. Refactor WebhookTestCase and webhook_view decorator to support config. Co-authored-by: Srinandha Murugesan Co-authored-by: Isaiah Marte Co-authored-by: Akshaj Katkuri --- zerver/decorator.py | 8 ++ zerver/lib/test_classes.py | 66 +++++++++++--- zerver/lib/webhooks/common.py | 82 ++++++++++++----- zerver/tests/test_webhooks_common.py | 128 +++++++++++++++++++++++---- 4 files changed, 234 insertions(+), 50 deletions(-) diff --git a/zerver/decorator.py b/zerver/decorator.py index 779b0f9fc6c4d..7a63586f83488 100644 --- a/zerver/decorator.py +++ b/zerver/decorator.py @@ -54,7 +54,9 @@ from zerver.lib.utils import has_api_key_format from zerver.lib.webhooks.common import ( MissingHTTPEventHeaderError, + WebhookSignatureConfig, notify_bot_owner_about_invalid_json, + validate_webhook_signature, ) from zerver.models import UserProfile from zerver.models.clients import get_client @@ -372,6 +374,7 @@ def webhook_view( webhook_client_name: str, notify_bot_owner_on_invalid_json: bool = True, all_event_types: Sequence[str] | None = None, + signature_config: WebhookSignatureConfig | None = None, ) -> Callable[[Callable[..., HttpResponse]], Callable[..., HttpResponse]]: # Unfortunately, callback protocols are insufficient for this: # https://mypy.readthedocs.io/en/stable/protocols.html#callback-protocols @@ -390,6 +393,11 @@ def _wrapped_func_arguments( allow_webhook_access=True, client_name=full_webhook_client_name(webhook_client_name), ) + validate_webhook_signature( + request, + user_profile, + signature_config, + ) request_notes = RequestNotes.get_notes(request) request_notes.is_webhook_view = True diff --git a/zerver/lib/test_classes.py b/zerver/lib/test_classes.py index 755f52a536a78..bf0f0b1bed25a 100644 --- a/zerver/lib/test_classes.py +++ b/zerver/lib/test_classes.py @@ -1,5 +1,6 @@ import asyncio import base64 +import json import os import re import shutil @@ -35,6 +36,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,9 +56,11 @@ 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 +from zerver.lib.integrations import WEBHOOK_SIGNATURE_CONFIGS from zerver.lib.mdiff import diff_strings from zerver.lib.message import access_message from zerver.lib.notification_data import UserMessageNotificationsData @@ -92,8 +96,10 @@ from zerver.lib.upload import upload_message_attachment_from_request from zerver.lib.user_groups import get_system_user_group_for_user from zerver.lib.webhooks.common import ( + WEBHOOK_SECRET_TOKEN_KEY, call_fixture_to_headers, check_send_webhook_message, + compute_webhook_signature, standardize_headers, ) from zerver.models import ( @@ -2547,6 +2553,8 @@ class WebhookTestCase(ZulipTestCase): DEFAULT_URL_TEMPLATE: str = ( "/api/v1/external/{webhook_dir_name}?stream={stream}&api_key={api_key}" ) + WEBHOOK_TEST_SECRET: str | None = None + VERIFY_WEBHOOK_SIGNATURES: bool = True def get_webhook_dir_name(self) -> str: module_parts = self.__module__.split(".") @@ -2563,6 +2571,19 @@ def setUp(self) -> None: self.url_template = self.URL_TEMPLATE or self.DEFAULT_URL_TEMPLATE self.url = self.build_webhook_url() + if self.WEBHOOK_TEST_SECRET is not None: + config = WEBHOOK_SIGNATURE_CONFIGS.get(self.webhook_dir_name.lower()) + if config is None: + raise AssertionError( + f"WEBHOOK_TEST_SECRET was set for '{self.webhook_dir_name}', " + f"but no WebhookSignatureConfig is registered in WEBHOOK_SIGNATURE_CONFIGS." + ) + set_bot_config( + self.test_user, + WEBHOOK_SECRET_TOKEN_KEY.format(integration_name=self.webhook_dir_name.lower()), + self.WEBHOOK_TEST_SECRET, + ) + function = import_string( f"zerver.webhooks.{self.webhook_dir_name}.view.api_{self.webhook_dir_name}_webhook" ) @@ -2661,22 +2682,42 @@ def check_webhook( """ self.subscribe(self.test_user, self.channel_name) + webhook_secret = self.WEBHOOK_TEST_SECRET + config = WEBHOOK_SIGNATURE_CONFIGS.get(self.webhook_dir_name.lower()) + if custom_payload is not None: payload = custom_payload else: payload = self.get_payload(fixture_name) + if content_type is not None: extra["content_type"] = content_type + + if webhook_secret is not None and config is not None: + if isinstance(payload, dict): + webhook_payload = json.dumps(payload) + else: + webhook_payload = payload + header_val = compute_webhook_signature( + force_bytes(webhook_secret), + force_bytes(webhook_payload), + config, + ) + + django_header = "HTTP_" + config.header.upper().replace("-", "_") + extra[django_header] = header_val + 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, - payload, - **extra, - ) + with self.settings(VERIFY_WEBHOOK_SIGNATURES=self.VERIFY_WEBHOOK_SIGNATURES): + msg = self.send_webhook_payload( + self.test_user, + self.url, + payload, + **extra, + ) except EmptyResponseError: if expect_noop: return @@ -2739,12 +2780,13 @@ def send_and_test_private_message( if sender is None: sender = self.test_user - msg = self.send_webhook_payload( - sender, - self.url, - payload, - **extra, - ) + with self.settings(VERIFY_WEBHOOK_SIGNATURES=self.VERIFY_WEBHOOK_SIGNATURES): + msg = self.send_webhook_payload( + sender, + self.url, + payload, + **extra, + ) self.assertEqual(msg.content, expected_message) return msg diff --git a/zerver/lib/webhooks/common.py b/zerver/lib/webhooks/common.py index ac2d8ba7e7a16..baff587b07945 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, @@ -58,6 +59,8 @@ SETUP_MESSAGE_TEMPLATE = "{integration} webhook has been successfully configured" SETUP_MESSAGE_USER_PART = " by {user_name}" +WEBHOOK_SECRET_TOKEN_KEY = "{integration_name}:webhook_secret_token" + OptionalUserSpecifiedTopicStr: TypeAlias = Annotated[str | None, ApiParamConfig("topic")] @@ -74,6 +77,16 @@ class WebhookConfigOption: validator: Callable[[str, str], str | bool | None] +@dataclass(frozen=True) +class WebhookSignatureConfig: + integration_name: str + header: str + algorithm: str = "sha256" + prefix: str = "" + # This will override the default compute_webhook_signature function if provided for unique formats + custom_formatter: Callable[[str], str] | None = None + + @dataclass class WebhookUrlOption: name: str @@ -321,36 +334,65 @@ def parse_multipart_string(body: str) -> dict[str, str]: def validate_webhook_signature( - request: HttpRequest, payload: str, signature: str, algorithm: str = "sha256" + request: HttpRequest, + user_profile: UserProfile, + config: WebhookSignatureConfig | None, ) -> None: - if not settings.VERIFY_WEBHOOK_SIGNATURES: # nocoverage + if not settings.VERIFY_WEBHOOK_SIGNATURES or not config: return - if algorithm not in hashlib.algorithms_available: - raise AssertionError( - _("The algorithm '{algorithm}' is not supported.").format(algorithm=algorithm) - ) - - webhook_secret: str | None = request.GET.get("webhook_secret") - if webhook_secret is None: + if config.algorithm not in hashlib.algorithms_available: raise JsonableError( - _( - "The webhook secret is missing. Please set the webhook_secret while generating the URL." - ) + _("The algorithm '{algorithm}' is not supported.").format(algorithm=config.algorithm) ) - webhook_secret_bytes = force_bytes(webhook_secret) - payload_bytes = force_bytes(payload) - signed_payload = hmac.new( - webhook_secret_bytes, - payload_bytes, - algorithm, - ).hexdigest() + signature_header = request.headers.get(config.header) + if not signature_header: + return + + try: + bot_config = get_bot_config(user_profile) + except ConfigError: + raise JsonableError(_("Webhook secret is not configured for this bot.")) + + webhook_secret = bot_config.get( + WEBHOOK_SECRET_TOKEN_KEY.format(integration_name=config.integration_name.lower()) + ) - if not constant_time_compare(signed_payload, signature): + if not webhook_secret or not webhook_secret.strip(): + raise JsonableError(_("Webhook secret is not configured for this bot.")) + + payload = request.body.decode("utf-8") + + expected_header_val = compute_webhook_signature( + force_bytes(webhook_secret), + force_bytes(payload), + config, + ) + if not constant_time_compare(expected_header_val, signature_header): raise JsonableError(_("Webhook signature verification failed.")) +def compute_webhook_signature( + secret_bytes: bytes, + payload_bytes: bytes, + config: WebhookSignatureConfig, +) -> str: + """Computes and formats the HMAC signature for a webhook payload.""" + signer = hmac.new( + secret_bytes, + payload_bytes, + config.algorithm, + ) + digest = signer.hexdigest() + + if config.custom_formatter is not None: + digest = config.custom_formatter(digest) + if config.prefix: + return f"{config.prefix}{digest}" + return digest + + def guess_zulip_user_from_external_account( realm: Realm, external_username: str, diff --git a/zerver/tests/test_webhooks_common.py b/zerver/tests/test_webhooks_common.py index 02db0bdd1ce09..240e0ee6e4cee 100644 --- a/zerver/tests/test_webhooks_common.py +++ b/zerver/tests/test_webhooks_common.py @@ -1,5 +1,3 @@ -import hashlib -import hmac from types import SimpleNamespace from unittest.mock import MagicMock, patch @@ -14,6 +12,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 ConfigError, set_bot_config from zerver.lib.exceptions import InvalidJSONError, JsonableError from zerver.lib.request import RequestNotes from zerver.lib.send_email import FromAddress @@ -22,9 +21,12 @@ from zerver.lib.webhooks.common import ( INVALID_JSON_MESSAGE, MISSING_EVENT_HEADER_MESSAGE, + WEBHOOK_SECRET_TOKEN_KEY, MissingHTTPEventHeaderError, + WebhookSignatureConfig, call_fixture_to_headers, check_send_webhook_message, + compute_webhook_signature, get_event_header, get_service_api_data, guess_zulip_user_from_external_account, @@ -152,34 +154,124 @@ 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_bot = get_user("webhook-bot@zulip.com", get_realm("zulip")) webhook_secret = "test_secret" + config = WebhookSignatureConfig( + integration_name="github", + header="X-Hub-Signature-256", + algorithm="sha256", + prefix="sha256=", + ) payload = '{"key": "value"}' - signature = hmac.new( - force_bytes(webhook_secret), force_bytes(payload), hashlib.sha256 - ).hexdigest() + signature = compute_webhook_signature( + force_bytes(webhook_secret), force_bytes(payload), config + ) - request.GET.update({"webhook_secret": webhook_secret}) - validate_webhook_signature(request, payload, signature) + # Unsupported algorithm check + config_invalid_algo = WebhookSignatureConfig( + integration_name="github", + header="X-Hub-Signature-256", + algorithm="nonexistent_algorithm", + ) + request = HostRequestMock(meta_data={}) + request.user = webhook_bot + request.GET = QueryDict("", mutable=True) + request._body = force_bytes(payload) + with self.assertRaisesRegex( + JsonableError, + "The algorithm 'nonexistent_algorithm' is not supported.", + ): + validate_webhook_signature(request, webhook_bot, config_invalid_algo) + + # Missing header early return check + request = HostRequestMock(meta_data={}) + request.user = webhook_bot + request.GET = QueryDict("", mutable=True) + request._body = force_bytes(payload) + validate_webhook_signature(request, webhook_bot, config) - # Invalid signature - invalid_signature = "invalid_signature" + # ConfigError early return check + request = HostRequestMock(meta_data={"HTTP_X_HUB_SIGNATURE_256": signature}) + request.user = webhook_bot + request.GET = QueryDict("", mutable=True) + request._body = force_bytes(payload) + with ( + patch("zerver.lib.webhooks.common.get_bot_config", side_effect=ConfigError), + self.assertRaisesRegex( + JsonableError, + "Webhook secret is not configured for this bot.", + ), + ): + validate_webhook_signature(request, webhook_bot, config) + + # Unconfigured secret initial pass + request = HostRequestMock(meta_data={"HTTP_X_HUB_SIGNATURE_256": signature}) + request.user = webhook_bot + request.GET = QueryDict("", mutable=True) + request._body = force_bytes(payload) + with ( + patch("zerver.lib.webhooks.common.get_bot_config", return_value={}), + self.assertRaisesRegex( + JsonableError, + "Webhook secret is not configured for this bot.", + ), + ): + validate_webhook_signature(request, webhook_bot, config) + + # Valid signature with configured token key + set_bot_config( + webhook_bot, WEBHOOK_SECRET_TOKEN_KEY.format(integration_name="github"), webhook_secret + ) + request = HostRequestMock(meta_data={"HTTP_X_HUB_SIGNATURE_256": signature}) + request.user = webhook_bot + request.GET = QueryDict("", mutable=True) + request._body = force_bytes(payload) + validate_webhook_signature(request, webhook_bot, config) + + # Invalid signature check + request.META["HTTP_X_HUB_SIGNATURE_256"] = "sha256=invalid_signature" + del request.headers with self.assertRaisesRegex( JsonableError, "Webhook signature verification failed.", ): - validate_webhook_signature(request, payload, invalid_signature) + validate_webhook_signature(request, webhook_bot, config) - # No webhook_secret parameter - request.GET.clear() + # Missing secret check using token key + set_bot_config(webhook_bot, WEBHOOK_SECRET_TOKEN_KEY.format(integration_name="github"), "") + request.META["HTTP_X_HUB_SIGNATURE_256"] = signature + del request.headers 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(request, webhook_bot, config=config) + + def test_compute_webhook_signature_formatter_and_prefix(self) -> None: + # Tests the custom_formatter + config_formatter = WebhookSignatureConfig( + integration_name="test", + header="X-Test-Signature", + custom_formatter=lambda d: f"sha256={d.upper()}", + ) + sig_formatter = compute_webhook_signature(b"secret", b"payload", config_formatter) + self.assertTrue(sig_formatter.startswith("sha256=")) + + # Tests prefix + config_prefix = WebhookSignatureConfig( + integration_name="test", + header="X-Test-Signature", + prefix="sha256=", + ) + sig_prefix = compute_webhook_signature(b"secret", b"payload", config_prefix) + self.assertTrue(sig_prefix.startswith("sha256=")) + + config_default = WebhookSignatureConfig( + integration_name="test", + header="X-Test-Signature", + ) + sig_default = compute_webhook_signature(b"secret", b"payload", config_default) + self.assertFalse(sig_default.startswith("sha256=")) def test_check_send_webhook_message_returns_id(self) -> None: webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip")) From fe88313dcd1036d27eb29ce4178ed841c5305c5d Mon Sep 17 00:00:00 2001 From: Srinandha Murugesan Date: Fri, 14 Aug 2026 06:07:21 +0000 Subject: [PATCH 2/2] webhook/github: Add GitHub signature verification using bot config. Verify incoming GitHub webhook payloads against signature headers using the secret key stored in BotConfigData. Add automated tests for missing, malformed, and valid signatures. Co-authored-by: Srinandha Murugesan Co-authored-by: Isaiah Marte Co-authored-by: Akshaj Katkuri --- zerver/lib/integrations.py | 22 ++++++++++++++++- zerver/webhooks/github/tests.py | 43 +++++++++++++++++++++++++++++++++ zerver/webhooks/github/view.py | 8 +++++- 3 files changed, 71 insertions(+), 2 deletions(-) diff --git a/zerver/lib/integrations.py b/zerver/lib/integrations.py index d6e24f30cc769..85a7a21cec166 100644 --- a/zerver/lib/integrations.py +++ b/zerver/lib/integrations.py @@ -12,7 +12,12 @@ from typing_extensions import override from zerver.lib.storage import static_path -from zerver.lib.webhooks.common import PresetUrlOption, WebhookConfigOption, WebhookUrlOption +from zerver.lib.webhooks.common import ( + PresetUrlOption, + WebhookConfigOption, + WebhookSignatureConfig, + WebhookUrlOption, +) from zerver.webhooks import fixtureless_integrations """This module declares all of the (documented) integrations available @@ -1211,6 +1216,21 @@ def is_enabled_in_catalog(self) -> bool: | hubot_integration_names ) + +def compute_sha256_signature_prefix(digest: str) -> str: + return f"sha256={digest}" + + +WEBHOOK_SIGNATURE_CONFIGS: dict[str, WebhookSignatureConfig] = { + "github": WebhookSignatureConfig( + integration_name="github", + header="X_HUB_SIGNATURE_256", + algorithm="sha256", + prefix="sha256=", + custom_formatter=compute_sha256_signature_prefix, + ), +} + # Add integrations that are not meant to have example screenshots here INTEGRATIONS_WITHOUT_SCREENSHOTS = ( # Integration frameworks diff --git a/zerver/webhooks/github/tests.py b/zerver/webhooks/github/tests.py index 0c3e746c982e3..edf32a8974f13 100644 --- a/zerver/webhooks/github/tests.py +++ b/zerver/webhooks/github/tests.py @@ -2,6 +2,7 @@ import orjson +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 @@ -19,9 +20,13 @@ TOPIC_DISCUSSION_ANSWERS = "webhook-tester discussion #5: Understanding Project Direc..." TOPIC_DISCUSSION_COMMENT = "testing-gh discussion #20: Lets discuss" TOPIC_SPONSORS = "sponsors" +WEBHOOK_SECRET = "testingthis" class GitHubWebhookTest(WebhookTestCase): + WEBHOOK_TEST_SECRET: str | None = WEBHOOK_SECRET + VERIFY_WEBHOOK_SIGNATURES: bool = True + def test_ping_event(self) -> None: expected_message = "GitHub webhook has been successfully configured by TomaszKolek." self.check_webhook("ping", TOPIC_REPO, expected_message) @@ -873,6 +878,44 @@ 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 self.settings(VERIFY_WEBHOOK_SIGNATURES=self.VERIFY_WEBHOOK_SIGNATURES): + result = self.client_post( + self.url, + info=self.get_payload("ping"), + content_type="application/json", + HTTP_X_GITHUB_EVENT="ping", + 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. + """ + self.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_missing_secret(self) -> None: + """Verifies that if no webhook secret is configured for the bot, + the request fails with a JsonableError.""" + set_bot_config(self.test_user, "github:webhook_secret_token", "") + with self.settings(VERIFY_WEBHOOK_SIGNATURES=self.VERIFY_WEBHOOK_SIGNATURES): + result = self.client_post( + self.url, + info=self.get_payload("ping"), + content_type="application/json", + HTTP_X_GITHUB_EVENT="ping", + HTTP_X_HUB_SIGNATURE_256="sha256=placeholder_hash_value", + ) + self.assert_json_error(result, "Webhook secret is not configured for this bot.") + 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 844dab93bef87..93bd26388c5f1 100644 --- a/zerver/webhooks/github/view.py +++ b/zerver/webhooks/github/view.py @@ -9,6 +9,7 @@ from zerver.decorator import log_unsupported_webhook_event, webhook_view from zerver.lib.exceptions import UnsupportedWebhookEventTypeError from zerver.lib.external_accounts import DEFAULT_EXTERNAL_ACCOUNTS +from zerver.lib.integrations import WEBHOOK_SIGNATURE_CONFIGS from zerver.lib.markdown.fenced_code import get_unused_fence from zerver.lib.mention import silent_mention_syntax_for_user from zerver.lib.partial import partial @@ -1179,7 +1180,12 @@ def get_topic_based_on_type(payload: WildValue, event: str) -> str: ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys()) -@webhook_view("GitHub", notify_bot_owner_on_invalid_json=True, all_event_types=ALL_EVENT_TYPES) +@webhook_view( + "GitHub", + notify_bot_owner_on_invalid_json=True, + all_event_types=ALL_EVENT_TYPES, + signature_config=WEBHOOK_SIGNATURE_CONFIGS["github"], +) @typed_endpoint def api_github_webhook( request: HttpRequest,