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)