push-notifications: Migrate to typed endpoint.

This commit is contained in:
Kenneth Rodrigues 2024-04-20 15:17:33 +05:30 committed by Tim Abbott
parent d03659ee1c
commit 081119f461
3 changed files with 16 additions and 26 deletions

View File

@ -108,6 +108,7 @@ OptionalTopic: TypeAlias = Annotated[
StringConstraints(strip_whitespace=True),
ApiParamConfig(whence="topic", aliases=("subject",)),
]
ApnsAppId: TypeAlias = Annotated[str, StringConstraints(pattern="^[.a-zA-Z0-9-]+$")]
# Reusable annotation metadata for Annotated types

View File

@ -1180,7 +1180,7 @@ class PushBouncerNotificationTest(BouncerTestCase):
result = self.client_post(
endpoint, {"token": token, "appid": "'; tables --"}, subdomain="zulip"
)
self.assert_json_error(result, "Invalid app ID")
self.assert_json_error(result, "appid has invalid format")
# Try to remove a non-existent token...
result = self.client_delete(endpoint, {"token": "abcd1234"}, subdomain="zulip")
@ -4618,7 +4618,7 @@ class TestPushApi(BouncerTestCase):
self.assert_json_error(result, "Missing 'appid' argument")
result = self.client_post(endpoint, {"token": label, "appid": "'; tables --"})
self.assert_json_error(result, "Invalid app ID")
self.assert_json_error(result, "appid has invalid format")
# Try to remove a non-existent token...
result = self.client_delete(endpoint, {"token": "abcd1234"})

View File

@ -1,4 +1,3 @@
import re
from typing import Optional
from django.conf import settings
@ -29,23 +28,12 @@ from zerver.lib.remote_server import (
send_server_data_to_push_bouncer,
send_to_push_bouncer,
)
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.validator import check_string
from zerver.lib.typed_endpoint import ApnsAppId, typed_endpoint
from zerver.models import PushDeviceToken, UserProfile
from zerver.views.errors import config_error
def check_app_id(var_name: str, val: object) -> str:
# Garbage values should be harmless, but we can be picky
# as insurance against bugs somewhere.
s = check_string(var_name, val)
if not re.fullmatch("[.a-zA-Z0-9-]+", s):
raise JsonableError(_("Invalid app ID"))
return s
def validate_token(token_str: str, kind: int) -> None:
if token_str == "" or len(token_str) > 4096:
raise JsonableError(_("Empty or invalid length token"))
@ -58,12 +46,13 @@ def validate_token(token_str: str, kind: int) -> None:
@human_users_only
@has_request_variables
@typed_endpoint
def add_apns_device_token(
request: HttpRequest,
user_profile: UserProfile,
token: str = REQ(),
appid: str = REQ(str_validator=check_app_id),
*,
token: str,
appid: ApnsAppId,
) -> HttpResponse:
validate_token(token, PushDeviceToken.APNS)
add_push_device_token(user_profile, token, PushDeviceToken.APNS, ios_app_id=appid)
@ -71,9 +60,9 @@ def add_apns_device_token(
@human_users_only
@has_request_variables
@typed_endpoint
def add_android_reg_id(
request: HttpRequest, user_profile: UserProfile, token: str = REQ()
request: HttpRequest, user_profile: UserProfile, *, token: str
) -> HttpResponse:
validate_token(token, PushDeviceToken.GCM)
add_push_device_token(user_profile, token, PushDeviceToken.GCM)
@ -81,9 +70,9 @@ def add_android_reg_id(
@human_users_only
@has_request_variables
@typed_endpoint
def remove_apns_device_token(
request: HttpRequest, user_profile: UserProfile, token: str = REQ()
request: HttpRequest, user_profile: UserProfile, *, token: str
) -> HttpResponse:
validate_token(token, PushDeviceToken.APNS)
remove_push_device_token(user_profile, token, PushDeviceToken.APNS)
@ -91,9 +80,9 @@ def remove_apns_device_token(
@human_users_only
@has_request_variables
@typed_endpoint
def remove_android_reg_id(
request: HttpRequest, user_profile: UserProfile, token: str = REQ()
request: HttpRequest, user_profile: UserProfile, *, token: str
) -> HttpResponse:
validate_token(token, PushDeviceToken.GCM)
remove_push_device_token(user_profile, token, PushDeviceToken.GCM)
@ -101,9 +90,9 @@ def remove_android_reg_id(
@human_users_only
@has_request_variables
@typed_endpoint
def send_test_push_notification_api(
request: HttpRequest, user_profile: UserProfile, token: Optional[str] = REQ(default=None)
request: HttpRequest, user_profile: UserProfile, *, token: Optional[str] = None
) -> HttpResponse:
# If a token is specified in the request, the test notification is supposed to be sent
# to that device. If no token is provided, the test notification should be sent to