Mirror
/
zulip
mirror of https://github.com/zulip/zulip.git
1
0
Fork 0
Code Issues Projects Releases Wiki Activity

webhooks: Migrate most webhooks to use @typed_endpoint. 

This converts most webhook integration views to use @typed_endpoint instead
of @has_request_variables, rewriting REQ parameters. For these
webhooks, it simply requires switching the decorator, rewriting the
type annotation of payload/message to WebhookPayload[WildValue], and
removing the REQ default that defines the to_wild_value converter.
This commit is contained in:
Zixuan James Li 2023-08-12 03:34:31 -04:00 committed by Tim Abbott
parent 574740dda4
commit 9377080f1f
77 changed files with 418 additions and 383 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

9
zerver/webhooks/airbrake/view.py
View File

@ -2,9 +2,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -13,11 +13,12 @@ AIRBRAKE_MESSAGE_TEMPLATE = '[{error_class}]({error_url}): "{error_message}" occ
@webhook_view("Airbrake")
@has_request_variables
@typed_endpoint
def api_airbrake_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
topic = get_topic(payload)
body = get_body(payload)

14
zerver/webhooks/alertmanager/view.py
View File

@ -2,23 +2,25 @@
from typing import Dict, List
from django.http import HttpRequest, HttpResponse
from typing_extensions import Annotated
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_string, to_wild_value
from zerver.lib.typed_endpoint import ApiParamConfig, WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Alertmanager")
@has_request_variables
@typed_endpoint
def api_alertmanager_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
name_field: str = REQ("name", default="instance"),
desc_field: str = REQ("desc", default="alertname"),
*,
payload: WebhookPayload[WildValue],
name_field: Annotated[str, ApiParamConfig("name")] = "instance",
desc_field: Annotated[str, ApiParamConfig("desc")] = "alertname",
) -> HttpResponse:
topics: Dict[str, Dict[str, List[str]]] = {}

9
zerver/webhooks/ansibletower/view.py
View File

@ -4,9 +4,9 @@ from typing import 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_bool, check_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -22,11 +22,12 @@ ANSIBLETOWER_JOB_HOST_ROW_TEMPLATE = "* {hostname}: {status}\n"
@webhook_view("AnsibleTower")
@has_request_variables
@typed_endpoint
def api_ansibletower_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = get_body(payload)
topic = payload["name"].tame(check_string)

9
zerver/webhooks/appfollow/view.py
View File

@ -4,19 +4,20 @@ import re
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("AppFollow")
@has_request_variables
@typed_endpoint
def api_appfollow_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
message = payload["text"].tame(check_string)
app_name_search = re.search(r"\A(.+)", message)

