grafana: Strengthen types using WildValue.

zerver/webhooks/grafana/view.py

@@ -1,10 +1,18 @@
-from typing import Any, Dict
-
 from django.http import HttpRequest, HttpResponse
 
 from zerver.decorator import webhook_view
 from zerver.lib.request import REQ, has_request_variables
 from zerver.lib.response import json_success
+from zerver.lib.validator import (
+    WildValue,
+    check_float,
+    check_int,
+    check_none_or,
+    check_string,
+    check_string_in,
+    check_union,
+    to_wild_value,
+)
 from zerver.lib.webhooks.common import check_send_webhook_message
 from zerver.models import UserProfile
 
@@ -24,51 +32,55 @@ ALL_EVENT_TYPES = ["ok", "pending", "alerting", "paused"]
 def api_grafana_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:
 
-    topic = GRAFANA_TOPIC_TEMPLATE.format(alert_title=payload["title"])
+    topic = GRAFANA_TOPIC_TEMPLATE.format(alert_title=payload["title"].tame(check_string))
 
     eval_matches_text = ""
-    eval_matches = payload.get("evalMatches")
+    if "evalMatches" in payload and payload["evalMatches"] is not None:
-    if eval_matches is not None:
+        for match in payload["evalMatches"]:
-        for match in eval_matches:
+            eval_matches_text += "**{}:** {}\n".format(
-            eval_matches_text += "**{}:** {}\n".format(match["metric"], match["value"])
+                match["metric"].tame(check_string),
+                match["value"].tame(check_none_or(check_union([check_int, check_float]))),
+            )
 
     message_text = ""
-    if payload.get("message") is not None:
+    if "message" in payload:
-        message_text = payload["message"] + "\n\n"
+        message_text = payload["message"].tame(check_string) + "\n\n"
 
-    if payload.get("state") is not None:
+    state = payload["state"].tame(
-        if payload.get("state") == "alerting":
+        check_string_in(["no_data", "paused", "alerting", "ok", "pending", "unknown"])
-            alert_status = GRAFANA_ALERT_STATUS_TEMPLATE.format(
-                alert_icon=":alert:", alert_state=payload["state"].upper()
     )
-        elif payload.get("state") == "ok":
+    if state == "alerting":
         alert_status = GRAFANA_ALERT_STATUS_TEMPLATE.format(
-                alert_icon=":squared_ok:", alert_state=payload["state"].upper()
+            alert_icon=":alert:", alert_state=state.upper()
+        )
+    elif state == "ok":
+        alert_status = GRAFANA_ALERT_STATUS_TEMPLATE.format(
+            alert_icon=":squared_ok:", alert_state=state.upper()
         )
     else:
         alert_status = GRAFANA_ALERT_STATUS_TEMPLATE.format(
-                alert_icon=":info:", alert_state=payload["state"].upper()
+            alert_icon=":info:", alert_state=state.upper()
         )
 
     body = GRAFANA_MESSAGE_TEMPLATE.format(
         alert_message=message_text,
         alert_status=alert_status,
-        rule_name=payload["ruleName"],
+        rule_name=payload["ruleName"].tame(check_string),
-        rule_url=payload["ruleUrl"],
+        rule_url=payload["ruleUrl"].tame(check_string),
         eval_matches=eval_matches_text,
     )
 
-    if payload.get("imageUrl") is not None:
+    if "imageUrl" in payload:
         body += "\n[Click to view visualization]({visualization})".format(
-            visualization=payload["imageUrl"]
+            visualization=payload["imageUrl"].tame(check_string)
         )
 
     body = body.strip()
 
     # send the message
-    check_send_webhook_message(request, user_profile, topic, body, payload.get("state"))
+    check_send_webhook_message(request, user_profile, topic, body, state)
 
     return json_success(request)