fix(notifications): use html template instead of dynamic html params where needed (#58658)

After releasing #58227, some
email notifications become broken. Example:
<img width="692" alt="image"
src="https://github.com/getsentry/sentry/assets/1127549/281171a9-74dc-4c2f-b724-f7bc8d7e52a9">

The cause is that `format_html` escaped template parameters that
contained HTML code in cases such as:
```
html_params["version"] = f'<a href="{version_url}">{escape(self.version_parsed)}</a>'
```
From security perspective, this was kinda correct behavior, since
parameter contains untrusted HTML. When building HTML, we should use
HTML-specific templates instead.

This PR:
* Updates the signature of the `get_description()` method in the base
GroupActivityNotification class and in all inherited classes:
  * old output: template, parameters, html_parameters (optional)
  * new output: text_template, html_template (optional), parameters
* The set of parameters would be the same for both modes -- text and
HTML
  * But we are going to redefine HTML template where needed.
* Redefines HTML templates for two cases:
* RegressionActivityNotification (fixes the above bug) -- we want to
render release link as `<a href>` tag
  * ResolvedInReleaseActivityNotification -- same
* The HTML template for most of cases is None and therefore normal text
template is used.

Author: oioki

src/sentry/notifications/notifications/activity/assigned.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Iterable, Mapping
+from typing import Any, Iterable, Mapping, Optional
 
 from sentry.models.activity import Activity
 from sentry.models.notificationsetting import NotificationSetting
@@ -68,8 +68,8 @@ class AssignedActivityNotification(GroupActivityNotification):
     def get_assignee(self) -> str:
         return get_assignee_str(self.activity, self.organization)
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
-        return "{author} assigned {an issue} to {assignee}", {"assignee": self.get_assignee()}, {}
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
+        return "{author} assigned {an issue} to {assignee}", None, {"assignee": self.get_assignee()}
 
     def get_notification_title(
         self, provider: ExternalProviders, context: Mapping[str, Any] | None = None

src/sentry/notifications/notifications/activity/base.py

@@ -2,7 +2,7 @@
 
 import abc
 from functools import cached_property
-from typing import TYPE_CHECKING, Any, Mapping, MutableMapping
+from typing import TYPE_CHECKING, Any, Mapping, MutableMapping, Optional
 from urllib.parse import urlparse, urlunparse
 
 from django.utils.html import format_html
@@ -81,7 +81,7 @@ def __init__(self, activity: Activity) -> None:
         super().__init__(activity)
         self.group = activity.group
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
         raise NotImplementedError
 
     def get_group_link(self) -> str:
@@ -121,11 +121,13 @@ def get_context(self) -> MutableMapping[str, Any]:
         expensive computation so it should only be called once. Override this
         method if the notification does not need HTML/text descriptions.
         """
-        description, params, html_params = self.get_description()
+        text_template, html_template, params = self.get_description()
+        text_description = self.description_as_text(text_template, params)
+        html_description = self.description_as_html(html_template or text_template, params)
         return {
             **self.get_base_context(),
-            "text_description": self.description_as_text(description, params),
-            "html_description": self.description_as_html(description, html_params or params),
+            "text_description": text_description,
+            "html_description": html_description,
         }
 
     def get_group_context(self) -> MutableMapping[str, Any]:
@@ -145,7 +147,7 @@ def get_group_context(self) -> MutableMapping[str, Any]:
     def get_notification_title(
         self, provider: ExternalProviders, context: Mapping[str, Any] | None = None
     ) -> str:
-        description, params, _ = self.get_description()
+        description, _, params = self.get_description()
         return self.description_as_text(description, params, True, provider)
 
     def get_subject(self, context: Mapping[str, Any] | None = None) -> str:

src/sentry/notifications/notifications/activity/escalating.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 
 from sentry.services.hybrid_cloud.actor import RpcActor
 from sentry.types.integrations import ExternalProviders
@@ -19,26 +19,26 @@ def get_notification_title(
 
         return self.title
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
         forecast = int(self.activity.data.get("forecast", 0))
         expired_snooze = self.activity.data.get("expired_snooze")
 
         if forecast:
             return (
                 "Sentry flagged this issue as escalating because over {forecast} {event} happened in an hour.",
+                None,
                 {"forecast": forecast, "event": "event" if forecast == 1 else "events"},
-                {},
             )
 
         if expired_snooze:
             return (
                 "Sentry flagged this issue as escalating because your archive condition has expired.",
-                {},
+                None,
                 {},
             )
 
         # Return a default basic message
-        return ("Sentry flagged this issue as escalating.", {}, {})
+        return ("Sentry flagged this issue as escalating.", None, {})
 
     def get_message_description(self, recipient: RpcActor, provider: ExternalProviders) -> Any:
         return self.get_context()["text_description"]

src/sentry/notifications/notifications/activity/note.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 
 from sentry.services.hybrid_cloud.actor import RpcActor
 from sentry.types.integrations import ExternalProviders
@@ -13,10 +13,10 @@ class NoteActivityNotification(GroupActivityNotification):
     metrics_key = "note_activity"
     template_path = "sentry/emails/activity/note"
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
         # Notes may contain {} characters so we should escape them.
         text = str(self.activity.data["text"]).replace("{", "{{").replace("}", "}}")
-        return text, {}, {}
+        return text, None, {}
 
     @property
     def title(self) -> str:

src/sentry/notifications/notifications/activity/regression.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
-from html import escape
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 from urllib.parse import urlencode
 
 from sentry_relay.processing import parse_release
@@ -21,8 +20,8 @@ def __init__(self, activity: Activity) -> None:
         self.version = self.activity.data.get("version", "")
         self.version_parsed = parse_release(self.version)["description"]
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
-        message, params, html_params = "{author} marked {an issue} as a regression", {}, {}
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
+        text_message, html_message, params = "{author} marked {an issue} as a regression", None, {}
 
         if self.version:
             version_url = self.organization.absolute_url(
@@ -32,11 +31,13 @@ def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
                 ),
             )
 
-            message += " in {version}"
+            html_message = text_message + ' in <a href="{url}">{version}</a>'
+            text_message += " in {version}"
+
+            params["url"] = version_url
             params["version"] = self.version_parsed
-            html_params["version"] = f'<a href="{version_url}">{escape(self.version_parsed)}</a>'
 
-        return message, params, html_params
+        return text_message, html_message, params
 
     def get_notification_title(
         self, provider: ExternalProviders, context: Mapping[str, Any] | None = None

src/sentry/notifications/notifications/activity/resolved.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 
 from .base import GroupActivityNotification
 
@@ -9,5 +9,5 @@ class ResolvedActivityNotification(GroupActivityNotification):
     metrics_key = "resolved_activity"
     title = "Resolved Issue"
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
-        return "{author} marked {an issue} as resolved", {}, {}
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
+        return "{author} marked {an issue} as resolved", None, {}

src/sentry/notifications/notifications/activity/resolved_in_release.py

@@ -1,7 +1,6 @@
 from __future__ import annotations
 
-from html import escape
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 from urllib.parse import urlencode
 
 from sentry_relay.processing import parse_release
@@ -21,7 +20,7 @@ def __init__(self, activity: Activity) -> None:
         self.version = self.activity.data.get("version", "")
         self.version_parsed = parse_release(self.version)["description"]
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
         if self.version:
             url = self.organization.absolute_url(
                 f"/organizations/{self.organization.slug}/releases/{self.version}/",
@@ -34,12 +33,17 @@ def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
                 ),
             )
 
+            params = {
+                "url": url,
+                "version": self.version_parsed,
+            }
+
             return (
                 "{author} marked {an issue} as resolved in {version}",
-                {"version": self.version_parsed},
-                {"version": f'<a href="{url}">{escape(self.version_parsed)}</a>'},
+                '{author} marked {an issue} as resolved in <a href="{url}">{version}</a>',
+                params,
             )
-        return "{author} marked {an issue} as resolved in an upcoming release", {}, {}
+        return "{author} marked {an issue} as resolved in an upcoming release", None, {}
 
     def get_notification_title(
         self, provider: ExternalProviders, context: Mapping[str, Any] | None = None

src/sentry/notifications/notifications/activity/unassigned.py

@@ -1,6 +1,6 @@
 from __future__ import annotations
 
-from typing import Any, Mapping
+from typing import Any, Mapping, Optional
 
 from sentry.types.integrations import ExternalProviders
 
@@ -11,8 +11,8 @@ class UnassignedActivityNotification(GroupActivityNotification):
     metrics_key = "unassigned_activity"
     title = "Unassigned"
 
-    def get_description(self) -> tuple[str, Mapping[str, Any], Mapping[str, Any]]:
-        return "{author} unassigned {an issue}", {}, {}
+    def get_description(self) -> tuple[str, Optional[str], Mapping[str, Any]]:
+        return "{author} unassigned {an issue}", None, {}
 
     def get_notification_title(
         self, provider: ExternalProviders, context: Mapping[str, Any] | None = None

tests/sentry/notifications/test_notifications.py

@@ -21,13 +21,15 @@
 from sentry.models.options.user_option import UserOption
 from sentry.models.rule import Rule
 from sentry.notifications.notifications.activity.assigned import AssignedActivityNotification
+from sentry.notifications.notifications.activity.regression import RegressionActivityNotification
 from sentry.silo import SiloMode
 from sentry.tasks.post_process import post_process_group
 from sentry.testutils.cases import APITestCase
 from sentry.testutils.helpers.datetime import before_now, iso_format
 from sentry.testutils.helpers.eventprocessing import write_event_to_cache
 from sentry.testutils.silo import assume_test_silo_mode, control_silo_test
 from sentry.testutils.skips import requires_snuba
+from sentry.types.activity import ActivityType
 from sentry.utils import json
 
 pytestmark = [requires_snuba]
@@ -197,6 +199,22 @@ def test_html_escape(self):
         assert "<b>test</b>" in html
         assert "<b>test</b>" not in html
 
+    @responses.activate
+    def test_regression_html_link(self):
+        notification = RegressionActivityNotification(
+            Activity(
+                project=self.project,
+                group=self.group,
+                user_id=self.user.id,
+                type=ActivityType.SET_REGRESSION,
+                data={"version": "777"},
+            )
+        )
+        context = notification.get_context()
+
+        assert "as a regression in 777" in context["text_description"]
+        assert "as a regression in <a href=" in context["html_description"]
+
     @responses.activate
     @patch("sentry.analytics.record")
     def test_sends_resolution_notification(self, record_analytics):