9
zerver/webhooks/appveyor/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -17,11 +17,12 @@ APPVEYOR_MESSAGE_TEMPLATE = """
@webhook_view("Appveyor")
@has_request_variables
@typed_endpoint
def api_appveyor_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = get_body_for_http_request(payload)
topic = get_topic_for_http_request(payload)

11
zerver/webhooks/azuredevops/view.py
View File

@ -4,9 +4,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import (
TOPIC_WITH_BRANCH_TEMPLATE,
@ -169,12 +169,13 @@ ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())
@webhook_view("AzureDevOps", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_azuredevops_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
) -> HttpResponse:
event = get_event_name(payload, branches)
if event is None:

9
zerver/webhooks/basecamp/view.py
View File

@ -5,9 +5,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -38,11 +38,12 @@ ALL_EVENT_TYPES = [
@webhook_view("Basecamp", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_basecamp_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = get_event_type(payload)

12
zerver/webhooks/beanstalk/view.py
View File

@ -3,11 +3,12 @@ import re
from typing import Dict, List, Optional, Tuple
from django.http import HttpRequest, HttpResponse
from pydantic import Json
from zerver.decorator import authenticated_rest_api_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_string, to_wild_value
from zerver.lib.typed_endpoint import typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import TOPIC_WITH_BRANCH_TEMPLATE, get_push_commits_event_message
from zerver.models import UserProfile
@ -53,12 +54,13 @@ def _transform_commits_list_to_common_format(commits: WildValue) -> List[Dict[st
# So we ask the user to replace them with %40
beanstalk_email_decode=True,
)
@has_request_variables
@typed_endpoint
def api_beanstalk_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
*,
payload: Json[WildValue],
branches: Optional[str] = None,
) -> HttpResponse:
# Beanstalk supports both SVN and Git repositories
# We distinguish between the two by checking for a

16
zerver/webhooks/beeminder/view.py
View File

@ -4,16 +4,9 @@ import time
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_float,
check_int,
check_string,
check_union,
to_wild_value,
)
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_float, check_int, check_string, check_union
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -31,11 +24,12 @@ def get_time(payload: WildValue) -> float:
@webhook_view("Beeminder")
@has_request_variables
@typed_endpoint
def api_beeminder_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
goal_name = payload["goal"]["slug"].tame(check_string)
limsum = payload["goal"]["limsum"].tame(check_string)

11
zerver/webhooks/bitbucket/view.py
View File

@ -3,21 +3,22 @@ from typing import Optional
from django.http import HttpRequest, HttpResponse
from zerver.decorator import authenticated_rest_api_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import TOPIC_WITH_BRANCH_TEMPLATE, get_push_commits_event_message
from zerver.models import UserProfile
@authenticated_rest_api_view(webhook_client_name="Bitbucket")
@has_request_variables
@typed_endpoint
def api_bitbucket_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
) -> HttpResponse:
repository = payload["repository"]

14
zerver/webhooks/bitbucket2/view.py
View File

@ -8,10 +8,11 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import log_unsupported_webhook_event, webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_string
from zerver.lib.webhooks.common import (
OptionalUserSpecifiedTopicStr,
check_send_webhook_message,
validate_extract_webhook_http_header,
)
@ -76,13 +77,14 @@ ALL_EVENT_TYPES = [
@webhook_view("Bitbucket2", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_bitbucket2_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
type = get_type(request, payload)
if type == "push":

14
zerver/webhooks/bitbucket3/view.py
View File

@ -6,10 +6,11 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import (
OptionalUserSpecifiedTopicStr,
check_send_webhook_message,
validate_extract_webhook_http_header,
)
@ -422,13 +423,14 @@ ALL_EVENT_TYPES = list(EVENT_HANDLER_MAP.keys())
@webhook_view("Bitbucket3", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_bitbucket3_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
eventkey: Optional[str]
if "eventKey" in payload:

9
zerver/webhooks/buildbot/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -11,11 +11,12 @@ ALL_EVENT_TYPES = ["new", "finished"]
@webhook_view("Buildbot", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_buildbot_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
topic = payload["project"].tame(check_string)
if not topic:

14
zerver/webhooks/canarytoken/view.py
View File

@ -1,23 +1,23 @@
# Webhooks for external integrations.
from typing import Optional
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_string, to_wild_value
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import OptionalUserSpecifiedTopicStr, check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Canarytokens")
@has_request_variables
@typed_endpoint
def api_canarytoken_webhook(
request: HttpRequest,
user_profile: UserProfile,
message: WildValue = REQ(argument_type="body", converter=to_wild_value),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
message: WebhookPayload[WildValue],
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
"""
Construct a response to a webhook event from a Thinkst canarytoken from

8
zerver/webhooks/circleci/view.py
View File

@ -3,8 +3,8 @@ from django.utils.translation import gettext as _
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import (
WildValue,
check_int,
@ -12,7 +12,6 @@ from zerver.lib.validator import (
check_string,
check_string_in,
check_url,
to_wild_value,
)
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import get_short_sha
@ -58,11 +57,12 @@ ALL_EVENT_TYPES = ["ping", "job-completed", "workflow-completed"]
@webhook_view("CircleCI", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_circleci_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
type = payload["type"].tame(check_string)
if type == "ping":

8
zerver/webhooks/clubhouse/view.py
View File

@ -5,8 +5,8 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import (
WildValue,
check_bool,
@ -15,7 +15,6 @@ from zerver.lib.validator import (
check_none_or,
check_string,
check_string_or_int,
to_wild_value,
)
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -767,11 +766,12 @@ EVENTS_SECONDARY_ACTIONS_FUNCTION_MAPPER: Dict[str, Callable[[WildValue], Iterat
@webhook_view("Clubhouse", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_clubhouse_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# Clubhouse has a tendency to send empty POST requests to
# third-party endpoints. It is unclear as to which event type

9
zerver/webhooks/codeship/view.py
View File

@ -2,9 +2,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -22,11 +22,12 @@ CODESHIP_STATUS_MAPPER = {
@webhook_view("Codeship")
@has_request_variables
@typed_endpoint
def api_codeship_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
payload = payload["build"]
topic = get_topic_for_http_request(payload)

9
zerver/webhooks/crashlytics/view.py
View File

@ -2,9 +2,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -18,11 +18,12 @@ VERIFICATION_EVENT = "verification"
@webhook_view("Crashlytics")
@has_request_variables
@typed_endpoint
def api_crashlytics_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = payload["event"]
if event == VERIFICATION_EVENT:

9
zerver/webhooks/delighted/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -32,11 +32,12 @@ def body_template(score: int) -> str:
@webhook_view("Delighted")
@has_request_variables
@typed_endpoint
def api_delighted_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
person = payload["event_data"]["person"]
email = person["email"].tame(check_string)

9
zerver/webhooks/deskdotcom/view.py
View File

@ -2,8 +2,8 @@
from django.http import HttpRequest, HttpResponse
from zerver.decorator import authenticated_rest_api_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import typed_endpoint
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -14,9 +14,12 @@ from zerver.models import UserProfile
# a template Zulip message within Desk.com and have the webhook extract that
# from the "data" param and post it, which this does.
@authenticated_rest_api_view(webhook_client_name="Desk")
@has_request_variables
@typed_endpoint
def api_deskdotcom_webhook(
request: HttpRequest, user_profile: UserProfile, data: str = REQ()
request: HttpRequest,
user_profile: UserProfile,
*,
data: str,
) -> HttpResponse:
topic = "Desk.com notification"
check_send_webhook_message(request, user_profile, topic, data)

12
zerver/webhooks/dialogflow/view.py
View File

@ -3,19 +3,21 @@ from django.http import HttpRequest, HttpResponse
from zerver.actions.message_send import check_send_private_message
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, RequestNotes, has_request_variables
from zerver.lib.request import RequestNotes
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.models import UserProfile, get_user
@webhook_view("Dialogflow")
@has_request_variables
@typed_endpoint
def api_dialogflow_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
email: str = REQ(),
*,
payload: WebhookPayload[WildValue],
email: str,
) -> HttpResponse:
status = payload["status"]["code"].tame(check_int)

8
zerver/webhooks/dropbox/view.py
View File

@ -3,18 +3,20 @@ from typing import Optional
from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, RequestVariableMissingError, has_request_variables
from zerver.lib.request import RequestVariableMissingError
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import typed_endpoint
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Dropbox", notify_bot_owner_on_invalid_json=False)
@has_request_variables
@typed_endpoint
def api_dropbox_webhook(
request: HttpRequest,
user_profile: UserProfile,
challenge: Optional[str] = REQ(default=None),
*,
challenge: Optional[str] = None,
) -> HttpResponse:
if request.method == "POST":
topic = "Dropbox"

9
zerver/webhooks/errbit/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -12,11 +12,12 @@ ERRBIT_MESSAGE_TEMPLATE = '[{error_class}]({error_url}): "{error_message}" occur
@webhook_view("Errbit")
@has_request_variables
@typed_endpoint
def api_errbit_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
topic = get_topic(payload)
body = get_body(payload)

9
zerver/webhooks/flock/view.py
View File

@ -2,19 +2,20 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Flock")
@has_request_variables
@typed_endpoint
def api_flock_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
text = payload["text"].tame(check_string)
if len(text) != 0:

9
zerver/webhooks/freshdesk/view.py
View File

@ -5,9 +5,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import authenticated_rest_api_view
from zerver.lib.email_notifications import convert_html_to_markdown
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -131,11 +131,12 @@ def format_freshdesk_ticket_creation_message(ticket: WildValue) -> str:
@authenticated_rest_api_view(webhook_client_name="Freshdesk")
@has_request_variables
@typed_endpoint
def api_freshdesk_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
ticket = payload["freshdesk_webhook"]

9
zerver/webhooks/freshping/view.py
View File

@ -2,9 +2,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message
from zerver.models import UserProfile
@ -21,11 +21,12 @@ ALL_EVENT_TYPES = list(CHECK_STATE_NAME_TO_EVENT_TYPE.values())
@webhook_view("Freshping", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_freshping_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = get_body_for_http_request(payload)
topic = get_topic_for_http_request(payload)

9
zerver/webhooks/freshstatus/view.py
View File

@ -8,10 +8,10 @@ from django.utils.translation import gettext as _
from zerver.actions.message_send import send_rate_limited_pm_notification_to_bot_owner
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.send_email import FromAddress
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message
from zerver.models import UserProfile
@ -79,11 +79,12 @@ ALL_EVENT_TYPES = [
@webhook_view("Freshstatus", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_freshstatus_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
try:
body = get_body_for_http_request(payload)

9
zerver/webhooks/front/view.py
View File

@ -5,9 +5,9 @@ from django.utils.translation import gettext as _
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -136,11 +136,12 @@ def get_body_based_on_event(event: str) -> Callable[[WildValue], str]:
@webhook_view("Front", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_front_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = payload["type"].tame(check_string)
if event not in EVENT_FUNCTION_MAPPER:

9
zerver/webhooks/gci/view.py
View File

@ -3,9 +3,9 @@ from typing import Callable, Optional
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_float, check_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_float, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -133,11 +133,12 @@ ALL_EVENT_TYPES = list(EVENTS_FUNCTION_MAPPER.keys())
@webhook_view("GoogleCodeIn", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_gci_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = get_event(payload)
if event is not None:

15
zerver/webhooks/gitea/view.py
View File

@ -4,9 +4,9 @@ from typing import Optional
from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.validator import WildValue, check_bool, check_int, check_string, to_wild_value
from zerver.lib.webhooks.common import get_http_headers_from_filename
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_string
from zerver.lib.webhooks.common import OptionalUserSpecifiedTopicStr, get_http_headers_from_filename
from zerver.lib.webhooks.git import get_pull_request_event_message
from zerver.models import UserProfile
@ -51,13 +51,14 @@ def format_pull_request_event(payload: WildValue, include_title: bool = False) -
@webhook_view("Gitea")
@has_request_variables
@typed_endpoint
def api_gitea_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
return gogs_webhook_main(
"Gitea",

21
zerver/webhooks/github/view.py
View File

@ -6,17 +6,11 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import log_unsupported_webhook_event, webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_int,
check_none_or,
check_string,
to_wild_value,
)
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import (
OptionalUserSpecifiedTopicStr,
check_send_webhook_message,
get_http_headers_from_filename,
get_setup_webhook_message,
@ -796,13 +790,14 @@ ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())
@webhook_view("GitHub", notify_bot_owner_on_invalid_json=True, all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_github_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
"""
GitHub sends the event as an HTTP header. We have our

9
zerver/webhooks/gocd/view.py
View File

@ -5,9 +5,9 @@ import os
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -19,11 +19,12 @@ Comment: {}"""
@webhook_view("Gocd")
@has_request_variables
@typed_endpoint
def api_gocd_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
modifications = payload["build_cause"]["material_revisions"][0]["modifications"][0]
result = payload["stages"][0]["result"].tame(check_string)

14
zerver/webhooks/gogs/view.py
View File

@ -5,10 +5,11 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_string
from zerver.lib.webhooks.common import (
OptionalUserSpecifiedTopicStr,
check_send_webhook_message,
get_http_headers_from_filename,
validate_extract_webhook_http_header,
@ -150,13 +151,14 @@ ALL_EVENT_TYPES = ["issue_comment", "issues", "create", "pull_request", "push",
@webhook_view("Gogs", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_gogs_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
return gogs_webhook_main(
"Gogs",

9
zerver/webhooks/gosquared/view.py
View File

@ -2,9 +2,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -22,11 +22,12 @@ ALL_EVENT_TYPES = ["chat_message", "traffic_spike"]
@webhook_view("GoSquared", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_gosquared_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = ""
topic = ""

8
zerver/webhooks/grafana/view.py
View File

@ -1,8 +1,8 @@
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import (
WildValue,
check_float,
@ -11,7 +11,6 @@ from zerver.lib.validator import (
check_string,
check_string_in,
check_union,
to_wild_value,
)
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -45,11 +44,12 @@ ALL_EVENT_TYPES = LEGACY_EVENT_TYPES + NEW_EVENT_TYPES
@webhook_view("Grafana", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_grafana_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# Grafana alerting system.
if "alerts" in payload:

16
zerver/webhooks/greenhouse/view.py
View File

@ -1,16 +1,9 @@
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string, check_url
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -38,11 +31,12 @@ def dict_list_to_string(some_list: WildValue) -> str:
@webhook_view("Greenhouse")
@has_request_variables
@typed_endpoint
def api_greenhouse_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
action = payload["action"].tame(check_string)
if action == "ping":

16
zerver/webhooks/groove/view.py
View File

@ -6,16 +6,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string, check_url
from zerver.lib.webhooks.common import (
check_send_webhook_message,
get_http_headers_from_filename,
@ -109,11 +102,12 @@ ALL_EVENT_TYPES = list(EVENTS_FUNCTION_MAPPER.keys())
@webhook_view("Groove", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_groove_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = validate_extract_webhook_http_header(request, "X-Groove-Event", "Groove")
assert event is not None

9
zerver/webhooks/harbor/view.py
View File

@ -6,9 +6,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import Realm, UserProfile
@ -90,11 +90,12 @@ ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())
@webhook_view("Harbor", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_harbor_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
operator_username = "**{}**".format(payload["operator"].tame(check_string))

9
zerver/webhooks/homeassistant/view.py
View File

@ -1,19 +1,20 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("HomeAssistant")
@has_request_variables
@typed_endpoint
def api_homeassistant_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# construct the body of the message
body = payload["message"].tame(check_string)

9
zerver/webhooks/ifttt/view.py
View File

@ -4,19 +4,20 @@ from django.utils.translation import gettext as _
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("IFTTT")
@has_request_variables
@typed_endpoint
def api_iftt_app_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
try:
topic = payload.get("topic").tame(check_none_or(check_string))

9
zerver/webhooks/insping/view.py
View File

@ -3,9 +3,9 @@ import time
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -18,11 +18,12 @@ State changed to **{state}**:
@webhook_view("Insping")
@has_request_variables
@typed_endpoint
def api_insping_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
data = payload["webhook_event_data"]

9
zerver/webhooks/intercom/view.py
View File

@ -6,9 +6,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import return_success_on_head_request, webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -327,11 +327,12 @@ ALL_EVENT_TYPES = list(EVENT_TO_FUNCTION_MAPPER.keys())
@webhook_view("Intercom", all_event_types=ALL_EVENT_TYPES)
# Intercom sends a HEAD request to validate the webhook URL. In this case, we just assume success.
@return_success_on_head_request
@has_request_variables
@typed_endpoint
def api_intercom_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event_type = payload["topic"].tame(check_string)
if event_type == "ping":

9
zerver/webhooks/jira/view.py
View File

@ -9,9 +9,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import AnomalousWebhookPayloadError, UnsupportedWebhookEventTypeError
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import Realm, UserProfile, get_user_by_delivery_email
@ -351,11 +351,12 @@ ALL_EVENT_TYPES = list(JIRA_CONTENT_FUNCTION_MAPPER.keys())
@webhook_view("Jira", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_jira_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = get_event_type(payload)
if event in IGNORED_EVENTS:

9
zerver/webhooks/jotform/view.py
View File

@ -3,19 +3,20 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Jotform")
@has_request_variables
@typed_endpoint
def api_jotform_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
topic = payload["formTitle"].tame(check_string)
submission_id = payload["submissionID"].tame(check_string)

9
zerver/webhooks/lidarr/view.py
View File

@ -4,9 +4,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_string
from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message
from zerver.models import UserProfile
@ -54,11 +54,12 @@ def get_tracks_content(tracks_data: List[Dict[str, str]]) -> str:
@webhook_view("Lidarr", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_lidarr_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = get_body_for_http_request(payload)
topic = get_topic_for_http_request(payload)

13
zerver/webhooks/linear/view.py
View File

@ -4,10 +4,10 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_string, to_wild_value
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import OptionalUserSpecifiedTopicStr, check_send_webhook_message
from zerver.models import UserProfile
CONTENT_MESSAGE_TEMPLATE = "\n~~~ quote\n{message}\n~~~\n"
@ -111,12 +111,13 @@ ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())
@webhook_view("Linear", notify_bot_owner_on_invalid_json=True, all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_linear_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
payload: WebhookPayload[WildValue],
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
event_type = get_event_type(payload)
if event_type is None:

9
zerver/webhooks/mention/view.py
View File

@ -2,19 +2,20 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Mention")
@has_request_variables
@typed_endpoint
def api_mention_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
title = payload["title"].tame(check_string)
source_url = payload["url"].tame(check_string)

9
zerver/webhooks/netlify/view.py
View File

@ -4,9 +4,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import (
check_send_webhook_message,
get_http_headers_from_filename,
@ -26,11 +26,12 @@ fixture_to_headers = get_http_headers_from_filename("HTTP_X_NETLIFY_EVENT")
@webhook_view("Netlify", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_netlify_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
message_template, event = get_template(request, payload)

8
zerver/webhooks/newrelic/view.py
View File

@ -4,8 +4,8 @@ from django.utils.translation import gettext as _
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import (
WildValue,
check_int,
@ -13,7 +13,6 @@ from zerver.lib.validator import (
check_none_or,
check_string,
check_union,
to_wild_value,
)
from zerver.lib.webhooks.common import check_send_webhook_message, unix_milliseconds_to_timestamp
from zerver.models import UserProfile
@ -51,11 +50,12 @@ ALL_EVENT_TYPES = list(set(OLD_EVENT_TYPES).union(set(NEW_EVENT_TYPES)))
@webhook_view("NewRelic", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_newrelic_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# Handle old format
# Once old is EOLed, delete if block and keep else block

16
zerver/webhooks/opbeat/view.py
View File

@ -4,16 +4,9 @@ from typing import 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_union,
to_wild_value,
)
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string, check_union
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -108,11 +101,12 @@ def format_object(
@webhook_view("Opbeat")
@has_request_variables
@typed_endpoint
def api_opbeat_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
"""
This uses the subject name from opbeat to make the topic,

9
zerver/webhooks/opencollective/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -12,11 +12,12 @@ AMOUNT_TEMPLATE = "{amount}"
@webhook_view("OpenCollective")
@has_request_variables
@typed_endpoint
def api_opencollective_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
name = get_name(payload)
amount = get_amount(payload)

9
zerver/webhooks/opsgenie/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -26,11 +26,12 @@ ALL_EVENT_TYPES = [
@webhook_view("Opsgenie", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_opsgenie_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# construct the body of the message
info = {

9
zerver/webhooks/pagerduty/view.py
View File

@ -6,9 +6,9 @@ from typing_extensions import TypeAlias
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -231,11 +231,12 @@ def send_formatted_pagerduty(
@webhook_view("PagerDuty", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_pagerduty_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
messages = payload.get("messages")
if messages:

9
zerver/webhooks/pingdom/view.py
View File

@ -3,9 +3,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -39,11 +39,12 @@ ALL_EVENT_TYPES = list(SUPPORTED_CHECK_TYPES)
@webhook_view("Pingdom", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_pingdom_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
check_type = get_check_type(payload)

9
zerver/webhooks/radarr/view.py
View File

@ -2,9 +2,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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, check_string_in, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_string, check_string_in
from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message
from zerver.models import UserProfile
@ -45,11 +45,12 @@ ALL_EVENT_TYPES = [
@webhook_view("Radarr", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_radarr_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = get_body_for_http_request(payload)
topic = get_topic_for_http_request(payload)

16
zerver/webhooks/raygun/view.py
View File

@ -4,16 +4,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import (
WildValue,
check_anything,
check_int,
check_list,
check_string,
to_wild_value,
)
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_anything, check_int, check_list, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -21,11 +14,12 @@ ALL_EVENT_TYPES = ["error_notification", "error_activity"]
@webhook_view("Raygun", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_raygun_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# The payload contains 'event' key. This 'event' key has a value of either
# 'error_notification' or 'error_activity'. 'error_notification' happens

9
zerver/webhooks/reviewboard/view.py
View File

@ -2,9 +2,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import (
check_send_webhook_message,
get_http_headers_from_filename,
@ -177,11 +177,12 @@ ALL_EVENT_TYPES = list(RB_MESSAGE_FUNCTIONS.keys())
@webhook_view("ReviewBoard", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_reviewboard_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event_type = validate_extract_webhook_http_header(
request, "X-ReviewBoard-Event", "Review Board"

11
zerver/webhooks/rhodecode/view.py
View File

@ -5,9 +5,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import TOPIC_WITH_BRANCH_TEMPLATE, get_push_commits_event_message
from zerver.models import UserProfile
@ -76,12 +76,13 @@ ALL_EVENT_TYPES = list(EVENT_FUNCTION_MAPPER.keys())
@webhook_view("RhodeCode", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_rhodecode_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
branches: Optional[str] = REQ(default=None),
*,
payload: WebhookPayload[WildValue],
branches: Optional[str] = None,
) -> HttpResponse:
event = get_event_name(payload, branches)
if event is None:

9
zerver/webhooks/rundeck/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -12,11 +12,12 @@ RUNDECK_TOPIC_TEMPLATE = "{job_name}"
@webhook_view("Rundeck")
@has_request_variables
@typed_endpoint
def api_rundeck_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
topic = get_topic(payload)
body = get_body(payload)

9
zerver/webhooks/semaphore/view.py
View File

@ -5,9 +5,9 @@ from urllib.parse import urlparse
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import get_short_sha
from zerver.models import UserProfile
@ -92,11 +92,12 @@ ALL_EVENT_TYPES = ["build", "tag", "unknown", "branch", "deploy", "pull_request"
@webhook_view("Semaphore", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_semaphore_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
content, project_name, branch_name, event = (
semaphore_classic(payload) if "event" in payload else semaphore_2(payload)

12
zerver/webhooks/slack_incoming/view.py
View File

@ -1,15 +1,16 @@
# Webhooks for external integrations.
import re
from itertools import zip_longest
from typing import List, Literal, Optional, TypedDict, cast
from typing import List, Literal, TypedDict, cast
from django.http import HttpRequest, HttpResponse
from django.utils.translation import gettext as _
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, RequestVariableMissingError, has_request_variables
from zerver.lib.request import RequestVariableMissingError
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import typed_endpoint
from zerver.lib.types import Validator
from zerver.lib.validator import (
WildValue,
@ -21,16 +22,17 @@ from zerver.lib.validator import (
check_url,
to_wild_value,
)
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.common import OptionalUserSpecifiedTopicStr, check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("SlackIncoming")
@has_request_variables
@typed_endpoint
def api_slack_incoming_webhook(
request: HttpRequest,
user_profile: UserProfile,
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
# Slack accepts webhook payloads as payload="encoded json" as
# application/x-www-form-urlencoded, as well as in the body as

9
zerver/webhooks/solano/view.py
View File

@ -3,9 +3,9 @@ from django.core.exceptions import ValidationError
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.webhooks.git import get_short_sha
from zerver.models import UserProfile
@ -21,11 +21,12 @@ ALL_EVENT_TYPES = ["first-fail", "stop", "received"]
@webhook_view("SolanoLabs", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_solano_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event = payload["event"].tame(check_string)
topic = "build update"

9
zerver/webhooks/sonarqube/view.py
View File

@ -2,9 +2,9 @@
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -109,11 +109,12 @@ def render_body_without_branch(payload: WildValue) -> str:
@webhook_view("Sonarqube")
@has_request_variables
@typed_endpoint
def api_sonarqube_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
project = payload["project"]["name"].tame(check_string)
branch = None

9
zerver/webhooks/sonarr/view.py
View File

@ -2,9 +2,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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_int, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_string
from zerver.lib.webhooks.common import check_send_webhook_message, get_setup_webhook_message
from zerver.models import UserProfile
@ -39,11 +39,12 @@ ALL_EVENT_TYPES = [
@webhook_view("Sonarr", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_sonarr_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
body = get_body_for_http_request(payload)
topic = get_topic_for_http_request(payload)

9
zerver/webhooks/splunk/view.py
View File

@ -2,9 +2,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import MAX_TOPIC_NAME_LENGTH, UserProfile
@ -18,11 +18,12 @@ Splunk alert from saved search:
@webhook_view("Splunk")
@has_request_variables
@typed_endpoint
def api_splunk_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
# use default values if expected data is not provided
search_name = payload.get("search_name", "Missing search_name").tame(check_string)

9
zerver/webhooks/statuspage/view.py
View File

@ -3,9 +3,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -53,11 +53,12 @@ def get_component_topic(payload: WildValue) -> str:
@webhook_view("Statuspage", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_statuspage_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
if "incident" in payload:
event = "incident"

18
zerver/webhooks/stripe/view.py
View File

@ -6,17 +6,10 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.timestamp import timestamp_to_datetime
from zerver.lib.validator import (
WildValue,
check_bool,
check_int,
check_none_or,
check_string,
to_wild_value,
)
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -54,12 +47,13 @@ ALL_EVENT_TYPES = [
@webhook_view("Stripe", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_stripe_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
stream: str = REQ(default="test"),
*,
payload: WebhookPayload[WildValue],
stream: str = "test",
) -> HttpResponse:
try:
topic, body = topic_and_body(payload)

9
zerver/webhooks/taiga/view.py
View File

@ -12,9 +12,9 @@ from django.http import HttpRequest, HttpResponse
from typing_extensions import TypeAlias
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_bool, check_none_or, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_bool, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -23,11 +23,12 @@ ReturnType: TypeAlias = Tuple[WildValue, WildValue]
@webhook_view("Taiga")
@has_request_variables
@typed_endpoint
def api_taiga_webhook(
request: HttpRequest,
user_profile: UserProfile,
message: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
message: WebhookPayload[WildValue],
) -> HttpResponse:
parsed_events = parse_message(message)
content = "".join(sorted(generate_content(event) + "\n" for event in parsed_events))

10
zerver/webhooks/teamcity/view.py
View File

@ -10,10 +10,11 @@ from zerver.actions.message_send import (
send_rate_limited_pm_notification_to_bot_owner,
)
from zerver.decorator import webhook_view
from zerver.lib.request import REQ, RequestNotes, has_request_variables
from zerver.lib.request import RequestNotes
from zerver.lib.response import json_success
from zerver.lib.send_email import FromAddress
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import Realm, UserProfile
@ -50,11 +51,12 @@ def get_teamcity_property_value(property_list: WildValue, name: str) -> Optional
@webhook_view("TeamCity")
@has_request_variables
@typed_endpoint
def api_teamcity_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
if "build" not in payload:
# Ignore third-party specific (e.g. Slack) payload formats

13
zerver/webhooks/thinkst/view.py
View File

@ -4,10 +4,10 @@ from typing import Optional, Tuple
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_string, check_union, to_wild_value
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_string, check_union
from zerver.lib.webhooks.common import OptionalUserSpecifiedTopicStr, check_send_webhook_message
from zerver.models import UserProfile
@ -107,12 +107,13 @@ def body(message: WildValue) -> str:
@webhook_view("Thinkst")
@has_request_variables
@typed_endpoint
def api_thinkst_webhook(
request: HttpRequest,
user_profile: UserProfile,
message: WildValue = REQ(argument_type="body", converter=to_wild_value),
user_specified_topic: Optional[str] = REQ("topic", default=None),
*,
message: WebhookPayload[WildValue],
user_specified_topic: OptionalUserSpecifiedTopicStr = None,
) -> HttpResponse:
"""
Construct a response to a webhook event from a Thinkst canary or canarytoken.

9
zerver/webhooks/trello/view/__init__.py
View File

@ -5,9 +5,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import return_success_on_head_request, webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -17,11 +17,12 @@ from .card_actions import IGNORED_CARD_ACTIONS, SUPPORTED_CARD_ACTIONS, process_
@webhook_view("Trello")
@return_success_on_head_request
@has_request_variables
@typed_endpoint
def api_trello_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
action_type = payload["action"]["type"].tame(check_string)
message = get_topic_and_body(payload, action_type)

9
zerver/webhooks/updown/view.py
View File

@ -5,9 +5,9 @@ from django.http import HttpRequest, HttpResponse
from zerver.decorator import webhook_view
from zerver.lib.exceptions import UnsupportedWebhookEventTypeError
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, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_int, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -75,11 +75,12 @@ ALL_EVENT_TYPES = list(EVENT_TYPE_BODY_MAPPER.keys())
@webhook_view("Updown", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_updown_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
for event in payload:
send_message_for_event(request, user_profile, event)

9
zerver/webhooks/uptimerobot/view.py
View File

@ -5,10 +5,10 @@ from django.utils.translation import gettext as _
from zerver.actions.message_send import send_rate_limited_pm_notification_to_bot_owner
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.send_email import FromAddress
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -32,11 +32,12 @@ ALL_EVENT_TYPES = ["up", "down"]
@webhook_view("UptimeRobot", all_event_types=ALL_EVENT_TYPES)
@has_request_variables
@typed_endpoint
def api_uptimerobot_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
event_type = payload["alert_type_friendly_name"].tame(check_string)
if event_type == "Up":

9
zerver/webhooks/wekan/view.py
View File

@ -1,9 +1,9 @@
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_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -37,11 +37,12 @@ def process_message_data(payload: WildValue) -> str:
@webhook_view("Wekan")
@has_request_variables
@typed_endpoint
def api_wekan_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
topic = "Wekan Notification"
body = get_message_body(payload)

9
zerver/webhooks/zabbix/view.py
View File

@ -5,10 +5,10 @@ from django.utils.translation import gettext as _
from zerver.actions.message_send import send_rate_limited_pm_notification_to_bot_owner
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.send_email import FromAddress
from zerver.lib.validator import WildValue, check_string, to_wild_value
from zerver.lib.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -30,11 +30,12 @@ ZABBIX_MESSAGE_TEMPLATE = """
@webhook_view("Zabbix")
@has_request_variables
@typed_endpoint
def api_zabbix_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
try:
body = get_body_for_http_request(payload)

9
zerver/webhooks/zapier/view.py
View File

@ -3,19 +3,20 @@ from django.utils.translation import gettext as _
from zerver.decorator import webhook_view
from zerver.lib.exceptions import JsonableError
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.typed_endpoint import WebhookPayload, typed_endpoint
from zerver.lib.validator import WildValue, check_none_or, check_string
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@webhook_view("Zapier", notify_bot_owner_on_invalid_json=False)
@has_request_variables
@typed_endpoint
def api_zapier_webhook(
request: HttpRequest,
user_profile: UserProfile,
payload: WildValue = REQ(argument_type="body", converter=to_wild_value),
*,
payload: WebhookPayload[WildValue],
) -> HttpResponse:
if payload.get("type").tame(check_none_or(check_string)) == "auth":
# The bot's details are used by our Zapier app to format a connection

11
zerver/webhooks/zendesk/view.py
View File

@ -2,8 +2,8 @@
from django.http import HttpRequest, HttpResponse
from zerver.decorator import authenticated_rest_api_view
from zerver.lib.request import REQ, has_request_variables
from zerver.lib.response import json_success
from zerver.lib.typed_endpoint import typed_endpoint
from zerver.lib.webhooks.common import check_send_webhook_message
from zerver.models import UserProfile
@ -15,13 +15,14 @@ def truncate(string: str, length: int) -> str:
@authenticated_rest_api_view(webhook_client_name="Zendesk")
@has_request_variables
@typed_endpoint
def api_zendesk_webhook(
request: HttpRequest,
user_profile: UserProfile,
ticket_title: str = REQ(),
ticket_id: str = REQ(),
message: str = REQ(),
*,
ticket_title: str,
ticket_id: str,
message: str,
) -> HttpResponse:
"""
Zendesk uses triggers with message templates. This webhook uses the