Coverage for src/local_deep_research/web/utils/range_response.py: 100%

1"""Byte-range responses for content already loaded into memory.""" 

2 

3from collections.abc import Mapping 

4import re 

5 

6from starlette.requests import Request 

7from starlette.responses import Response 

8 

9 

10def in_memory_range_response( 

11 content: bytes, 

12 *, 

13 request: Request, 

14 media_type: str, 

15 headers: Mapping[str, str] | None = None, 

16) -> Response: 

17 """Honor one GET byte range without writing the content to a file. 

18 

19 Unsupported or malformed ranges use the complete response. These downloads 

20 have no ETag or Last-Modified validator, so If-Range also selects the full 

21 response instead of risking a slice from a changed document. 

22 """ 

23 response = Response(content, media_type=media_type, headers=headers) 

24 response.headers["Accept-Ranges"] = "bytes" 

25 if request.method != "GET" or "if-range" in request.headers: 

26 return response 

27 

28 ranges = request.headers.getlist("range") 

29 # Do not partially honor a multi-range request or generate multipart data. 

30 # Ignore oversized numeric input before the parser converts it to integers. 

31 if len(ranges) != 1 or len(ranges[0]) > 1024 or "," in ranges[0]: 

32 return response 

33 unit, separator, value = ranges[0].partition("=") 

34 if not separator or unit.strip(" \t").lower() != "bytes": 

35 return response 

36 match = re.fullmatch(r"([0-9]*)[ \t]*-[ \t]*([0-9]*)", value.strip(" \t")) 

37 if match is None or not any(match.groups()): 

38 return response 

39 

40 length = len(content) 

41 first, last = match.groups() 

42 try: 

43 if not first: 

44 suffix_length = int(last) 

45 if suffix_length == 0: 

46 return response 

47 start, stop = max(0, length - suffix_length), length 

48 else: 

49 start = int(first) 

50 stop = int(last) + 1 if last else length 

51 if last and stop <= start: 

52 return response 

53 except ValueError: 

54 # Respect an interpreter's configured integer-conversion limit too. 

55 return response 

56 

57 if start >= length: 

58 response.status_code = 416 

59 response.body = b"" 

60 # nosemgrep: semgrep.rules.sql-string-concatenation, reason: HTTP Content-Range uses integer byte offsets and never reaches SQL 

61 response.headers["Content-Range"] = f"bytes */{length}" 

62 else: 

63 stop = min(stop, length) 

64 response.status_code = 206 

65 response.body = content[start:stop] 

66 # nosemgrep: semgrep.rules.sql-string-concatenation, reason: HTTP Content-Range uses integer byte offsets and never reaches SQL 

67 response.headers["Content-Range"] = f"bytes {start}-{stop - 1}/{length}" 

68 response.headers["Content-Length"] = str(len(response.body)) 

69 return response