Coverage for gco / services / request_size_middleware.py: 100.00%

1"""Shared request-body size enforcement for GCO FastAPI services.""" 

2 

3from __future__ import annotations 

4 

5from collections import deque 

6 

7from starlette.responses import JSONResponse 

8from starlette.types import ASGIApp, Message, Receive, Scope, Send 

9 

10# <pyflowchart-code-diagram> BEGIN - auto-inserted, do not edit 

11# Generated at (UTC): 2026-09-01T14:42:56Z 

12# Generated from Git commit: 89b000378ed5a912a38c06f4feab2b029936ebcc 

13# Flowchart(s) generated from this file: 

14# * ``RequestSizeLimitMiddleware.__call__`` -> ``diagrams/code_diagrams/gco/services/request_size_middleware.RequestSizeLimitMiddleware___call__.html`` 

15# (PNG: ``diagrams/code_diagrams/gco/services/request_size_middleware.RequestSizeLimitMiddleware___call__.png``) 

16# Regenerate with ``SOURCE_DATE_EPOCH=<unix-seconds> GCO_DIAGRAM_SOURCE_COMMIT=<40-char-sha> python diagrams/generate.py --code-only``. 

17# <pyflowchart-code-diagram> END 

18 

19 

20DEFAULT_MAX_REQUEST_BODY_BYTES = 1_048_576 

21 

22 

23class RequestSizeLimitMiddleware: 

24 """Reject request bodies larger than the configured byte limit. 

25 

26 ``Content-Length`` is only an early-rejection optimization. Every accepted 

27 request is read from the ASGI receive channel and counted before it reaches 

28 authentication or a route handler, so a missing, malformed, negative, or 

29 deliberately under-reported header cannot bypass the limit. Buffered 

30 messages are replayed unchanged to preserve the exact bytes used by HMAC 

31 validation and downstream parsing. 

32 """ 

33 

34 def __init__(self, app: ASGIApp, max_body_bytes: int = DEFAULT_MAX_REQUEST_BODY_BYTES) -> None: 

35 if max_body_bytes < 0: 

36 raise ValueError("max_body_bytes must be non-negative") 

37 self.app = app 

38 self.max_body_bytes = max_body_bytes 

39 

40 async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None: 

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

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

43 return 

44 

45 if self._declared_size_exceeds_limit(scope): 

46 await self._too_large_response()(scope, receive, send) 

47 return 

48 

49 buffered_messages: deque[Message] = deque() 

50 received_bytes = 0 

51 while True: 

52 message = await receive() 

53 message_type = message["type"] 

54 

55 if message_type == "http.request": 

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

57 if received_bytes > self.max_body_bytes: 

58 await self._too_large_response()(scope, receive, send) 

59 return 

60 buffered_messages.append(message) 

61 if not message.get("more_body", False): 

62 break 

63 else: 

64 buffered_messages.append(message) 

65 if message_type == "http.disconnect": 

66 break 

67 

68 async def replay_receive() -> Message: 

69 if buffered_messages: 

70 return buffered_messages.popleft() 

71 return await receive() 

72 

73 await self.app(scope, replay_receive, send) 

74 

75 def _declared_size_exceeds_limit(self, scope: Scope) -> bool: 

76 """Return true if any valid Content-Length value is already too large.""" 

77 for name, raw_value in scope.get("headers", []): 

78 if name.lower() != b"content-length": 

79 continue 

80 try: 

81 declared_size = int(raw_value.decode("ascii")) 

82 except UnicodeDecodeError, ValueError: 

83 continue 

84 if declared_size > self.max_body_bytes: 

85 return True 

86 return False 

87 

88 def _too_large_response(self) -> JSONResponse: 

89 return JSONResponse( 

90 status_code=413, 

91 content={"detail": f"Request body exceeds maximum size of {self.max_body_bytes} bytes"}, 

92 )