Coverage for src/lexigram/web/middleware/body_limit.py: 22%

1"""Request body size limit middleware. 

2 

3Rejects requests whose declared ``Content-Length`` exceeds a configurable 

4threshold before the body is read. This prevents an attacker from posting a 

5gigabyte-sized payload that would exhaust application memory. 

6 

7Usage:: 

8 

9 app.add_middleware(RequestBodySizeLimitMiddleware, max_body_size=10 * 1024 * 1024) 

10 

11Or via ``WebConfig.max_body_size`` — ``WebProvider`` adds this middleware 

12automatically when ``max_body_size`` is set on the config. 

13""" 

14 

15from __future__ import annotations 

16 

17from typing import Any 

18 

19from lexigram.logging import get_logger 

20 

21logger = get_logger(__name__) 

22 

23_10_MB = 10 * 1024 * 1024 # 10 MiB — used as default sentinel 

24 

25 

26class RequestBodySizeLimitMiddleware: 

27 """ASGI middleware that rejects oversized request bodies. 

28 

29 Checks the ``Content-Length`` header on every incoming HTTP request. If 

30 the declared size exceeds ``max_body_size`` a ``413 Request Entity Too 

31 Large`` response is returned immediately — the body is never read into 

32 memory. 

33 

34 Note: Two enforcement paths: the declared ``Content-Length`` is checked 

35 before the body is read (fast 413); a body without a declared length 

36 (chunked transfer-encoding) is counted as it streams and is rejected 

37 with 413 once the accumulated bytes exceed ``max_body_size``. 

38 

39 Args: 

40 app: The ASGI application to wrap. 

41 max_body_size: Maximum allowed body size in bytes. Defaults to 

42 10 MiB (10 × 1024 × 1024 bytes). 

43 """ 

44 

45 def __init__(self, app: Any, max_body_size: int = _10_MB) -> None: 

46 """Initialise the middleware. 

47 

48 Args: 

49 app: The inner ASGI application. 

50 max_body_size: Byte limit. Requests with a ``Content-Length`` 

51 header exceeding this value receive a 413 response. 

52 """ 

53 self.app = app 

54 self.max_body_size = max_body_size 

55 

56 async def __call__(self, scope: dict[str, Any], receive: Any, send: Any) -> None: 

57 """Enforce the body size limit on HTTP requests. 

58 

59 Args: 

60 scope: ASGI connection scope. 

61 receive: ASGI receive callable. 

62 send: ASGI send callable. 

63 

64 Note: 

65 Two enforcement paths: the declared ``Content-Length`` is checked 

66 before the body is read (fast 413); a body without a declared 

67 length (chunked transfer-encoding) is counted as it streams and 

68 is rejected with 413 once the accumulated bytes exceed 

69 ``max_body_size``. 

70 """ 

71 if scope["type"] != "http": 

72 await self.app(scope, receive, send) 

73 return 

74 

75 headers: dict[bytes, bytes] = { 

76 k.lower(): v for k, v in scope.get("headers", []) 

77 } 

78 content_length_bytes = headers.get(b"content-length") 

79 

80 if content_length_bytes is not None: 

81 try: 

82 content_length = int(content_length_bytes) 

83 except ValueError: 

84 content_length = 0 

85 

86 if content_length > self.max_body_size: 

87 logger.warning( 

88 "request_body_too_large", 

89 content_length=content_length, 

90 max_body_size=self.max_body_size, 

91 path=scope.get("path", ""), 

92 ) 

93 response = _413_response() 

94 await response(scope, receive, send) 

95 return 

96 

97 # Stream-count bodies without a declared length (chunked encoding). 

98 received_bytes = 0 

99 aborted = False 

100 

101 async def counting_receive() -> Any: 

102 nonlocal received_bytes, aborted 

103 message = await receive() 

104 if message["type"] == "http.request" and not aborted: 

105 received_bytes += len(message.get("body", b"")) 

106 if received_bytes > self.max_body_size: 

107 aborted = True 

108 logger.warning( 

109 "request_body_too_large_streamed", 

110 received_bytes=received_bytes, 

111 max_body_size=self.max_body_size, 

112 path=scope.get("path", ""), 

113 ) 

114 response = _413_response() 

115 await response(scope, receive, send) 

116 return { 

117 "type": "http.request", 

118 "body": b"", 

119 "more_body": False, 

120 } 

121 return message 

122 

123 await self.app(scope, counting_receive, send) 

124 

125 

126def _413_response() -> Any: 

127 """Return a minimal 413 ASGI response callable.""" 

128 

129 async def respond(scope: dict[str, Any], receive: Any, send: Any) -> None: 

130 await send( 

131 { 

132 "type": "http.response.start", 

133 "status": 413, 

134 "headers": [ 

135 [b"content-type", b"application/json"], 

136 [b"connection", b"close"], 

137 ], 

138 } 

139 ) 

140 await send( 

141 { 

142 "type": "http.response.body", 

143 "body": b'{"detail":"Request Entity Too Large"}', 

144 "more_body": False, 

145 } 

146 ) 

147 

148 return respond 

149 

150 

151__all__ = ["RequestBodySizeLimitMiddleware"]