codekingpro/portable-devtools

1import re
2from typing import TYPE_CHECKING, Tuple, Type, TypeVar
3
4from .typedefs import Handler, Middleware
5from .web_exceptions import HTTPMove, HTTPPermanentRedirect
6from .web_request import Request
7from .web_response import StreamResponse
8from .web_urldispatcher import SystemRoute
9
10__all__ = (
11    "middleware",
12    "normalize_path_middleware",
13)
14
15if TYPE_CHECKING:
16    from .web_app import Application
17
18_Func = TypeVar("_Func")
19
20
21async def _check_request_resolves(request: Request, path: str) -> Tuple[bool, Request]:
22    alt_request = request.clone(rel_url=path)
23
24    match_info = await request.app.router.resolve(alt_request)
25    alt_request._match_info = match_info
26
27    if match_info.http_exception is None:
28        return True, alt_request
29
30    return False, request
31
32
33def middleware(f: _Func) -> _Func:
34    f.__middleware_version__ = 1  # type: ignore[attr-defined]
35    return f
36
37
38def normalize_path_middleware(
39    *,
40    append_slash: bool = True,
41    remove_slash: bool = False,
42    merge_slashes: bool = True,
43    redirect_class: Type[HTTPMove] = HTTPPermanentRedirect,
44) -> Middleware:
45    """Factory for producing a middleware that normalizes the path of a request.
46
47    Normalizing means:
48        - Add or remove a trailing slash to the path.
49        - Double slashes are replaced by one.
50
51    The middleware returns as soon as it finds a path that resolves
52    correctly. The order if both merge and append/remove are enabled is
53        1) merge slashes
54        2) append/remove slash
55        3) both merge slashes and append/remove slash.
56    If the path resolves with at least one of those conditions, it will
57    redirect to the new path.
58
59    Only one of `append_slash` and `remove_slash` can be enabled. If both
60    are `True` the factory will raise an assertion error
61
62    If `append_slash` is `True` the middleware will append a slash when
63    needed. If a resource is defined with trailing slash and the request
64    comes without it, it will append it automatically.
65
66    If `remove_slash` is `True`, `append_slash` must be `False`. When enabled
67    the middleware will remove trailing slashes and redirect if the resource
68    is defined
69
70    If merge_slashes is True, merge multiple consecutive slashes in the
71    path into one.
72    """
73    correct_configuration = not (append_slash and remove_slash)
74    assert correct_configuration, "Cannot both remove and append slash"
75
76    @middleware
77    async def impl(request: Request, handler: Handler) -> StreamResponse:
78        if isinstance(request.match_info.route, SystemRoute):
79            paths_to_check = []
80            if "?" in request.raw_path:
81                path, query = request.raw_path.split("?", 1)
82                query = "?" + query
83            else:
84                query = ""
85                path = request.raw_path
86
87            if merge_slashes:
88                paths_to_check.append(re.sub("//+", "/", path))
89            if append_slash and not request.path.endswith("/"):
90                paths_to_check.append(path + "/")
91            if remove_slash and request.path.endswith("/"):
92                paths_to_check.append(path[:-1])
93            if merge_slashes and append_slash:
94                paths_to_check.append(re.sub("//+", "/", path + "/"))
95            if merge_slashes and remove_slash:
96                merged_slashes = re.sub("//+", "/", path)
97                paths_to_check.append(merged_slashes[:-1])
98
99            for path in paths_to_check:
100                path = re.sub("^//+", "/", path)  # SECURITY: GHSA-v6wp-4m6f-gcjg
101                resolves, request = await _check_request_resolves(request, path)
102                if resolves:
103                    raise redirect_class(request.raw_path + query)
104
105        return await handler(request)
106
107    return impl
108
109
110def _fix_request_current_app(app: "Application") -> Middleware:
111    @middleware
112    async def impl(request: Request, handler: Handler) -> StreamResponse:
113        match_info = request.match_info
114        prev = match_info.current_app
115        match_info.current_app = app
116        try:
117            return await handler(request)
118        finally:
119            match_info.current_app = prev
120
121    return impl
122