diff --git a/zerver/webhooks/gosquared/view.py b/zerver/webhooks/gosquared/view.py index 839adecc56..e13932cee4 100644 --- a/zerver/webhooks/gosquared/view.py +++ b/zerver/webhooks/gosquared/view.py @@ -1,11 +1,10 @@ -from typing import Any, Dict - from django.http import HttpRequest, HttpResponse from zerver.decorator import webhook_view from zerver.lib.exceptions import UnsupportedWebhookEventType from zerver.lib.request import REQ, has_request_variables from zerver.lib.response import json_success +from zerver.lib.validator import WildValue, check_bool, check_int, check_string, to_wild_value from zerver.lib.webhooks.common import check_send_webhook_message from zerver.models import UserProfile @@ -27,7 +26,7 @@ ALL_EVENT_TYPES = ["chat_message", "traffic_spike"] def api_gosquared_webhook( request: HttpRequest, user_profile: UserProfile, - payload: Dict[str, Dict[str, Any]] = REQ(argument_type="body"), + payload: WildValue = REQ(argument_type="body", converter=to_wild_value), ) -> HttpResponse: body = "" topic = "" @@ -35,10 +34,10 @@ def api_gosquared_webhook( # Unfortunately, there is no other way to infer the event type # than just inferring it from the payload's attributes # Traffic spike/dip event - if payload.get("concurrents") is not None and payload.get("siteDetails") is not None: - domain_name = payload["siteDetails"]["domain"] - user_num = payload["concurrents"] - user_acc = payload["siteDetails"]["acct"] + if "concurrents" in payload and "siteDetails" in payload: + domain_name = payload["siteDetails"]["domain"].tame(check_string) + user_num = payload["concurrents"].tame(check_int) + user_acc = payload["siteDetails"]["acct"].tame(check_string) acc_url = "https://www.gosquared.com/now/" + user_acc body = TRAFFIC_SPIKE_TEMPLATE.format( website_name=domain_name, website_url=acc_url, user_num=user_num @@ -49,13 +48,13 @@ def api_gosquared_webhook( # Live chat message event elif payload.get("message") is not None and payload.get("person") is not None: # Only support non-private messages - if not payload["message"]["private"]: - session_title = payload["message"]["session"]["title"] + if not payload["message"]["private"].tame(check_bool): + session_title = payload["message"]["session"]["title"].tame(check_string) topic = f"Live chat session - {session_title}" body = CHAT_MESSAGE_TEMPLATE.format( - status=payload["person"]["status"], - name=payload["person"]["_anon"]["name"], - content=payload["message"]["content"], + status=payload["person"]["status"].tame(check_string), + name=payload["person"]["_anon"]["name"].tame(check_string), + content=payload["message"]["content"].tame(check_string), ) check_send_webhook_message(request, user_profile, topic, body, "chat_message") else: