diff --git a/zerver/webhooks/groove/view.py b/zerver/webhooks/groove/view.py index 72d3da761a..14c45ca70f 100644 --- a/zerver/webhooks/groove/view.py +++ b/zerver/webhooks/groove/view.py @@ -1,6 +1,6 @@ # Webhooks for external integrations. from functools import partial -from typing import Any, Callable, Dict, Optional +from typing import Callable, Dict, Optional from django.http import HttpRequest, HttpResponse @@ -8,6 +8,14 @@ from zerver.decorator import webhook_view from zerver.lib.exceptions import UnsupportedWebhookEventType 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, get_http_headers_from_filename, @@ -34,52 +42,64 @@ AGENT_REPLIED_TEMPLATE = """ """.strip() -def ticket_started_body(payload: Dict[str, Any]) -> str: - return TICKET_STARTED_TEMPLATE.format(**payload) +def ticket_started_body(payload: WildValue) -> str: + return TICKET_STARTED_TEMPLATE.format( + customer_name=payload["customer_name"].tame(check_string), + number=payload["number"].tame(check_int), + title=payload["title"].tame(check_string), + app_url=payload["app_url"].tame(check_url), + summary=payload["summary"].tame(check_string), + ) -def ticket_assigned_body(payload: Dict[str, Any]) -> Optional[str]: - state = payload["state"] +def ticket_assigned_body(payload: WildValue) -> Optional[str]: + state = payload["state"].tame(check_string) kwargs = { "state": "open" if state == "opened" else state, - "number": payload["number"], - "title": payload["title"], - "app_url": payload["app_url"], + "number": payload["number"].tame(check_int), + "title": payload["title"].tame(check_string), + "app_url": payload["app_url"].tame(check_url), } - assignee = payload["assignee"] - assigned_group = payload["assigned_group"] + assignee = payload["assignee"].tame(check_none_or(check_string)) + assigned_group = payload["assigned_group"].tame(check_none_or(check_string)) if assignee or assigned_group: if assignee and assigned_group: - kwargs["assignee_info"] = "{assignee} from {assigned_group}".format(**payload) + kwargs["assignee_info"] = "{assignee} from {assigned_group}".format( + assignee=assignee, assigned_group=assigned_group + ) elif assignee: - kwargs["assignee_info"] = "{assignee}".format(**payload) + kwargs["assignee_info"] = "{assignee}".format(assignee=assignee) elif assigned_group: - kwargs["assignee_info"] = "{assigned_group}".format(**payload) + kwargs["assignee_info"] = "{assigned_group}".format(assigned_group=assigned_group) return TICKET_ASSIGNED_TEMPLATE.format(**kwargs) else: return None -def replied_body(payload: Dict[str, Any], actor: str, action: str) -> str: +def replied_body(payload: WildValue, actor: str, action: str) -> str: actor_url = "http://api.groovehq.com/v1/{}/".format(actor + "s") - actor = payload["links"]["author"]["href"].split(actor_url)[1] - number = payload["links"]["ticket"]["href"].split("http://api.groovehq.com/v1/tickets/")[1] + actor = payload["links"]["author"]["href"].tame(check_url).split(actor_url)[1] + number = ( + payload["links"]["ticket"]["href"] + .tame(check_url) + .split("http://api.groovehq.com/v1/tickets/")[1] + ) body = AGENT_REPLIED_TEMPLATE.format( actor=actor, action=action, number=number, - app_ticket_url=payload["app_ticket_url"], - plain_text_body=payload["plain_text_body"], + app_ticket_url=payload["app_ticket_url"].tame(check_url), + plain_text_body=payload["plain_text_body"].tame(check_string), ) return body -EVENTS_FUNCTION_MAPPER: Dict[str, Callable[[Dict[str, Any]], Optional[str]]] = { +EVENTS_FUNCTION_MAPPER: Dict[str, Callable[[WildValue], Optional[str]]] = { "ticket_started": ticket_started_body, "ticket_assigned": ticket_assigned_body, "agent_replied": partial(replied_body, actor="agent", action="replied to"), @@ -95,7 +115,7 @@ ALL_EVENT_TYPES = list(EVENTS_FUNCTION_MAPPER.keys()) def api_groove_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 = validate_extract_webhook_http_header(request, "X-Groove-Event", "Groove") assert event is not None