From 31c7f8266d4504f40ec3fd491cf651b138653714 Mon Sep 17 00:00:00 2001
From: Hari Prashant Bhimaraju <haripb01@gmail.com>
Date: Thu, 23 Jun 2022 20:51:26 +0530
Subject: [PATCH] sonarqube: Strengthen types using WildValue.

---
 zerver/webhooks/sonarqube/view.py | 51 +++++++++++++++----------------
 1 file changed, 24 insertions(+), 27 deletions(-)

diff --git a/zerver/webhooks/sonarqube/view.py b/zerver/webhooks/sonarqube/view.py
index 670881c6cc..48926851c0 100644
--- a/zerver/webhooks/sonarqube/view.py
+++ b/zerver/webhooks/sonarqube/view.py
@@ -1,12 +1,10 @@
 # Webhooks for external integrations.
-
-from typing import Any, Dict, List, Mapping
-
 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_none_or, check_string, to_wild_value
 from zerver.lib.webhooks.common import check_send_webhook_message
 from zerver.models import UserProfile
 
@@ -35,15 +33,15 @@ def parse_metric_name(metric_name: str) -> str:
     return " ".join(metric_name.split("_"))
 
 
-def parse_condition(condition: Mapping[str, Any]) -> str:
-    metric = condition["metric"]
+def parse_condition(condition: WildValue) -> str:
+    metric = condition["metric"].tame(check_string)
 
     metric_name = parse_metric_name(metric)
-    operator = condition["operator"]
+    operator = condition["operator"].tame(check_string)
     operator = INVERSE_OPERATORS.get(operator, operator)
-    value = condition.get("value", "no value")
-    status = condition["status"].lower()
-    threshold = condition["errorThreshold"]
+    value = condition.get("value", "no value").tame(check_string)
+    status = condition["status"].tame(check_string).lower()
+    threshold = condition["errorThreshold"].tame(check_string)
 
     if value == "no value":
         return TEMPLATES["no_value"].format(metric_name, status)
@@ -53,28 +51,28 @@ def parse_condition(condition: Mapping[str, Any]) -> str:
     return template.format(metric_name, status, value, operator, threshold)
 
 
-def parse_conditions(conditions: List[Mapping[str, Any]]) -> str:
+def parse_conditions(conditions: WildValue) -> str:
     return "\n".join(
         [
             parse_condition(condition)
             for condition in conditions
-            if condition["status"].lower() != "ok" and condition["status"].lower() != "no_value"
+            if condition["status"].tame(check_string).lower() != "ok"
+            and condition["status"].tame(check_string).lower() != "no_value"
         ]
     )
 
 
-def render_body_with_branch(payload: Mapping[str, Any]) -> str:
-    project_name = payload["project"]["name"]
-    project_url = payload["project"]["url"]
-    quality_gate_status = payload["qualityGate"]["status"].lower()
+def render_body_with_branch(payload: WildValue) -> str:
+    project_name = payload["project"]["name"].tame(check_string)
+    project_url = payload["project"]["url"].tame(check_string)
+    quality_gate_status = payload["qualityGate"]["status"].tame(check_string).lower()
     if quality_gate_status == "ok":
         quality_gate_status = "success"
     else:
         quality_gate_status = "error"
-    branch = payload["branch"]["name"]
+    branch = payload["branch"]["name"].tame(check_string)
 
-    conditions = payload["qualityGate"]["conditions"]
-    conditions = parse_conditions(conditions)
+    conditions = parse_conditions(payload["qualityGate"]["conditions"])
 
     if not conditions:
         return MESSAGE_WITH_BRANCH_AND_WITHOUT_CONDITIONS.format(
@@ -88,16 +86,15 @@ def render_body_with_branch(payload: Mapping[str, Any]) -> str:
     return msg
 
 
-def render_body_without_branch(payload: Mapping[str, Any]) -> str:
-    project_name = payload["project"]["name"]
-    project_url = payload["project"]["url"]
-    quality_gate_status = payload["qualityGate"]["status"].lower()
+def render_body_without_branch(payload: WildValue) -> str:
+    project_name = payload["project"]["name"].tame(check_string)
+    project_url = payload["project"]["url"].tame(check_string)
+    quality_gate_status = payload["qualityGate"]["status"].tame(check_string).lower()
     if quality_gate_status == "ok":
         quality_gate_status = "success"
     else:
         quality_gate_status = "error"
-    conditions = payload["qualityGate"]["conditions"]
-    conditions = parse_conditions(conditions)
+    conditions = parse_conditions(payload["qualityGate"]["conditions"])
 
     if not conditions:
         return MESSAGE_WITHOUT_BRANCH_AND_CONDITIONS.format(
@@ -116,12 +113,12 @@ def render_body_without_branch(payload: Mapping[str, Any]) -> str:
 def api_sonarqube_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:
-    project = payload["project"]["name"]
+    project = payload["project"]["name"].tame(check_string)
     branch = None
     if "branch" in payload.keys():
-        branch = payload["branch"].get("name", None)
+        branch = payload["branch"].get("name").tame(check_none_or(check_string))
     if branch:
         topic = TOPIC_WITH_BRANCH.format(project, branch)
         message = render_body_with_branch(payload)