diff --git a/zerver/webhooks/alertmanager/view.py b/zerver/webhooks/alertmanager/view.py index d8a2de8dc6..2c3f14506c 100644 --- a/zerver/webhooks/alertmanager/view.py +++ b/zerver/webhooks/alertmanager/view.py @@ -1,11 +1,12 @@ # Webhooks for external integrations. -from typing import Any, Dict, List +from typing import Dict, List from django.http import HttpRequest, HttpResponse from zerver.decorator import webhook_view from zerver.lib.request import REQ, has_request_variables from zerver.lib.response import json_success +from zerver.lib.validator import WildValue, check_string, to_wild_value from zerver.lib.webhooks.common import check_send_webhook_message from zerver.models import UserProfile @@ -15,7 +16,7 @@ from zerver.models import UserProfile def api_alertmanager_webhook( request: HttpRequest, user_profile: UserProfile, - payload: Dict[str, Any] = REQ(argument_type="body"), + payload: WildValue = REQ(argument_type="body", converter=to_wild_value), name_field: str = REQ("name", default="instance"), desc_field: str = REQ("desc", default="alertname"), ) -> HttpResponse: @@ -25,15 +26,17 @@ def api_alertmanager_webhook( labels = alert.get("labels", {}) annotations = alert.get("annotations", {}) - name = labels.get(name_field, annotations.get(name_field, "(unknown)")) - desc = labels.get(desc_field, annotations.get(desc_field, f"")) + name = labels.get(name_field, annotations.get(name_field, "(unknown)")).tame(check_string) + desc = labels.get( + desc_field, annotations.get(desc_field, f"") + ).tame(check_string) - url = alert.get("generatorURL").replace("tab=1", "tab=0") + url = alert["generatorURL"].tame(check_string).replace("tab=1", "tab=0") body = f"{desc} ([graph]({url}))" if name not in topics: topics[name] = {"firing": [], "resolved": []} - topics[name][alert["status"]].append(body) + topics[name][alert["status"].tame(check_string)].append(body) for topic, statuses in topics.items(): for status, messages in statuses.items():