gosquared: Strengthen types using WildValue.

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:
+    if "concurrents" in payload and "siteDetails" in payload:
-        domain_name = payload["siteDetails"]["domain"]
+        domain_name = payload["siteDetails"]["domain"].tame(check_string)
-        user_num = payload["concurrents"]
+        user_num = payload["concurrents"].tame(check_int)
-        user_acc = payload["siteDetails"]["acct"]
+        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"]:
+        if not payload["message"]["private"].tame(check_bool):
-            session_title = payload["message"]["session"]["title"]
+            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"],
+                status=payload["person"]["status"].tame(check_string),
-                name=payload["person"]["_anon"]["name"],
+                name=payload["person"]["_anon"]["name"].tame(check_string),
-                content=payload["message"]["content"],
+                content=payload["message"]["content"].tame(check_string),
             )
             check_send_webhook_message(request, user_profile, topic, body, "chat_message")
     else: