17386711911

Committed 01 Sep 2025 08:32PM UTC coverage: 98.608% (+0.08%) from 98.527%

Build # 17386711911

Build Type

push

github

Committed by

alorence

Commit Message

[docs] Update security.rst

Run Details

76 of 76 branches covered (100.0%)

1275 of 1293 relevant lines covered (98.61%)

17.61 hits per line

Source File
Press 'n' to go to next uncovered line, 'b' for previous

98.78

modernrpc/server.py

from __future__ import annotations

import functools
import logging
from typing import TYPE_CHECKING, Callable, Union

from django.utils.module_loading import import_string
from django.views.decorators.csrf import csrf_exempt
from django.views.decorators.http import require_POST

from modernrpc import Protocol, RpcRequestContext
from modernrpc.compat import async_csrf_exempt, async_require_post
from modernrpc.config import settings
from modernrpc.constants import NOT_SET, SYSTEM_NAMESPACE_DOTTED_PATH
from modernrpc.core import ProcedureWrapper
from modernrpc.exceptions import RPCException, RPCInternalError, RPCMethodNotFound
from modernrpc.helpers import check_flags_compatibility, first_true
from modernrpc.views import handle_rpc_request, handle_rpc_request_async

if TYPE_CHECKING:
    from django.http import HttpRequest

    from modernrpc.handler import RpcHandler
    from modernrpc.types import AuthPredicateType


logger = logging.getLogger(__name__)

RpcErrorHandler = Callable[[BaseException, RpcRequestContext], Union[RPCException, None]]


class RegistryMixin:
    def __init__(self, auth: AuthPredicateType = NOT_SET) -> None:
        self._registry: dict[str, ProcedureWrapper] = {}
        self.auth = auth

    def register_procedure(
        self,
        procedure: Callable | None = None,
        name: str | None = None,
        protocol: Protocol = Protocol.ALL,
        auth: AuthPredicateType = NOT_SET,
        context_target: str | None = None,
    ) -> Callable:
        """
        Registers a procedure for handling RPC (Remote Procedure Call) requests. This function can be used as a
        decorator to register the given callable function as an RPC procedure.

        :param procedure: A callable function to be registered as an RPC procedure
        :param name: A custom name for the procedure. If not provided, the callable function's name is used by default.
        :param protocol: Specifies the protocol (e.g., JSON-RPC, XML-RPC) under which the procedure can be executed.
                         Defaults: Protocol.ALL
        :param auth: Defines user authentication settings or access rules for the procedure. Defaults to NOT_SET, in
                     which case the server's authentication settings will be used.
        :param context_target: Specify the procedure argument name for accessing the RPC request context.

        :raises ValueError: If a procedure can't be registered
        """

        def decorated(func: Callable) -> Callable:
            if name and name.startswith("rpc."):
                raise ValueError(
                    'According to JSON-RPC specs, method names starting with "rpc." are reserved for system extensions '
                    "and must not be used. See https://www.jsonrpc.org/specification#extensions for more information."
                )

            wrapper = ProcedureWrapper(
                func,
                name,
                protocol=protocol,
                auth=self.auth if auth == NOT_SET else auth,
                context_target=context_target,
            )

            if wrapper.name in self._registry and wrapper != self._registry[wrapper.name]:
                raise ValueError(f"Procedure {wrapper.name} is already registered")

            self._registry[wrapper.name] = wrapper
            logger.debug(f"Registered procedure {wrapper.name}")
            return func

        # If @server.register_procedure() is used with parenthesis (with or without argument)
        if procedure is None:
            return decorated

        # If @server.register_procedure is used without a parenthesis
        return decorated(procedure)

    @property
    def procedures(self) -> dict[str, ProcedureWrapper]:
        return self._registry


class RpcNamespace(RegistryMixin): ...


