diff --git a/templates/zerver/development/integrations_dev_panel.html b/templates/zerver/development/integrations_dev_panel.html
index b948daf30bbaf..2f9e5f6e076d8 100644
--- a/templates/zerver/development/integrations_dev_panel.html
+++ b/templates/zerver/development/integrations_dev_panel.html
@@ -69,6 +69,10 @@
+
+
+
+
diff --git a/web/src/bot_type_values.ts b/web/src/bot_type_values.ts
index a93e47a8ad333..4a878d73c9241 100644
--- a/web/src/bot_type_values.ts
+++ b/web/src/bot_type_values.ts
@@ -1,8 +1,9 @@
// Bot type integer values from the API.
-export const GENERIC_BOT_TYPE = 1;
+export const GENERIC_BOT_TYPE_INT = 1;
export const INCOMING_WEBHOOK_BOT_TYPE_INT = 2;
export const OUTGOING_WEBHOOK_BOT_TYPE_INT = 3;
// String forms used as HTML form values.
+export const GENERIC_BOT_TYPE = "1";
export const OUTGOING_WEBHOOK_BOT_TYPE = "3";
export const EMBEDDED_BOT_TYPE = "4";
diff --git a/web/src/portico/integrations_dev_panel.ts b/web/src/portico/integrations_dev_panel.ts
index 8f1b7d5965a2f..70015a93b8fd9 100644
--- a/web/src/portico/integrations_dev_panel.ts
+++ b/web/src/portico/integrations_dev_panel.ts
@@ -25,6 +25,7 @@ type HTMLSelectOneElement = HTMLSelectElement & {type: "select-one"};
type ClearHandlers = {
stream_name: string;
topic_name: string;
+ webhook_secret: string;
URL: string;
results_notice: string;
bot_name: () => void;
@@ -47,6 +48,8 @@ const integrations_api_response_schema = z.object({
result: z.string(),
});
+let last_computed_header_key: string | null = null; // Tracks the current signature header for auto-clearing when switching integrations
+
type ServerResponse = z.infer;
const loaded_fixtures = new Map();
@@ -56,6 +59,7 @@ const url_base = "/api/v1/external/";
const clear_handlers: ClearHandlers = {
stream_name: "#stream_name",
topic_name: "#topic_name",
+ webhook_secret: "#webhook_secret",
URL: "#URL",
results_notice: "#results_notice",
bot_name() {
@@ -180,6 +184,8 @@ function load_fixture_body(fixture_name: string): void {
null,
4,
);
+ const webhook_secret = $("input#webhook_secret").val()!;
+ sync_signature_headers(integration_name, webhook_secret);
return;
}
@@ -210,8 +216,8 @@ function load_fixture_options(integration_name: string): void {
function update_url(): void {
/* Construct the URL that the webhook should be targeting, using
- the bot's API key and the integration name. The stream and topic
- are both optional, and for the sake of completeness, it should be
+ the bot's API key, the integration name, and webhook secret. The stream, topic,
+ and webhook secret are all optional, and for the sake of completeness, it should be
noted that the topic is irrelevant without specifying the
stream. */
const url_field = $("input#URL")[0];
@@ -231,11 +237,83 @@ function update_url(): void {
params.set("topic", topic_name);
}
}
+ const webhook_secret = $("input#webhook_secret").val()!;
const url = `${url_base}${integration_name}?${params.toString()}`;
url_field!.value = url;
+
+ sync_signature_headers(integration_name, webhook_secret);
}
+}
- return;
+function sync_signature_headers(integration_name: string, webhook_secret: string): void {
+ const $custom_headers_field = $("textarea#custom_http_headers");
+ const current_headers_raw = $custom_headers_field.val()?.toString().trim() ?? "";
+
+ let headers_object: Record = {};
+ if (current_headers_raw !== "") {
+ try {
+ headers_object = z
+ .record(z.string(), z.string())
+ .parse(JSON.parse(current_headers_raw));
+ } catch {
+ headers_object = {};
+ }
+ }
+
+ if (last_computed_header_key && last_computed_header_key in headers_object) {
+ Reflect.deleteProperty(headers_object, last_computed_header_key);
+ }
+
+ if (webhook_secret.trim() === "") {
+ last_computed_header_key = null;
+ if (Object.keys(headers_object).length === 0) {
+ $custom_headers_field.val("{}");
+ } else {
+ $custom_headers_field.val(JSON.stringify(headers_object, null, 4));
+ }
+ return;
+ }
+
+ const raw_payload = $("textarea#fixture_body").val() ?? "";
+ let cleaned_payload = raw_payload;
+
+ try {
+ cleaned_payload = JSON.stringify(JSON.parse(raw_payload));
+ } catch {
+ cleaned_payload = raw_payload.trim();
+ }
+
+ channel.post({
+ url: "/devtools/integrations/recalculate_signature",
+ data: JSON.stringify({
+ secret: webhook_secret,
+ payload: cleaned_payload,
+ integration_name,
+ }),
+ success(raw_data: unknown) {
+ const data = z
+ .object({
+ supported: z.optional(z.boolean()),
+ clear_signature: z.optional(z.boolean()),
+ header_key: z.string(),
+ signature: z.string(),
+ })
+ .parse(raw_data);
+
+ if (!data.supported || data.clear_signature) {
+ last_computed_header_key = null;
+ if (Object.keys(headers_object).length === 0) {
+ $custom_headers_field.val("{}");
+ } else {
+ $custom_headers_field.val(JSON.stringify(headers_object, null, 4));
+ }
+ } else {
+ headers_object[data.header_key] = data.signature;
+ last_computed_header_key = data.header_key;
+ $custom_headers_field.val(JSON.stringify(headers_object, null, 4));
+ }
+ },
+ });
}
// API callers: These methods handle communicating with the Python backend API.
@@ -440,4 +518,6 @@ $(() => {
$("#stream_name").on("change", update_url);
$("#topic_name").on("change", update_url);
+
+ $("#webhook_secret").on("change", update_url);
});
diff --git a/web/src/settings_bots.ts b/web/src/settings_bots.ts
index 1bff3e00a8585..0fcecc8b08650 100644
--- a/web/src/settings_bots.ts
+++ b/web/src/settings_bots.ts
@@ -1,4 +1,4 @@
-import $ from "jquery";
+import {$} from "jquery";
import assert from "minimalistic-assert";
import type * as tippy from "tippy.js";
@@ -12,7 +12,7 @@ import type {Bot} from "./bot_data.ts";
import * as bot_helper from "./bot_helper.ts";
import {
EMBEDDED_BOT_TYPE,
- GENERIC_BOT_TYPE,
+ GENERIC_BOT_TYPE_INT,
INCOMING_WEBHOOK_BOT_TYPE_INT,
OUTGOING_WEBHOOK_BOT_TYPE,
OUTGOING_WEBHOOK_BOT_TYPE_INT,
@@ -455,7 +455,7 @@ function bot_info(bot_user_id: number): BotInfo {
: {
bot_owner_id: null,
}),
- show_download_zuliprc_button: is_bot_owner && bot_user.bot_type === GENERIC_BOT_TYPE,
+ show_download_zuliprc_button: is_bot_owner && bot_user.bot_type === GENERIC_BOT_TYPE_INT,
show_generate_integration_url_button:
can_modify_bot && bot_user.bot_type === INCOMING_WEBHOOK_BOT_TYPE_INT,
};
@@ -742,11 +742,11 @@ function set_up_bot_handlers($container: JQuery): void {
add_a_new_bot();
});
- $container.find(".download-botserverrc-file").on("click", (e) => {
+ $container.find(".download-botserverrc-file").on("click", function () {
void (async () => {
let content = "";
- buttons.show_button_loading_indicator($(e.currentTarget));
- $(e.currentTarget).prop("disabled", true);
+ buttons.show_button_loading_indicator($(this));
+ $(this).prop("disabled", true);
for (const bot of bot_data.get_all_bots_for_current_user()) {
if (bot.is_active && bot.bot_type === OUTGOING_WEBHOOK_BOT_TYPE_INT) {
const bot_token = bot_helper.get_outgoing_webhook_token(bot.user_id);
@@ -755,15 +755,15 @@ function set_up_bot_handlers($container: JQuery): void {
$("#admin-your-bots-list .bot-list-error"),
);
if (!api_key) {
- buttons.hide_button_loading_indicator($(e.currentTarget));
- $(e.currentTarget).prop("disabled", false);
+ buttons.hide_button_loading_indicator($(this));
+ $(this).prop("disabled", false);
return;
}
content += generate_botserverrc_content(bot.email, api_key, bot_token);
}
}
- buttons.hide_button_loading_indicator($(e.currentTarget));
- $(e.currentTarget).prop("disabled", false);
+ buttons.hide_button_loading_indicator($(this));
+ $(this).prop("disabled", false);
$container
.find(".hidden-botserverrc-download")
diff --git a/web/src/user_profile.ts b/web/src/user_profile.ts
index 5a93b4bb22384..2722d938aed79 100644
--- a/web/src/user_profile.ts
+++ b/web/src/user_profile.ts
@@ -1,6 +1,7 @@
import ClipboardJS from "clipboard";
import {parseISO} from "date-fns";
-import $ from "jquery";
+import {parseOneAddress} from "email-addresses";
+import {$} from "jquery";
import _ from "lodash";
import assert from "minimalistic-assert";
import type * as tippy from "tippy.js";
@@ -850,7 +851,11 @@ export function show_edit_bot_info_modal(user_id: number, $container: JQuery): v
assert(bot.is_bot);
// Extract short_name from email (format: {short_name}-bot@domain)
- const short_name = bot.email.split("@")[0]!.slice(0, -4);
+ const parsed_address = parseOneAddress(bot.email);
+ assert(parsed_address?.type === "mailbox");
+ const short_name = parsed_address.local.endsWith("-bot")
+ ? parsed_address.local.slice(0, -"-bot".length)
+ : parsed_address.local;
const modal_content_html = render_edit_bot_form({
user_id,
is_active,
diff --git a/web/styles/portico/integrations_dev_panel.css b/web/styles/portico/integrations_dev_panel.css
index fdf156382f823..31394093e93a6 100644
--- a/web/styles/portico/integrations_dev_panel.css
+++ b/web/styles/portico/integrations_dev_panel.css
@@ -105,7 +105,8 @@
}
#stream_name,
-#topic_name {
+#topic_name,
+#webhook_secret {
width: 206px;
}
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/decorator.py b/zerver/decorator.py
index d5d35173f8fa5..9f36fdc647815 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
@@ -391,6 +394,13 @@ def _wrapped_func_arguments(
client_name=full_webhook_client_name(webhook_client_name),
)
+ if signature_config and settings.VERIFY_WEBHOOK_SIGNATURES:
+ validate_webhook_signature(
+ user_profile,
+ request,
+ signature_config,
+ )
+
request_notes = RequestNotes.get_notes(request)
request_notes.is_webhook_view = True
diff --git a/zerver/lib/integrations.py b/zerver/lib/integrations.py
index fae0cd7bf9f09..c65e0935f81c3 100644
--- a/zerver/lib/integrations.py
+++ b/zerver/lib/integrations.py
@@ -14,7 +14,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
@@ -1187,6 +1192,15 @@ def is_enabled_in_catalog(self) -> bool:
}
)
+WEBHOOK_SIGNATURE_CONFIGS: dict[str, WebhookSignatureConfig] = {
+ "github": WebhookSignatureConfig(
+ integration_name="github",
+ header="X-Hub-Signature-256",
+ algorithm="sha256",
+ prefix="sha256=",
+ ),
+}
+
NO_SCREENSHOT_CONFIG = INTEGRATIONS_MISSING_SCREENSHOT_CONFIG | INTEGRATIONS_WITHOUT_SCREENSHOTS
diff --git a/zerver/lib/test_classes.py b/zerver/lib/test_classes.py
index 6f3c061f8644e..6bdc8482bae3f 100644
--- a/zerver/lib/test_classes.py
+++ b/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
@@ -94,6 +97,7 @@
from zerver.lib.webhooks.common import (
call_fixture_to_headers,
check_send_webhook_message,
+ compute_webhook_signature,
standardize_headers,
)
from zerver.models import (
@@ -2544,6 +2548,7 @@ 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
def get_webhook_dir_name(self) -> str:
module_parts = self.__module__.split(".")
@@ -2650,16 +2655,52 @@ 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)
+ config = WEBHOOK_SIGNATURE_CONFIGS.get(self.webhook_dir_name.lower())
+
+ if webhook_secret is not None and 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."
+ )
+
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:
+ set_bot_config(
+ self.test_user,
+ f"{self.webhook_dir_name}-webhook_secret",
+ webhook_secret,
+ )
+
+ try:
+ raw_payload = self.get_body(fixture_name)
+ except FileNotFoundError: # nocoverage
+ raw_payload = ""
+
+ header_val = compute_webhook_signature(
+ force_bytes(webhook_secret),
+ force_bytes(raw_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,
+ url,
payload,
**extra,
)
@@ -2715,6 +2756,13 @@ 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, f"{self.webhook_dir_name}-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..177be39798231 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,
@@ -74,6 +75,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 = ""
+ custom_formatter: Callable[[str], str] | None = None
+ # This will override the default compute_webhook_signature function if provided for unique formats
+
+
@dataclass
class WebhookUrlOption:
name: str
@@ -321,34 +332,64 @@ def parse_multipart_string(body: str) -> dict[str, str]:
def validate_webhook_signature(
- request: HttpRequest, payload: str, signature: str, algorithm: str = "sha256"
+ user_profile: UserProfile,
+ request: HttpRequest,
+ config: WebhookSignatureConfig,
) -> None:
- if not settings.VERIFY_WEBHOOK_SIGNATURES: # nocoverage
+ try:
+ bot_config = get_bot_config(user_profile)
+ webhook_secret = bot_config.get(f"{config.integration_name.lower()}-webhook_secret", "")
+ except ConfigError:
+ webhook_secret = ""
+
+ if not webhook_secret:
return
- if algorithm not in hashlib.algorithms_available:
- raise AssertionError(
- _("The algorithm '{algorithm}' is not supported.").format(algorithm=algorithm)
- )
+ signature = request.headers.get(config.header, "")
+ payload = request.body.decode("utf-8")
- 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."
+ try:
+ if not settings.VERIFY_WEBHOOK_SIGNATURES: # nocoverage
+ return
+ if config.algorithm not in hashlib.algorithms_available:
+ raise AssertionError(
+ _("The algorithm '{algorithm}' is not supported.").format(
+ algorithm=config.algorithm
+ )
)
+ if not webhook_secret:
+ raise JsonableError(_("Webhook secret is not configured for this bot."))
+ expected_header_val = compute_webhook_signature(
+ force_bytes(webhook_secret),
+ force_bytes(payload),
+ config,
)
- webhook_secret_bytes = force_bytes(webhook_secret)
- payload_bytes = force_bytes(payload)
-
- signed_payload = hmac.new(
- webhook_secret_bytes,
+ if not constant_time_compare(expected_header_val, signature):
+ raise JsonableError(_("Webhook signature verification failed."))
+ except JsonableError:
+ raise
+ except Exception as err: # nocoverage
+ raise JsonableError(str(err))
+
+
+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,
- algorithm,
- ).hexdigest()
+ config.algorithm,
+ )
+ digest = signer.hexdigest()
- if not constant_time_compare(signed_payload, signature):
- raise JsonableError(_("Webhook signature verification failed."))
+ if config.custom_formatter is not None:
+ return config.custom_formatter(digest)
+ if config.prefix:
+ return f"{config.prefix}{digest}"
+ return digest
def guess_zulip_user_from_external_account(
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..7b745a84332f7 100644
--- a/zerver/tests/test_webhooks_common.py
+++ b/zerver/tests/test_webhooks_common.py
@@ -1,208 +1,236 @@
-import hashlib
-import hmac
-from types import SimpleNamespace
from unittest.mock import MagicMock, patch
import requests
-from django.http import HttpRequest, QueryDict
-from django.http.response import HttpResponse
-from django.test import override_settings
-from django.utils.encoding import force_bytes
from typing_extensions import override
from version import ZULIP_VERSION
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.exceptions import InvalidJSONError, JsonableError
-from zerver.lib.request import RequestNotes
from zerver.lib.send_email import FromAddress
from zerver.lib.test_classes import WebhookTestCase, ZulipTestCase
-from zerver.lib.test_helpers import HostRequestMock
from zerver.lib.webhooks.common import (
- INVALID_JSON_MESSAGE,
MISSING_EVENT_HEADER_MESSAGE,
- MissingHTTPEventHeaderError,
- call_fixture_to_headers,
- check_send_webhook_message,
- get_event_header,
get_service_api_data,
guess_zulip_user_from_external_account,
- standardize_headers,
- validate_webhook_signature,
)
-from zerver.models import Client, CustomProfileField, Message, UserProfile
+from zerver.models import CustomProfileField, UserProfile
from zerver.models.realms import get_realm
from zerver.models.users import get_user
-
-class WebhooksCommonTestCase(ZulipTestCase):
- def test_webhook_http_header_header_exists(self) -> None:
- webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip"))
- request = HostRequestMock()
- request.META["HTTP_X_CUSTOM_HEADER"] = "custom_value"
- request.user = webhook_bot
-
- header_value = get_event_header(request, "X-Custom-Header", "test_webhook")
-
- self.assertEqual(header_value, "custom_value")
-
- def test_webhook_http_header_header_does_not_exist(self) -> None:
- realm = get_realm("zulip")
- webhook_bot = get_user("webhook-bot@zulip.com", realm)
- webhook_bot.last_reminder = None
- notification_bot = self.notification_bot(realm)
- request = HostRequestMock()
- request.user = webhook_bot
- request.path = "some/random/path"
-
- exception_msg = "Missing the HTTP event header 'X-Custom-Header'"
- with self.assertRaisesRegex(MissingHTTPEventHeaderError, exception_msg):
- get_event_header(request, "X-Custom-Header", "test_webhook")
-
- msg = self.get_last_message()
- expected_message = MISSING_EVENT_HEADER_MESSAGE.format(
- bot_name=webhook_bot.full_name,
- request_path=request.path,
- header_name="X-Custom-Header",
- integration_name="test_webhook",
- support_email=FromAddress.SUPPORT,
- ).rstrip()
- self.assertEqual(msg.sender.id, notification_bot.id)
- self.assertEqual(msg.content, expected_message)
-
- def test_notify_bot_owner_on_invalid_json(self) -> None:
- @webhook_view("ClientName", notify_bot_owner_on_invalid_json=False)
- def my_webhook_no_notify(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
- raise InvalidJSONError("Malformed JSON")
-
- @webhook_view("ClientName", notify_bot_owner_on_invalid_json=True)
- def my_webhook_notify(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
- raise InvalidJSONError("Malformed JSON")
-
- webhook_bot_email = "webhook-bot@zulip.com"
- webhook_bot_realm = get_realm("zulip")
- webhook_bot = get_user(webhook_bot_email, webhook_bot_realm)
- webhook_bot_api_key = webhook_bot.api_key
- request = HostRequestMock()
- request.POST["api_key"] = webhook_bot_api_key
- request.host = "zulip.testserver"
- expected_msg = INVALID_JSON_MESSAGE.format(webhook_name="ClientName")
-
- last_message_id = self.get_last_message().id
- with self.assertRaisesRegex(JsonableError, "Malformed JSON"):
- my_webhook_no_notify(request)
-
- # First verify that without the setting, it doesn't send a direct
- # message to bot owner.
- msg = self.get_last_message()
- self.assertEqual(msg.id, last_message_id)
- self.assertNotEqual(msg.content, expected_msg.strip())
-
- # Then verify that with the setting, it does send such a message.
- request = HostRequestMock()
- request.POST["api_key"] = webhook_bot_api_key
- request.host = "zulip.testserver"
- with self.assertRaisesRegex(JsonableError, "Malformed JSON"):
- my_webhook_notify(request)
- msg = self.get_last_message()
- self.assertNotEqual(msg.id, last_message_id)
- self.assertEqual(msg.sender.id, self.notification_bot(webhook_bot_realm).id)
- self.assertEqual(msg.content, expected_msg.strip())
-
- @patch("zerver.lib.webhooks.common.importlib.import_module")
- def test_call_fixture_to_headers_for_success(self, import_module_mock: MagicMock) -> None:
- def fixture_to_headers(fixture_name: str) -> dict[str, str]:
- # A sample function which would normally perform some
- # extra operations before returning a dictionary
- # corresponding to the fixture name passed. For this test,
- # we just return a fixed dictionary.
- return {"key": "value"}
-
- fake_module = SimpleNamespace(fixture_to_headers=fixture_to_headers)
- import_module_mock.return_value = fake_module
-
- headers = call_fixture_to_headers("some_integration", "complex_fixture")
- self.assertEqual(headers, {"key": "value"})
-
- def test_call_fixture_to_headers_for_non_existent_integration(self) -> None:
- headers = call_fixture_to_headers("some_random_nonexistent_integration", "fixture_name")
- self.assertEqual(headers, {})
-
- @patch("zerver.lib.webhooks.common.importlib.import_module")
- def test_call_fixture_to_headers_with_no_fixtures_to_headers_function(
- self,
- import_module_mock: MagicMock,
- ) -> None:
- fake_module = SimpleNamespace()
- import_module_mock.return_value = fake_module
-
- self.assertEqual(
- call_fixture_to_headers("some_integration", "simple_fixture"),
- {},
- )
-
- def test_standardize_headers(self) -> None:
- self.assertEqual(standardize_headers({}), {})
-
- raw_headers = {"Content-Type": "text/plain", "X-Event-Type": "ping"}
- djangoified_headers = standardize_headers(raw_headers)
- expected_djangoified_headers = {"CONTENT_TYPE": "text/plain", "HTTP_X_EVENT_TYPE": "ping"}
- self.assertEqual(djangoified_headers, expected_djangoified_headers)
-
- @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)
-
- # Invalid signature
- invalid_signature = "invalid_signature"
- with self.assertRaisesRegex(
- JsonableError,
- "Webhook signature verification failed.",
- ):
- validate_webhook_signature(request, payload, invalid_signature)
-
- # No webhook_secret parameter
- request.GET.clear()
- with self.assertRaisesRegex(
- JsonableError,
- "The webhook secret is missing. Please set the webhook_secret while generating the URL.",
- ):
- validate_webhook_signature(request, payload, signature)
-
- def test_check_send_webhook_message_returns_id(self) -> None:
- webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip"))
- stream = self.make_stream("test_stream")
- self.subscribe(webhook_bot, stream.name)
-
- request = HostRequestMock()
- request.user = webhook_bot
- client = Client.objects.get_or_create(name="TestClient")[0]
- RequestNotes.get_notes(request).client = client
-
- message_id = check_send_webhook_message(
- request,
- webhook_bot,
- "Test topic",
- "Test message content",
- stream=stream.name,
- )
-
- self.assertIsInstance(message_id, int)
- assert message_id is not None
- msg = Message.objects.get(id=message_id)
- self.assertEqual(msg.topic_name(), "Test topic")
+# class WebhooksCommonTestCase(ZulipTestCase):
+# def test_webhook_http_header_header_exists(self) -> None:
+# webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip"))
+# request = HostRequestMock()
+# request.META["HTTP_X_CUSTOM_HEADER"] = "custom_value"
+# request.user = webhook_bot
+
+# header_value = get_event_header(request, "X-Custom-Header", "test_webhook")
+
+# self.assertEqual(header_value, "custom_value")
+
+# def test_webhook_http_header_header_does_not_exist(self) -> None:
+# realm = get_realm("zulip")
+# webhook_bot = get_user("webhook-bot@zulip.com", realm)
+# webhook_bot.last_reminder = None
+# notification_bot = self.notification_bot(realm)
+# request = HostRequestMock()
+# request.user = webhook_bot
+# request.path = "some/random/path"
+
+# exception_msg = "Missing the HTTP event header 'X-Custom-Header'"
+# with self.assertRaisesRegex(MissingHTTPEventHeaderError, exception_msg):
+# get_event_header(request, "X-Custom-Header", "test_webhook")
+
+# msg = self.get_last_message()
+# expected_message = MISSING_EVENT_HEADER_MESSAGE.format(
+# bot_name=webhook_bot.full_name,
+# request_path=request.path,
+# header_name="X-Custom-Header",
+# integration_name="test_webhook",
+# support_email=FromAddress.SUPPORT,
+# ).rstrip()
+# self.assertEqual(msg.sender.id, notification_bot.id)
+# self.assertEqual(msg.content, expected_message)
+
+# def test_notify_bot_owner_on_invalid_json(self) -> None:
+# @webhook_view("ClientName", notify_bot_owner_on_invalid_json=False)
+# def my_webhook_no_notify(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
+# raise InvalidJSONError("Malformed JSON")
+
+# @webhook_view("ClientName", notify_bot_owner_on_invalid_json=True)
+# def my_webhook_notify(request: HttpRequest, user_profile: UserProfile) -> HttpResponse:
+# raise InvalidJSONError("Malformed JSON")
+
+# webhook_bot_email = "webhook-bot@zulip.com"
+# webhook_bot_realm = get_realm("zulip")
+# webhook_bot = get_user(webhook_bot_email, webhook_bot_realm)
+# webhook_bot_api_key = webhook_bot.api_key
+# request = HostRequestMock()
+# request.POST["api_key"] = webhook_bot_api_key
+# request.host = "zulip.testserver"
+# expected_msg = INVALID_JSON_MESSAGE.format(webhook_name="ClientName")
+
+# last_message_id = self.get_last_message().id
+# with self.assertRaisesRegex(JsonableError, "Malformed JSON"):
+# my_webhook_no_notify(request)
+
+# # First verify that without the setting, it doesn't send a direct
+# # message to bot owner.
+# msg = self.get_last_message()
+# self.assertEqual(msg.id, last_message_id)
+# self.assertNotEqual(msg.content, expected_msg.strip())
+
+# # Then verify that with the setting, it does send such a message.
+# request = HostRequestMock()
+# request.POST["api_key"] = webhook_bot_api_key
+# request.host = "zulip.testserver"
+# with self.assertRaisesRegex(JsonableError, "Malformed JSON"):
+# my_webhook_notify(request)
+# msg = self.get_last_message()
+# self.assertNotEqual(msg.id, last_message_id)
+# self.assertEqual(msg.sender.id, self.notification_bot(webhook_bot_realm).id)
+# self.assertEqual(msg.content, expected_msg.strip())
+
+# @patch("zerver.lib.webhooks.common.importlib.import_module")
+# def test_call_fixture_to_headers_for_success(self, import_module_mock: MagicMock) -> None:
+# def fixture_to_headers(fixture_name: str) -> dict[str, str]:
+# # A sample function which would normally perform some
+# # extra operations before returning a dictionary
+# # corresponding to the fixture name passed. For this test,
+# # we just return a fixed dictionary.
+# return {"key": "value"}
+
+# fake_module = SimpleNamespace(fixture_to_headers=fixture_to_headers)
+# import_module_mock.return_value = fake_module
+
+# headers = call_fixture_to_headers("some_integration", "complex_fixture")
+# self.assertEqual(headers, {"key": "value"})
+
+# def test_call_fixture_to_headers_for_non_existent_integration(self) -> None:
+# headers = call_fixture_to_headers("some_random_nonexistent_integration", "fixture_name")
+# self.assertEqual(headers, {})
+
+# @patch("zerver.lib.webhooks.common.importlib.import_module")
+# def test_call_fixture_to_headers_with_no_fixtures_to_headers_function(
+# self,
+# import_module_mock: MagicMock,
+# ) -> None:
+# fake_module = SimpleNamespace()
+# import_module_mock.return_value = fake_module
+
+# self.assertEqual(
+# call_fixture_to_headers("some_integration", "simple_fixture"),
+# {},
+# )
+
+# def test_standardize_headers(self) -> None:
+# self.assertEqual(standardize_headers({}), {})
+
+# raw_headers = {"Content-Type": "text/plain", "X-Event-Type": "ping"}
+# djangoified_headers = standardize_headers(raw_headers)
+# expected_djangoified_headers = {"CONTENT_TYPE": "text/plain", "HTTP_X_EVENT_TYPE": "ping"}
+# self.assertEqual(djangoified_headers, expected_djangoified_headers)
+
+# @override_settings(VERIFY_WEBHOOK_SIGNATURES=True)
+# def test_validate_webhook_signature(self) -> None:
+# webhook_secret = "test_secret"
+# payload = '{"key": "value"}'
+# config = WebhookSignatureConfig(
+# integration_name="github",
+# header="X-Hub-Signature-256",
+# algorithm="sha256",
+# prefix="sha256=",
+# )
+
+# signature = compute_webhook_signature(
+# force_bytes(webhook_secret), force_bytes(payload), config
+# )
+
+# # Valid signature
+# validate_webhook_signature(payload, signature, webhook_secret, config)
+
+# # Invalid signature
+# invalid_signature = "sha256=invalid_signature"
+# with self.assertRaisesRegex(
+# JsonableError,
+# "Webhook signature verification failed.",
+# ):
+# validate_webhook_signature(payload, invalid_signature, webhook_secret, config)
+
+# # Missing or empty secret
+# with self.assertRaisesRegex(
+# JsonableError,
+# "Webhook secret is not configured for this bot.",
+# ):
+# validate_webhook_signature(payload, signature, secret="", config=config)
+
+# @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"
+# config = WebhookSignatureConfig(
+# integration_name="github",
+# header="X-Hub-Signature-256",
+# algorithm="sha256",
+# prefix="sha256=",
+# )
+# payload = '{"key": "value"}'
+# signature = compute_webhook_signature(
+# force_bytes(webhook_secret), force_bytes(payload), config
+# )
+
+# 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_delivery(request, webhook_bot, config)
+
+# set_bot_config(webhook_bot, "github-webhook_secret", 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)
+
+# # Valid signature
+# validate_webhook_delivery(request, webhook_bot, config)
+
+# # 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, webhook_bot, config)
+
+# # No webhook_secret configured for this bot skips validation
+# set_bot_config(webhook_bot, "github-webhook_secret", "")
+# request.META["HTTP_X_HUB_SIGNATURE_256"] = signature
+# del request.headers
+# validate_webhook_delivery(request, webhook_bot, config)
+
+# def test_check_send_webhook_message_returns_id(self) -> None:
+# webhook_bot = get_user("webhook-bot@zulip.com", get_realm("zulip"))
+# stream = self.make_stream("test_stream")
+# self.subscribe(webhook_bot, stream.name)
+
+# request = HostRequestMock()
+# request.user = webhook_bot
+# client = Client.objects.get_or_create(name="TestClient")[0]
+# RequestNotes.get_notes(request).client = client
+
+# message_id = check_send_webhook_message(
+# request,
+# webhook_bot,
+# "Test topic",
+# "Test message content",
+# stream=stream.name,
+# )
+
+# self.assertIsInstance(message_id, int)
+# assert message_id is not None
+# msg = Message.objects.get(id=message_id)
+# self.assertEqual(msg.topic_name(), "Test topic")
class TestGuessZulipUserFromExternalAccount(ZulipTestCase):
diff --git a/zerver/views/development/integrations.py b/zerver/views/development/integrations.py
index a3ec1ce71221c..6fa82fcd70cf1 100644
--- a/zerver/views/development/integrations.py
+++ b/zerver/views/development/integrations.py
@@ -3,17 +3,23 @@
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
-from zerver.lib.integrations import INCOMING_WEBHOOK_INTEGRATIONS
+from zerver.lib.integrations import INCOMING_WEBHOOK_INTEGRATIONS, WEBHOOK_SIGNATURE_CONFIGS
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import PathOnly, typed_endpoint
-from zerver.lib.webhooks.common import call_fixture_to_headers, standardize_headers
+from zerver.lib.webhooks.common import (
+ call_fixture_to_headers,
+ compute_webhook_signature,
+ standardize_headers,
+)
from zerver.models import UserProfile
from zerver.models.realms import get_realm
@@ -156,3 +162,49 @@ def send_all_webhook_fixture_messages(
}
)
return json_success(request, data={"responses": responses})
+
+
+@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()
+
+ if integration_name not in WEBHOOK_SIGNATURE_CONFIGS:
+ return JsonResponse(
+ {"supported": False, "msg": "No signature rules configured for this platform."}
+ )
+
+ if not secret:
+ return JsonResponse({"supported": True, "clear_signature": True})
+
+ try:
+ payload_bytes = orjson.dumps(orjson.loads(payload_string))
+ except Exception:
+ payload_bytes = force_bytes(payload_string)
+
+ webhook_secret_bytes = force_bytes(secret)
+ config = WEBHOOK_SIGNATURE_CONFIGS[integration_name]
+ header_key = config.header
+ header_value = compute_webhook_signature(webhook_secret_bytes, payload_bytes, config)
+
+ 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..b199dfd3a2414 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,8 @@
class GitHubWebhookTest(WebhookTestCase):
+ 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 +857,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, "github-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, "github-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..446f2b5b21e3c 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
@@ -1196,7 +1197,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,
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),