mirror of https://github.com/zulip/zulip.git
102 lines
3.4 KiB
Python
102 lines
3.4 KiB
Python
from functools import wraps
|
|
from typing import Any, Callable
|
|
|
|
from django.http import HttpRequest, HttpResponse
|
|
from django.urls import path
|
|
from django.urls.resolvers import URLPattern
|
|
from django.utils.crypto import constant_time_compare
|
|
from django.utils.translation import gettext as _
|
|
from typing_extensions import Concatenate, ParamSpec
|
|
|
|
from zerver.decorator import get_basic_credentials, process_client
|
|
from zerver.lib.exceptions import (
|
|
ErrorCode,
|
|
JsonableError,
|
|
RemoteServerDeactivatedError,
|
|
UnauthorizedError,
|
|
)
|
|
from zerver.lib.rate_limiter import rate_limit
|
|
from zerver.lib.request import RequestNotes
|
|
from zerver.lib.rest import get_target_view_function_or_response
|
|
from zerver.lib.subdomains import get_subdomain
|
|
from zerver.models import Realm
|
|
from zilencer.models import RemoteZulipServer, get_remote_server_by_uuid
|
|
|
|
ParamT = ParamSpec("ParamT")
|
|
|
|
|
|
class InvalidZulipServerError(JsonableError):
|
|
code = ErrorCode.INVALID_ZULIP_SERVER
|
|
data_fields = ["role"]
|
|
|
|
def __init__(self, role: str) -> None:
|
|
self.role: str = role
|
|
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return "Zulip server auth failure: {role} is not registered -- did you run `manage.py register_server`?"
|
|
|
|
|
|
class InvalidZulipServerKeyError(InvalidZulipServerError):
|
|
@staticmethod
|
|
def msg_format() -> str:
|
|
return "Zulip server auth failure: key does not match role {role}"
|
|
|
|
|
|
def validate_remote_server(
|
|
request: HttpRequest,
|
|
role: str,
|
|
api_key: str,
|
|
) -> RemoteZulipServer:
|
|
try:
|
|
remote_server = get_remote_server_by_uuid(role)
|
|
except RemoteZulipServer.DoesNotExist:
|
|
raise InvalidZulipServerError(role)
|
|
if not constant_time_compare(api_key, remote_server.api_key):
|
|
raise InvalidZulipServerKeyError(role)
|
|
|
|
if remote_server.deactivated:
|
|
raise RemoteServerDeactivatedError()
|
|
|
|
if get_subdomain(request) != Realm.SUBDOMAIN_FOR_ROOT_DOMAIN:
|
|
raise JsonableError(_("Invalid subdomain for push notifications bouncer"))
|
|
RequestNotes.get_notes(request).remote_server = remote_server
|
|
process_client(request)
|
|
return remote_server
|
|
|
|
|
|
def authenticated_remote_server_view(
|
|
view_func: Callable[Concatenate[HttpRequest, RemoteZulipServer, ParamT], HttpResponse]
|
|
) -> Callable[Concatenate[HttpRequest, ParamT], HttpResponse]:
|
|
@wraps(view_func)
|
|
def _wrapped_view_func(
|
|
request: HttpRequest, /, *args: ParamT.args, **kwargs: ParamT.kwargs
|
|
) -> HttpResponse:
|
|
role, api_key = get_basic_credentials(request)
|
|
if "@" in role:
|
|
raise JsonableError(_("Must validate with valid Zulip server API key"))
|
|
try:
|
|
remote_server = validate_remote_server(request, role, api_key)
|
|
except JsonableError as e:
|
|
raise UnauthorizedError(e.msg)
|
|
|
|
rate_limit(request)
|
|
return view_func(request, remote_server, *args, **kwargs)
|
|
|
|
return _wrapped_view_func
|
|
|
|
|
|
def remote_server_dispatch(request: HttpRequest, **kwargs: Any) -> HttpResponse:
|
|
result = get_target_view_function_or_response(request, kwargs)
|
|
if isinstance(result, HttpResponse):
|
|
return result
|
|
target_function, view_flags = result
|
|
return authenticated_remote_server_view(target_function)(request, **kwargs)
|
|
|
|
|
|
def remote_server_path(
|
|
route: str,
|
|
**handlers: Callable[Concatenate[HttpRequest, RemoteZulipServer, ParamT], HttpResponse],
|
|
) -> URLPattern:
|
|
return path(route, remote_server_dispatch, handlers)
|