class RpcServer(RegistryMixin):
    def __init__(
        self,
        register_system_procedures: bool = True,
        supported_protocol: Protocol = Protocol.ALL,
        auth: AuthPredicateType = NOT_SET,
        error_handler: RpcErrorHandler | None = None,
    ) -> None:
        super().__init__(auth)
        self.request_handlers_classes = list(
            filter(
                lambda cls: check_flags_compatibility(cls.protocol, supported_protocol),
                (import_string(klass) for klass in settings.MODERNRPC_HANDLERS),
            )
        )

        if register_system_procedures:
            system_namespace = import_string(SYSTEM_NAMESPACE_DOTTED_PATH)
            self.register_namespace(system_namespace, "system")

        self.error_handler = error_handler

    def register_namespace(self, namespace: RpcNamespace, name: str | None = None) -> None:
        prefix = f"{name}." if name else ""
        logger.debug(f"About to register {len(namespace.procedures)} procedures into namespace '{prefix}'")

        for procedure_name, wrapper in namespace.procedures.items():
            self.register_procedure(
                wrapper.func_or_coro,
                name=f"{prefix}{procedure_name}",
                protocol=wrapper.protocol,
                auth=wrapper.auth,
                context_target=wrapper.context_target,
            )

    def get_procedure_wrapper(self, name: str, protocol: Protocol) -> ProcedureWrapper:
        """Return the procedure wrapper with given name compatible with given protocol, or raise RPCMethodNotFound"""
        try:
            wrapper = self.procedures[name]
        except KeyError:
            raise RPCMethodNotFound(name) from None

        if check_flags_compatibility(wrapper.protocol, protocol):
            return wrapper

        raise RPCMethodNotFound(name) from None

    def get_request_handler(self, request: HttpRequest) -> RpcHandler | None:
        klass = first_true(self.request_handlers_classes, pred=lambda cls: cls.can_handle(request))
        try:
            return klass()
        except TypeError:
            return None

    def on_error(self, exception: BaseException, context: RpcRequestContext) -> RPCException:
        """
        If an error handler is defined, call it to run arbitrary code and return its not-None result.
        Else, check the given exception type and return it if it is a subclass of RPCException.
        Convert any other exception into RPCInternalError.
        """
        if self.error_handler:
            try:
                self.error_handler(exception, context)
            except Exception as exc:
                exception = exc

        if isinstance(exception, RPCException):
            return exception
        return RPCInternalError(message=str(exception))

    @property
    def view(self) -> Callable:
        """
        Returns a synchronous view function that can be used in Django URL patterns.
        The view is decorated with csrf_exempt and require_POST.

        :return: A callable view function
        """
        view_func = functools.partial(handle_rpc_request, server=self)
        return csrf_exempt(require_POST(view_func))

    @property
    def async_view(self) -> Callable:
        """
        Returns an asynchronous view function that can be used in Django URL patterns.
        The view is decorated with csrf_exempt and require_POST.

        :return: An awaitable async view function
        """
        view_func = functools.partial(handle_rpc_request_async, server=self)
        return async_csrf_exempt(async_require_post(view_func))

