123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121 |
- import re
- from typing import TYPE_CHECKING, Awaitable, Callable, Tuple, Type, TypeVar
- from .web_exceptions import HTTPPermanentRedirect, _HTTPMove
- from .web_request import Request
- from .web_response import StreamResponse
- from .web_urldispatcher import SystemRoute
- __all__ = (
- "middleware",
- "normalize_path_middleware",
- )
- if TYPE_CHECKING: # pragma: no cover
- from .web_app import Application
- _Func = TypeVar("_Func")
- async def _check_request_resolves(request: Request, path: str) -> Tuple[bool, Request]:
- alt_request = request.clone(rel_url=path)
- match_info = await request.app.router.resolve(alt_request)
- alt_request._match_info = match_info # type: ignore
- if match_info.http_exception is None:
- return True, alt_request
- return False, request
- def middleware(f: _Func) -> _Func:
- f.__middleware_version__ = 1 # type: ignore
- return f
- _Handler = Callable[[Request], Awaitable[StreamResponse]]
- _Middleware = Callable[[Request, _Handler], Awaitable[StreamResponse]]
- def normalize_path_middleware(
- *,
- append_slash: bool = True,
- remove_slash: bool = False,
- merge_slashes: bool = True,
- redirect_class: Type[_HTTPMove] = HTTPPermanentRedirect
- ) -> _Middleware:
- """
- Middleware factory which produces a middleware that normalizes
- the path of a request. By normalizing it means:
- - Add or remove a trailing slash to the path.
- - Double slashes are replaced by one.
- The middleware returns as soon as it finds a path that resolves
- correctly. The order if both merge and append/remove are enabled is
- 1) merge slashes
- 2) append/remove slash
- 3) both merge slashes and append/remove slash.
- If the path resolves with at least one of those conditions, it will
- redirect to the new path.
- Only one of `append_slash` and `remove_slash` can be enabled. If both
- are `True` the factory will raise an assertion error
- If `append_slash` is `True` the middleware will append a slash when
- needed. If a resource is defined with trailing slash and the request
- comes without it, it will append it automatically.
- If `remove_slash` is `True`, `append_slash` must be `False`. When enabled
- the middleware will remove trailing slashes and redirect if the resource
- is defined
- If merge_slashes is True, merge multiple consecutive slashes in the
- path into one.
- """
- correct_configuration = not (append_slash and remove_slash)
- assert correct_configuration, "Cannot both remove and append slash"
- @middleware
- async def impl(request: Request, handler: _Handler) -> StreamResponse:
- if isinstance(request.match_info.route, SystemRoute):
- paths_to_check = []
- if "?" in request.raw_path:
- path, query = request.raw_path.split("?", 1)
- query = "?" + query
- else:
- query = ""
- path = request.raw_path
- if merge_slashes:
- paths_to_check.append(re.sub("//+", "/", path))
- if append_slash and not request.path.endswith("/"):
- paths_to_check.append(path + "/")
- if remove_slash and request.path.endswith("/"):
- paths_to_check.append(path[:-1])
- if merge_slashes and append_slash:
- paths_to_check.append(re.sub("//+", "/", path + "/"))
- if merge_slashes and remove_slash:
- merged_slashes = re.sub("//+", "/", path)
- paths_to_check.append(merged_slashes[:-1])
- for path in paths_to_check:
- path = re.sub("^//+", "/", path) # SECURITY: GHSA-v6wp-4m6f-gcjg
- resolves, request = await _check_request_resolves(request, path)
- if resolves:
- raise redirect_class(request.raw_path + query)
- return await handler(request)
- return impl
- def _fix_request_current_app(app: "Application") -> _Middleware:
- @middleware
- async def impl(request: Request, handler: _Handler) -> StreamResponse:
- with request.match_info.set_current_app(app):
- return await handler(request)
- return impl
|