diff --git a/zerver/webhooks/uptimerobot/view.py b/zerver/webhooks/uptimerobot/view.py index 7889d96c5c..6053832fc5 100644 --- a/zerver/webhooks/uptimerobot/view.py +++ b/zerver/webhooks/uptimerobot/view.py @@ -1,5 +1,4 @@ -from typing import Any, Dict - +from django.core.exceptions import ValidationError from django.http import HttpRequest, HttpResponse from django.utils.translation import gettext as _ @@ -9,6 +8,7 @@ from zerver.lib.exceptions import JsonableError from zerver.lib.request import REQ, has_request_variables from zerver.lib.response import json_success from zerver.lib.send_email import FromAddress +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 @@ -36,17 +36,18 @@ ALL_EVENT_TYPES = ["up", "down"] def api_uptimerobot_webhook( request: HttpRequest, user_profile: UserProfile, - payload: Dict[str, Any] = REQ(argument_type="body"), + payload: WildValue = REQ(argument_type="body", converter=to_wild_value), ) -> HttpResponse: - if payload["alert_type_friendly_name"] == "Up": + event_type = payload["alert_type_friendly_name"].tame(check_string) + if event_type == "Up": event = "up" - elif payload["alert_type_friendly_name"] == "Down": + elif event_type == "Down": event = "down" try: - body = get_body_for_http_request(payload) + body = get_body_for_http_request(payload, event_type) subject = get_subject_for_http_request(payload) - except KeyError: + except ValidationError: message = MISCONFIGURED_PAYLOAD_ERROR_MESSAGE.format( bot_name=user_profile.full_name, support_email=FromAddress.SUPPORT, @@ -59,14 +60,32 @@ def api_uptimerobot_webhook( return json_success(request) -def get_subject_for_http_request(payload: Dict[str, Any]) -> str: - return UPTIMEROBOT_TOPIC_TEMPLATE.format(monitor_friendly_name=payload["monitor_friendly_name"]) +def get_subject_for_http_request(payload: WildValue) -> str: + return UPTIMEROBOT_TOPIC_TEMPLATE.format( + monitor_friendly_name=payload["monitor_friendly_name"].tame(check_string) + ) -def get_body_for_http_request(payload: Dict[str, Any]) -> str: - if payload["alert_type_friendly_name"] == "Up": - body = UPTIMEROBOT_MESSAGE_UP_TEMPLATE.format(**payload) - elif payload["alert_type_friendly_name"] == "Down": - body = UPTIMEROBOT_MESSAGE_DOWN_TEMPLATE.format(**payload) +def get_body_for_http_request(payload: WildValue, event_type: str) -> str: + if event_type == "Up": + monitor_friendly_name = payload["monitor_friendly_name"].tame(check_string) + monitor_url = payload["monitor_url"].tame(check_string) + alert_details = payload["alert_details"].tame(check_string) + alert_friendly_duration = payload["alert_friendly_duration"].tame(check_string) + body = UPTIMEROBOT_MESSAGE_UP_TEMPLATE.format( + monitor_friendly_name=monitor_friendly_name, + monitor_url=monitor_url, + alert_details=alert_details, + alert_friendly_duration=alert_friendly_duration, + ) + elif event_type == "Down": + monitor_friendly_name = payload["monitor_friendly_name"].tame(check_string) + monitor_url = payload["monitor_url"].tame(check_string) + alert_details = payload["alert_details"].tame(check_string) + body = UPTIMEROBOT_MESSAGE_DOWN_TEMPLATE.format( + monitor_friendly_name=monitor_friendly_name, + monitor_url=monitor_url, + alert_details=alert_details, + ) return body