1	from __future__ import annotations	18✔
2
3	import functools	18✔
4	import logging	18✔
5	from typing import TYPE_CHECKING, Callable, Union	18✔
6
7	from django.utils.module_loading import import_string	18✔
8	from django.views.decorators.csrf import csrf_exempt	18✔
9	from django.views.decorators.http import require_POST	18✔
10
11	from modernrpc import Protocol, RpcRequestContext	18✔
12	from modernrpc.compat import async_csrf_exempt, async_require_post	18✔
13	from modernrpc.config import settings	18✔
14	from modernrpc.constants import NOT_SET, SYSTEM_NAMESPACE_DOTTED_PATH	18✔
15	from modernrpc.core import ProcedureWrapper	18✔
16	from modernrpc.exceptions import RPCException, RPCInternalError, RPCMethodNotFound	18✔
17	from modernrpc.helpers import check_flags_compatibility, first_true	18✔
18	from modernrpc.views import handle_rpc_request, handle_rpc_request_async	18✔
19
20	if TYPE_CHECKING:
21	from django.http import HttpRequest
22
23	from modernrpc.handler import RpcHandler
24	from modernrpc.types import AuthPredicateType
25
26
27	logger = logging.getLogger(__name__)	18✔
28
29	RpcErrorHandler = Callable[[BaseException, RpcRequestContext], Union[RPCException, None]]	18✔
30
31
32	class RegistryMixin:	18✔
33	def __init__(self, auth: AuthPredicateType = NOT_SET) -> None:	18✔
34	self._registry: dict[str, ProcedureWrapper] = {}	18✔
35	self.auth = auth	18✔
36
37	def register_procedure(	18✔
38	self,
39	procedure: Callable \| None = None,
40	name: str \| None = None,
41	protocol: Protocol = Protocol.ALL,
42	auth: AuthPredicateType = NOT_SET,
43	context_target: str \| None = None,
44	) -> Callable:
45	"""
46	Registers a procedure for handling RPC (Remote Procedure Call) requests. This function can be used as a
47	decorator to register the given callable function as an RPC procedure.
48
49	:param procedure: A callable function to be registered as an RPC procedure
50	:param name: A custom name for the procedure. If not provided, the callable function's name is used by default.
51	:param protocol: Specifies the protocol (e.g., JSON-RPC, XML-RPC) under which the procedure can be executed.
52	Defaults: Protocol.ALL
53	:param auth: Defines user authentication settings or access rules for the procedure. Defaults to NOT_SET, in
54	which case the server's authentication settings will be used.
55	:param context_target: Specify the procedure argument name for accessing the RPC request context.
56
57	:raises ValueError: If a procedure can't be registered
58	"""
59
60	def decorated(func: Callable) -> Callable:	18✔
61	if name and name.startswith("rpc."):	18✔
62	raise ValueError(	18✔
63	'According to JSON-RPC specs, method names starting with "rpc." are reserved for system extensions '
64	"and must not be used. See https://www.jsonrpc.org/specification#extensions for more information."
65	)
66
67	wrapper = ProcedureWrapper(	18✔
68	func,
69	name,
70	protocol=protocol,
71	auth=self.auth if auth == NOT_SET else auth,
72	context_target=context_target,
73	)
74
75	if wrapper.name in self._registry and wrapper != self._registry[wrapper.name]:	18✔
76	raise ValueError(f"Procedure {wrapper.name} is already registered")	×
77
78	self._registry[wrapper.name] = wrapper	18✔
79	logger.debug(f"Registered procedure {wrapper.name}")	18✔
80	return func	18✔
81
82	# If @server.register_procedure() is used with parenthesis (with or without argument)
83	if procedure is None:	18✔
84	return decorated	18✔
85
86	# If @server.register_procedure is used without a parenthesis
87	return decorated(procedure)	18✔
88
89	@property	18✔
90	def procedures(self) -> dict[str, ProcedureWrapper]:	18✔
91	return self._registry	18✔
92
93
94	class RpcNamespace(RegistryMixin): ...	18✔
95
96
97	class RpcServer(RegistryMixin):	18✔
98	def __init__(	18✔
99	self,
100	register_system_procedures: bool = True,
101	supported_protocol: Protocol = Protocol.ALL,
102	auth: AuthPredicateType = NOT_SET,
103	error_handler: RpcErrorHandler \| None = None,
104	) -> None:
105	super().__init__(auth)	18✔
106	self.request_handlers_classes = list(	18✔
107	filter(
108	lambda cls: check_flags_compatibility(cls.protocol, supported_protocol),
109	(import_string(klass) for klass in settings.MODERNRPC_HANDLERS),
110	)
111	)
112
113	if register_system_procedures:	18✔
114	system_namespace = import_string(SYSTEM_NAMESPACE_DOTTED_PATH)	18✔
115	self.register_namespace(system_namespace, "system")	18✔
116
117	self.error_handler = error_handler	18✔
118
119	def register_namespace(self, namespace: RpcNamespace, name: str \| None = None) -> None:	18✔
120	prefix = f"{name}." if name else ""	18✔
121	logger.debug(f"About to register {len(namespace.procedures)} procedures into namespace '{prefix}'")	18✔
122
123	for procedure_name, wrapper in namespace.procedures.items():	18✔
124	self.register_procedure(	18✔
125	wrapper.func_or_coro,
126	name=f"{prefix}{procedure_name}",
127	protocol=wrapper.protocol,
128	auth=wrapper.auth,
129	context_target=wrapper.context_target,
130	)
131
132	def get_procedure_wrapper(self, name: str, protocol: Protocol) -> ProcedureWrapper:	18✔
133	"""Return the procedure wrapper with given name compatible with given protocol, or raise RPCMethodNotFound"""
134	try:	18✔
135	wrapper = self.procedures[name]	18✔
136	except KeyError:	18✔
137	raise RPCMethodNotFound(name) from None	18✔
138
139	if check_flags_compatibility(wrapper.protocol, protocol):	18✔
140	return wrapper	18✔
141
142	raise RPCMethodNotFound(name) from None	18✔
143
144	def get_request_handler(self, request: HttpRequest) -> RpcHandler \| None:	18✔
145	klass = first_true(self.request_handlers_classes, pred=lambda cls: cls.can_handle(request))	18✔
146	try:	18✔
147	return klass()	18✔
148	except TypeError:	18✔
149	return None	18✔
150
151	def on_error(self, exception: BaseException, context: RpcRequestContext) -> RPCException:	18✔
152	"""
153	If an error handler is defined, call it to run arbitrary code and return its not-None result.
154	Else, check the given exception type and return it if it is a subclass of RPCException.
155	Convert any other exception into RPCInternalError.
156	"""
157	if self.error_handler:	18✔
158	try:	18✔
159	self.error_handler(exception, context)	18✔
160	except Exception as exc:	18✔
161	exception = exc	18✔
162
163	if isinstance(exception, RPCException):	18✔
164	return exception	18✔
165	return RPCInternalError(message=str(exception))	18✔
166
167	@property	18✔
168	def view(self) -> Callable:	18✔
169	"""
170	Returns a synchronous view function that can be used in Django URL patterns.
171	The view is decorated with csrf_exempt and require_POST.
172
173	:return: A callable view function
174	"""
175	view_func = functools.partial(handle_rpc_request, server=self)	18✔
176	return csrf_exempt(require_POST(view_func))	18✔
177
178	@property	18✔
179	def async_view(self) -> Callable:	18✔
180	"""
181	Returns an asynchronous view function that can be used in Django URL patterns.
182	The view is decorated with csrf_exempt and require_POST.
183
184	:return: An awaitable async view function
185	"""
186	view_func = functools.partial(handle_rpc_request_async, server=self)	18✔
187	return async_csrf_exempt(async_require_post(view_func))	18✔

alorence / django-modern-rpc / 17386711911

Source File Press 'n' to go to next uncovered line, 'b' for previous

Source File
Press 'n' to go to next uncovered line, 'b' for previous