Cleanded up commit history

package.json

@@ -114,6 +114,7 @@
     "@types/convert-source-map": "^2.0.3",
     "@types/css-tree": "^3.2.0",
     "@types/eslint-config-prettier": "^6.11.3",
+    "@types/estree": "^1.0.9",
     "@types/gtag.js": "^0.0.20",
     "@types/is-url": "^1.2.32",
     "@types/jquery": "^4.0.0",

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

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
@@ -1196,6 +1201,15 @@ def is_enabled_in_catalog(self) -> bool:
     | hubot_integration_names
 )
 
+WEBHOOK_SIGNATURE_CONFIGS: dict[str, WebhookSignatureConfig] = {
+    "github": WebhookSignatureConfig(
+        integration_name="github",
+        header="X_HUB_SIGNATURE_256",
+        algorithm="sha256",
+        prefix="sha256=",
+    ),
+}
+
 # Add integrations that are not meant to have example screenshots here
 INTEGRATIONS_WITHOUT_SCREENSHOTS = (
     # Integration frameworks

zerver/lib/test_classes.py

@@ -35,6 +35,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 +55,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 +95,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 +2552,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 +2570,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"
         )
@@ -2653,26 +2673,38 @@ def check_webhook(
         """
         self.subscribe(self.test_user, self.channel_name)
 
+        url = self.url
+        webhook_secret = self.WEBHOOK_TEST_SECRET
+        config = WEBHOOK_SIGNATURE_CONFIGS.get(self.webhook_dir_name.lower())
+
         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:
+            header_val = compute_webhook_signature(
+                force_bytes(webhook_secret),
+                force_bytes(payload),
+                config,
+            )
+
+            django_header = "HTTP_" + config.header.upper().replace("-", "_")
+            if django_header not in extra:
+                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,
-            )
-        except EmptyResponseError:
-            if expect_noop:
-                return
-            else:
-                raise AssertionError(
-                    "No message was sent. Pass expect_noop=True if this is intentional."
-                )
+        with self.settings(VERIFY_WEBHOOK_SIGNATURES=self.VERIFY_WEBHOOK_SIGNATURES):
+            try:
+                msg = self.send_webhook_payload(self.test_user, url, payload, **extra)
+            except EmptyResponseError:
+                if expect_noop:
+                    return
+                else:
+                    raise AssertionError(
+                        "No message was sent. Pass expect_noop=True if this is intentional."
+                    )
 
         if expect_noop:
             raise Exception(
@@ -2718,6 +2750,7 @@ def send_and_test_private_message(
         Most webhooks send to streams, and you will want to look at
         check_webhook.
         """
+
         payload = self.get_payload(fixture_name)
         extra["content_type"] = content_type
 
@@ -2728,12 +2761,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

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:
+    if config.algorithm not in hashlib.algorithms_available:
         raise AssertionError(
-            _("The algorithm '{algorithm}' is not supported.").format(algorithm=algorithm)
+            _("The algorithm '{algorithm}' is not supported.").format(algorithm=config.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)
-    payload_bytes = force_bytes(payload)
+    signature_header = request.headers.get(config.header)
+    if not signature_header:
+        return
 
-    signed_payload = hmac.new(
-        webhook_secret_bytes,
-        payload_bytes,
-        algorithm,
-    ).hexdigest()
+    try:
+        bot_config = get_bot_config(user_profile)
+    except ConfigError:
+        return
 
-    if not constant_time_compare(signed_payload, signature):
+    webhook_secret = bot_config.get(
+        WEBHOOK_SECRET_TOKEN_KEY.format(integration_name=config.integration_name.lower())
+    )
+
+    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,