groove: Strengthen types using WildValue.

zerver/webhooks/groove/view.py

@@ -1,6 +1,6 @@
 # Webhooks for external integrations.
 from functools import partial
-from typing import Any, Callable, Dict, Optional
+from typing import Callable, Dict, Optional
 
 from django.http import HttpRequest, HttpResponse
 
@@ -8,6 +8,14 @@ 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_int,
+    check_none_or,
+    check_string,
+    check_url,
+    to_wild_value,
+)
 from zerver.lib.webhooks.common import (
     check_send_webhook_message,
     get_http_headers_from_filename,
@@ -34,52 +42,64 @@ AGENT_REPLIED_TEMPLATE = """
 """.strip()
 
 
-def ticket_started_body(payload: Dict[str, Any]) -> str:
+def ticket_started_body(payload: WildValue) -> str:
-    return TICKET_STARTED_TEMPLATE.format(**payload)
+    return TICKET_STARTED_TEMPLATE.format(
+        customer_name=payload["customer_name"].tame(check_string),
+        number=payload["number"].tame(check_int),
+        title=payload["title"].tame(check_string),
+        app_url=payload["app_url"].tame(check_url),
+        summary=payload["summary"].tame(check_string),
+    )
 
 
-def ticket_assigned_body(payload: Dict[str, Any]) -> Optional[str]:
+def ticket_assigned_body(payload: WildValue) -> Optional[str]:
-    state = payload["state"]
+    state = payload["state"].tame(check_string)
     kwargs = {
         "state": "open" if state == "opened" else state,
-        "number": payload["number"],
+        "number": payload["number"].tame(check_int),
-        "title": payload["title"],
+        "title": payload["title"].tame(check_string),
-        "app_url": payload["app_url"],
+        "app_url": payload["app_url"].tame(check_url),
     }
 
-    assignee = payload["assignee"]
+    assignee = payload["assignee"].tame(check_none_or(check_string))
-    assigned_group = payload["assigned_group"]
+    assigned_group = payload["assigned_group"].tame(check_none_or(check_string))
 
     if assignee or assigned_group:
         if assignee and assigned_group:
-            kwargs["assignee_info"] = "{assignee} from {assigned_group}".format(**payload)
+            kwargs["assignee_info"] = "{assignee} from {assigned_group}".format(
+                assignee=assignee, assigned_group=assigned_group
+            )
         elif assignee:
-            kwargs["assignee_info"] = "{assignee}".format(**payload)
+            kwargs["assignee_info"] = "{assignee}".format(assignee=assignee)
         elif assigned_group:
-            kwargs["assignee_info"] = "{assigned_group}".format(**payload)
+            kwargs["assignee_info"] = "{assigned_group}".format(assigned_group=assigned_group)
 
         return TICKET_ASSIGNED_TEMPLATE.format(**kwargs)
     else:
         return None
 
 
-def replied_body(payload: Dict[str, Any], actor: str, action: str) -> str:
+def replied_body(payload: WildValue, actor: str, action: str) -> str:
     actor_url = "http://api.groovehq.com/v1/{}/".format(actor + "s")
-    actor = payload["links"]["author"]["href"].split(actor_url)[1]
+    actor = payload["links"]["author"]["href"].tame(check_url).split(actor_url)[1]
-    number = payload["links"]["ticket"]["href"].split("http://api.groovehq.com/v1/tickets/")[1]
+    number = (
+        payload["links"]["ticket"]["href"]
+        .tame(check_url)
+        .split("http://api.groovehq.com/v1/tickets/")[1]
+    )
 
     body = AGENT_REPLIED_TEMPLATE.format(
         actor=actor,
         action=action,
         number=number,
-        app_ticket_url=payload["app_ticket_url"],
+        app_ticket_url=payload["app_ticket_url"].tame(check_url),
-        plain_text_body=payload["plain_text_body"],
+        plain_text_body=payload["plain_text_body"].tame(check_string),
     )
 
     return body
 
 
-EVENTS_FUNCTION_MAPPER: Dict[str, Callable[[Dict[str, Any]], Optional[str]]] = {
+EVENTS_FUNCTION_MAPPER: Dict[str, Callable[[WildValue], Optional[str]]] = {
     "ticket_started": ticket_started_body,
     "ticket_assigned": ticket_assigned_body,
     "agent_replied": partial(replied_body, actor="agent", action="replied to"),
@@ -95,7 +115,7 @@ ALL_EVENT_TYPES = list(EVENTS_FUNCTION_MAPPER.keys())
 def api_groove_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:
     event = validate_extract_webhook_http_header(request, "X-Groove-Event", "Groove")
     assert event is not None