Coverage for /pythoncovmergedfiles/medio/medio/src/aiohttp/aiohttp/web_middlewares.py: 26%

1import re 

2import warnings 

3from typing import TYPE_CHECKING, Tuple, Type, TypeVar 

4 

5from .typedefs import Handler, Middleware 

6from .web_exceptions import HTTPMove, HTTPPermanentRedirect 

7from .web_request import Request 

8from .web_response import StreamResponse 

9from .web_urldispatcher import SystemRoute 

10 

11__all__ = ( 

12 "middleware", 

13 "normalize_path_middleware", 

14) 

15 

16if TYPE_CHECKING: # pragma: no cover 

17 from .web_app import Application 

18 

19_Func = TypeVar("_Func") 

20 

21 

22async def _check_request_resolves(request: Request, path: str) -> Tuple[bool, Request]: 

23 alt_request = request.clone(rel_url=path) 

24 

25 match_info = await request.app.router.resolve(alt_request) 

26 alt_request._match_info = match_info 

27 

28 if match_info.http_exception is None: 

29 return True, alt_request 

30 

31 return False, request 

32 

33 

34def middleware(f: _Func) -> _Func: 

35 warnings.warn( 

36 "Middleware decorator is deprecated since 4.0 " 

37 "and its behaviour is default, " 

38 "you can simply remove this decorator.", 

39 DeprecationWarning, 

40 stacklevel=2, 

41 ) 

42 return f 

43 

44 

45def normalize_path_middleware( 

46 *, 

47 append_slash: bool = True, 

48 remove_slash: bool = False, 

49 merge_slashes: bool = True, 

50 redirect_class: Type[HTTPMove] = HTTPPermanentRedirect, 

51) -> Middleware: 

52 """Factory for producing a middleware that normalizes the path of a request. 

53 

54 Normalizing means: 

55 - Add or remove a trailing slash to the path. 

56 - Double slashes are replaced by one. 

57 

58 The middleware returns as soon as it finds a path that resolves 

59 correctly. The order if both merge and append/remove are enabled is 

60 1) merge slashes 

61 2) append/remove slash 

62 3) both merge slashes and append/remove slash. 

63 If the path resolves with at least one of those conditions, it will 

64 redirect to the new path. 

65 

66 Only one of `append_slash` and `remove_slash` can be enabled. If both 

67 are `True` the factory will raise an assertion error 

68 

69 If `append_slash` is `True` the middleware will append a slash when 

70 needed. If a resource is defined with trailing slash and the request 

71 comes without it, it will append it automatically. 

72 

73 If `remove_slash` is `True`, `append_slash` must be `False`. When enabled 

74 the middleware will remove trailing slashes and redirect if the resource 

75 is defined 

76 

77 If merge_slashes is True, merge multiple consecutive slashes in the 

78 path into one. 

79 """ 

80 correct_configuration = not (append_slash and remove_slash) 

81 assert correct_configuration, "Cannot both remove and append slash" 

82 

83 async def impl(request: Request, handler: Handler) -> StreamResponse: 

84 if isinstance(request.match_info.route, SystemRoute): 

85 paths_to_check = [] 

86 if "?" in request.raw_path: 

87 path, query = request.raw_path.split("?", 1) 

88 query = "?" + query 

89 else: 

90 query = "" 

91 path = request.raw_path 

92 

93 if merge_slashes: 

94 paths_to_check.append(re.sub("//+", "/", path)) 

95 if append_slash and not request.path.endswith("/"): 

96 paths_to_check.append(path + "/") 

97 if remove_slash and request.path.endswith("/"): 

98 paths_to_check.append(path[:-1]) 

99 if merge_slashes and append_slash: 

100 paths_to_check.append(re.sub("//+", "/", path + "/")) 

101 if merge_slashes and remove_slash and path.endswith("/"): 

102 merged_slashes = re.sub("//+", "/", path) 

103 paths_to_check.append(merged_slashes[:-1]) 

104 

105 for path in paths_to_check: 

106 path = re.sub("^//+", "/", path) # SECURITY: GHSA-v6wp-4m6f-gcjg 

107 resolves, request = await _check_request_resolves(request, path) 

108 if resolves: 

109 raise redirect_class(request.raw_path + query) 

110 

111 return await handler(request) 

112 

113 return impl 

114 

115 

116def _fix_request_current_app(app: "Application") -> Middleware: 

117 async def impl(request: Request, handler: Handler) -> StreamResponse: 

118 with request.match_info.set_current_app(app): 

119 return await handler(request) 

120 

121 return impl