From 08eb34d0d687685a1dbb4d475cd5cc1c8690db44 Mon Sep 17 00:00:00 2001 From: Hari Prashant Bhimaraju Date: Sat, 8 Oct 2022 23:51:17 +0530 Subject: [PATCH] greenhouse: Strengthen types using WildValue. --- zerver/webhooks/greenhouse/view.py | 37 ++++++++++++++++++------------ 1 file changed, 22 insertions(+), 15 deletions(-) diff --git a/zerver/webhooks/greenhouse/view.py b/zerver/webhooks/greenhouse/view.py index bec9f2b163..71b0e97670 100644 --- a/zerver/webhooks/greenhouse/view.py +++ b/zerver/webhooks/greenhouse/view.py @@ -1,10 +1,16 @@ -from typing import Any, Dict, List - 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_int, + check_none_or, + check_string, + check_url, + to_wild_value, +) from zerver.lib.webhooks.common import check_send_webhook_message from zerver.models import UserProfile @@ -16,12 +22,12 @@ MESSAGE_TEMPLATE = """ """.strip() -def dict_list_to_string(some_list: List[Any]) -> str: +def dict_list_to_string(some_list: WildValue) -> str: internal_template = "" for item in some_list: - item_type = item.get("type", "").title() - item_value = item.get("value") - item_url = item.get("url") + item_type = item.get("type", "").tame(check_string).title() + item_value = item.get("value").tame(check_none_or(check_string)) + item_url = item.get("url").tame(check_none_or(check_url)) if item_type and item_value: internal_template += f"{item_value} ({item_type}), " elif item_type and item_url: @@ -36,29 +42,30 @@ def dict_list_to_string(some_list: List[Any]) -> str: def api_greenhouse_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: - if payload["action"] == "ping": + action = payload["action"].tame(check_string) + if action == "ping": return json_success(request) - if payload["action"] == "update_candidate": + if action == "update_candidate": candidate = payload["payload"]["candidate"] else: candidate = payload["payload"]["application"]["candidate"] - action = payload["action"].replace("_", " ").title() + action = action.replace("_", " ").title() application = payload["payload"]["application"] body = MESSAGE_TEMPLATE.format( action=action, - first_name=candidate["first_name"], - last_name=candidate["last_name"], - candidate_id=str(candidate["id"]), - role=application["jobs"][0]["name"], + first_name=candidate["first_name"].tame(check_string), + last_name=candidate["last_name"].tame(check_string), + candidate_id=str(candidate["id"].tame(check_int)), + role=application["jobs"][0]["name"].tame(check_string), emails=dict_list_to_string(application["candidate"]["email_addresses"]), attachments=dict_list_to_string(application["candidate"]["attachments"]), ) - topic = "{} - {}".format(action, str(candidate["id"])) + topic = "{} - {}".format(action, str(candidate["id"].tame(check_int))) check_send_webhook_message(request, user_profile, topic, body) return json_success(request)