diff --git a/zerver/webhooks/lidarr/view.py b/zerver/webhooks/lidarr/view.py index 63d3451177..8128a66383 100644 --- a/zerver/webhooks/lidarr/view.py +++ b/zerver/webhooks/lidarr/view.py @@ -1,4 +1,4 @@ -from typing import Any, Dict, List +from typing import Dict, List from django.http import HttpRequest, HttpResponse @@ -6,6 +6,7 @@ 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_bool, check_string, to_wild_value from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message from zerver.models import UserProfile @@ -39,7 +40,7 @@ LIDARR_TRACKS_LIMIT = 20 ALL_EVENT_TYPES = ["Test", "Grab", "Rename", "Retag", "Download"] -def get_tracks_content(tracks_data: List[Dict[str, Any]]) -> str: +def get_tracks_content(tracks_data: List[Dict[str, str]]) -> str: tracks_content = "" for track in tracks_data[:LIDARR_TRACKS_LIMIT]: tracks_content += LIDARR_TRACKS_ROW_TEMPLATE.format(track_title=track.get("title")) @@ -57,77 +58,87 @@ def get_tracks_content(tracks_data: List[Dict[str, Any]]) -> str: def api_lidarr_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: body = get_body_for_http_request(payload) subject = get_subject_for_http_request(payload) - check_send_webhook_message(request, user_profile, subject, body, payload["eventType"]) + check_send_webhook_message( + request, user_profile, subject, body, payload["eventType"].tame(check_string) + ) return json_success(request) -def get_subject_for_http_request(payload: Dict[str, Any]) -> str: - if payload["eventType"] == "Test": +def get_subject_for_http_request(payload: WildValue) -> str: + if payload["eventType"].tame(check_string) == "Test": topic = LIDARR_TOPIC_TEMPLATE_TEST else: - topic = LIDARR_TOPIC_TEMPLATE.format(artist_name=payload["artist"]["name"]) + topic = LIDARR_TOPIC_TEMPLATE.format( + artist_name=payload["artist"]["name"].tame(check_string) + ) return topic -def get_body_for_album_grabbed_event(payload: Dict[str, Any]) -> str: +def get_body_for_album_grabbed_event(payload: WildValue) -> str: return LIDARR_MESSAGE_TEMPLATE_ALBUM_GRABBED.format( - artist_name=payload["artist"]["name"], album_name=payload["albums"][0]["title"] + artist_name=payload["artist"]["name"].tame(check_string), + album_name=payload["albums"][0]["title"].tame(check_string), ) -def get_body_for_tracks_renamed_event(payload: Dict[str, Any]) -> str: - return LIDARR_MESSAGE_TEMPLATE_TRACKS_RENAMED.format(artist_name=payload["artist"]["name"]) +def get_body_for_tracks_renamed_event(payload: WildValue) -> str: + return LIDARR_MESSAGE_TEMPLATE_TRACKS_RENAMED.format( + artist_name=payload["artist"]["name"].tame(check_string) + ) -def get_body_for_tracks_retagged_event(payload: Dict[str, Any]) -> str: - return LIDARR_MESSAGE_TEMPLATE_TRACKS_RETAGGED.format(artist_name=payload["artist"]["name"]) +def get_body_for_tracks_retagged_event(payload: WildValue) -> str: + return LIDARR_MESSAGE_TEMPLATE_TRACKS_RETAGGED.format( + artist_name=payload["artist"]["name"].tame(check_string) + ) -def get_body_for_tracks_imported_upgrade_event(payload: Dict[str, Any]) -> str: +def get_body_for_tracks_imported_upgrade_event(payload: WildValue) -> str: tracks_data = [] for track in payload["tracks"]: - tracks_data.append({"title": track["title"]}) + tracks_data.append({"title": track["title"].tame(check_string)}) data = { - "artist_name": payload["artist"]["name"], + "artist_name": payload["artist"]["name"].tame(check_string), "tracks_final_data": get_tracks_content(tracks_data), } return LIDARR_MESSAGE_TEMPLATE_TRACKS_IMPORTED_UPGRADE.format(**data) -def get_body_for_tracks_imported_event(payload: Dict[str, Any]) -> str: +def get_body_for_tracks_imported_event(payload: WildValue) -> str: tracks_data = [] for track in payload["tracks"]: - tracks_data.append({"title": track["title"]}) + tracks_data.append({"title": track["title"].tame(check_string)}) data = { - "artist_name": payload["artist"]["name"], + "artist_name": payload["artist"]["name"].tame(check_string), "tracks_final_data": get_tracks_content(tracks_data), } return LIDARR_MESSAGE_TEMPLATE_TRACKS_IMPORTED.format(**data) -def get_body_for_http_request(payload: Dict[str, Any]) -> str: - if payload["eventType"] == "Test": +def get_body_for_http_request(payload: WildValue) -> str: + event_type = payload["eventType"].tame(check_string) + if event_type == "Test": return get_setup_webhook_message("Lidarr") - elif payload["eventType"] == "Grab": + elif event_type == "Grab": return get_body_for_album_grabbed_event(payload) - elif payload["eventType"] == "Rename": + elif event_type == "Rename": return get_body_for_tracks_renamed_event(payload) - elif payload["eventType"] == "Retag": + elif event_type == "Retag": return get_body_for_tracks_retagged_event(payload) - elif payload["eventType"] == "Download" and "isUpgrade" in payload: - if payload["isUpgrade"]: + elif event_type == "Download" and "isUpgrade" in payload: + if payload["isUpgrade"].tame(check_bool): return get_body_for_tracks_imported_upgrade_event(payload) else: return get_body_for_tracks_imported_event(payload) else: - raise UnsupportedWebhookEventType(payload["eventType"]) + raise UnsupportedWebhookEventType(event_type)