front: Strengthen types using WildValue.

zerver/webhooks/front/view.py

@@ -1,4 +1,4 @@
-from typing import Any, Dict, Tuple
+from typing import Any, Tuple
 
 from django.http import HttpRequest, HttpResponse
 from django.utils.translation import gettext as _
@@ -7,31 +7,32 @@ from zerver.decorator import webhook_view
 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.validator import WildValue, check_string, to_wild_value
 from zerver.lib.webhooks.common import check_send_webhook_message
 from zerver.models import UserProfile
 
 
-def get_message_data(payload: Dict[str, Any]) -> Tuple[str, str, str, str]:
-    link = "https://app.frontapp.com/open/" + payload["target"]["data"]["id"]
-    outbox = payload["conversation"]["recipient"]["handle"]
-    inbox = payload["source"]["data"][0]["address"]
-    subject = payload["conversation"]["subject"]
+def get_message_data(payload: WildValue) -> Tuple[str, str, str, str]:
+    link = "https://app.frontapp.com/open/" + payload["target"]["data"]["id"].tame(check_string)
+    outbox = payload["conversation"]["recipient"]["handle"].tame(check_string)
+    inbox = payload["source"]["data"][0]["address"].tame(check_string)
+    subject = payload["conversation"]["subject"].tame(check_string)
     return link, outbox, inbox, subject
 
 
-def get_source_name(payload: Dict[str, Any]) -> str:
-    first_name = payload["source"]["data"]["first_name"]
-    last_name = payload["source"]["data"]["last_name"]
+def get_source_name(payload: WildValue) -> str:
+    first_name = payload["source"]["data"]["first_name"].tame(check_string)
+    last_name = payload["source"]["data"]["last_name"].tame(check_string)
     return f"{first_name} {last_name}"
 
 
-def get_target_name(payload: Dict[str, Any]) -> str:
-    first_name = payload["target"]["data"]["first_name"]
-    last_name = payload["target"]["data"]["last_name"]
+def get_target_name(payload: WildValue) -> str:
+    first_name = payload["target"]["data"]["first_name"].tame(check_string)
+    last_name = payload["target"]["data"]["last_name"].tame(check_string)
     return f"{first_name} {last_name}"
 
 
-def get_inbound_message_body(payload: Dict[str, Any]) -> str:
+def get_inbound_message_body(payload: WildValue) -> str:
     link, outbox, inbox, subject = get_message_data(payload)
     return (
         "[Inbound message]({link}) from **{outbox}** to **{inbox}**:\n"
@@ -41,7 +42,7 @@ def get_inbound_message_body(payload: Dict[str, Any]) -> str:
     )
 
 
-def get_outbound_message_body(payload: Dict[str, Any]) -> str:
+def get_outbound_message_body(payload: WildValue) -> str:
     link, outbox, inbox, subject = get_message_data(payload)
     return (
         "[Outbound message]({link}) from **{inbox}** to **{outbox}**:\n"
@@ -51,20 +52,20 @@ def get_outbound_message_body(payload: Dict[str, Any]) -> str:
     )
 
 
-def get_outbound_reply_body(payload: Dict[str, Any]) -> str:
+def get_outbound_reply_body(payload: WildValue) -> str:
     link, outbox, inbox, subject = get_message_data(payload)
     return "[Outbound reply]({link}) from **{inbox}** to **{outbox}**.".format(
         link=link, inbox=inbox, outbox=outbox
     )
 
 
-def get_comment_body(payload: Dict[str, Any]) -> str:
+def get_comment_body(payload: WildValue) -> str:
     name = get_source_name(payload)
-    comment = payload["target"]["data"]["body"]
+    comment = payload["target"]["data"]["body"].tame(check_string)
     return f"**{name}** left a comment:\n```quote\n{comment}\n```"
 
 
-def get_conversation_assigned_body(payload: Dict[str, Any]) -> str:
+def get_conversation_assigned_body(payload: WildValue) -> str:
     source_name = get_source_name(payload)
     target_name = get_target_name(payload)
 
@@ -76,40 +77,40 @@ def get_conversation_assigned_body(payload: Dict[str, Any]) -> str:
     )
 
 
-def get_conversation_unassigned_body(payload: Dict[str, Any]) -> str:
+def get_conversation_unassigned_body(payload: WildValue) -> str:
     name = get_source_name(payload)
     return f"Unassigned by **{name}**."
 
 
-def get_conversation_archived_body(payload: Dict[str, Any]) -> str:
+def get_conversation_archived_body(payload: WildValue) -> str:
     name = get_source_name(payload)
     return f"Archived by **{name}**."
 
 
-def get_conversation_reopened_body(payload: Dict[str, Any]) -> str:
+def get_conversation_reopened_body(payload: WildValue) -> str:
     name = get_source_name(payload)
     return f"Reopened by **{name}**."
 
 
-def get_conversation_deleted_body(payload: Dict[str, Any]) -> str:
+def get_conversation_deleted_body(payload: WildValue) -> str:
     name = get_source_name(payload)
     return f"Deleted by **{name}**."
 
 
-def get_conversation_restored_body(payload: Dict[str, Any]) -> str:
+def get_conversation_restored_body(payload: WildValue) -> str:
     name = get_source_name(payload)
     return f"Restored by **{name}**."
 
 
-def get_conversation_tagged_body(payload: Dict[str, Any]) -> str:
+def get_conversation_tagged_body(payload: WildValue) -> str:
     name = get_source_name(payload)
-    tag = payload["target"]["data"]["name"]
+    tag = payload["target"]["data"]["name"].tame(check_string)
     return f"**{name}** added tag **{tag}**."
 
 
-def get_conversation_untagged_body(payload: Dict[str, Any]) -> str:
+def get_conversation_untagged_body(payload: WildValue) -> str:
     name = get_source_name(payload)
-    tag = payload["target"]["data"]["name"]
+    tag = payload["target"]["data"]["name"].tame(check_string)
     return f"**{name}** removed tag **{tag}**."
 
 
@@ -141,14 +142,14 @@ def get_body_based_on_event(event: str) -> Any:
 def api_front_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 = payload["type"]
+    event = payload["type"].tame(check_string)
     if event not in EVENT_FUNCTION_MAPPER:
         raise JsonableError(_("Unknown webhook request"))
 
-    topic = payload["conversation"]["id"]
+    topic = payload["conversation"]["id"].tame(check_string)
     body = get_body_based_on_event(event)(payload)
     check_send_webhook_message(request, user_profile, topic, body, event)