Coverage for .venv/lib/python3.13/site-packages/litellm/proxy/openai_files_endpoints/general_upload_validation.py: 32%

1""" 

2Upload validation applied to every purpose at POST /v1/files. 

3 

4batch_file_validation.py checks the JSONL shape of purpose="batch" uploads; this 

5module applies the same fast-fail-before-forwarding shape (size cap, allowed and 

6blocked extensions, path-traversal filenames) regardless of purpose. 

7""" 

8 

9from dataclasses import dataclass 

10from pathlib import Path 

11from typing import BinaryIO, Final, NoReturn 

12 

13from typing_extensions import assert_never 

14 

15from litellm.proxy._types import ProxyException 

16from litellm.proxy.common_utils.path_utils import safe_filename 

17 

18MB: Final = 1024 * 1024 

19 

20 

21def coerce_optional_int_setting(raw: object) -> int | None: 

22 """A general_settings value declared as an optional integer, e.g. max_file_size_mb. 

23 

24 bool is an int subclass, so an explicit isinstance(raw, bool) exclusion is needed 

25 or a YAML `true`/`false` would silently pass as 1/0. 

26 """ 

27 if raw is None: 27 ↛ 29line 27 didn't jump to line 29 because the condition on line 27 was always true

28 return None 

29 if isinstance(raw, int) and not isinstance(raw, bool): 

30 return raw 

31 raise TypeError(f"expected an integer, got {raw!r}") 

32 

33 

34def coerce_optional_str_list_setting(raw: object) -> tuple[str, ...] | None: 

35 if raw is None: 

36 return None 

37 if not isinstance(raw, list) or not all(isinstance(item, str) for item in raw): 

38 raise TypeError(f"expected a list of strings, got {raw!r}") 

39 return tuple(raw) 

40 

41 

42@dataclass(frozen=True, slots=True) 

43class UploadedFileTooLarge: 

44 size_bytes: int 

45 limit_mb: int 

46 

47 

48@dataclass(frozen=True, slots=True) 

49class UploadedFileExtensionNotAllowed: 

50 extension: str 

51 

52 

53@dataclass(frozen=True, slots=True) 

54class UploadedFileBlockedExtension: 

55 extension: str 

56 

57 

58@dataclass(frozen=True, slots=True) 

59class UploadedFileUnsafeFilename: 

60 filename: str 

61 

62 

63UploadValidationFailure = ( 

64 UploadedFileTooLarge | UploadedFileExtensionNotAllowed | UploadedFileBlockedExtension | UploadedFileUnsafeFilename 

65) 

66 

67 

68def _file_size_bytes(file_source: bytes | BinaryIO) -> int: 

69 if isinstance(file_source, bytes): 

70 return len(file_source) 

71 original_position: Final = file_source.tell() 

72 file_source.seek(0, 2) 

73 size: Final = file_source.tell() 

74 file_source.seek(original_position) 

75 return size 

76 

77 

78def check_upload_file_size( 

79 file_source: bytes | BinaryIO, 

80 max_file_size_mb: int | None, 

81) -> UploadedFileTooLarge | None: 

82 if max_file_size_mb is None or max_file_size_mb <= 0: 

83 return None 

84 size_bytes: Final = _file_size_bytes(file_source) 

85 if size_bytes > max_file_size_mb * MB: 

86 return UploadedFileTooLarge(size_bytes=size_bytes, limit_mb=max_file_size_mb) 

87 return None 

88 

89 

90def _normalized_extension(filename: str | None) -> str: 

91 if not filename: 

92 return "" 

93 try: 

94 return Path(safe_filename(filename)).suffix.lower() 

95 except ValueError: 

96 return "" 

97 

98 

99def check_allowed_extension( 

100 filename: str | None, 

101 allowed_extensions: tuple[str, ...] | None, 

102) -> UploadedFileExtensionNotAllowed | None: 

103 if allowed_extensions is None: 

104 return None 

105 extension: Final = _normalized_extension(filename) 

106 normalized_allowed: Final = frozenset(item.lower() for item in allowed_extensions) 

107 if extension and extension in normalized_allowed: 

108 return None 

109 return UploadedFileExtensionNotAllowed(extension=extension) 

110 

111 

112def check_blocked_extension( 

113 filename: str | None, 

114 blocked_extensions: tuple[str, ...] | None, 

115) -> UploadedFileBlockedExtension | None: 

116 if not blocked_extensions: 

117 return None 

118 extension: Final = _normalized_extension(filename) 

119 normalized_blocked: Final = frozenset(item.lower() for item in blocked_extensions) 

120 if extension and extension in normalized_blocked: 

121 return UploadedFileBlockedExtension(extension=extension) 

122 return None 

123 

124 

125def check_unsafe_filename(filename: str | None) -> UploadedFileUnsafeFilename | None: 

126 """Reject a filename before it can influence any storage path or backend call. 

127 

128 Only flags a genuine traversal component ("..") or a null byte, so an ordinary 

129 name like "report.v2.pdf" or ".env" is never rejected. 

130 """ 

131 if not filename: 131 ↛ 132line 131 didn't jump to line 132 because the condition on line 131 was never true

132 return None 

133 if "\x00" in filename: 133 ↛ 134line 133 didn't jump to line 134 because the condition on line 133 was never true

134 return UploadedFileUnsafeFilename(filename=filename) 

135 normalized: Final = filename.replace("\\", "/") 

136 if any(part == ".." for part in normalized.split("/")): 136 ↛ 137line 136 didn't jump to line 137 because the condition on line 136 was never true

137 return UploadedFileUnsafeFilename(filename=filename) 

138 return None 

139 

140 

141def raise_upload_validation_failure(failure: UploadValidationFailure) -> NoReturn: 

142 match failure: 

143 case UploadedFileTooLarge(size_bytes=size_bytes, limit_mb=limit_mb): 

144 raise ProxyException( 

145 message=( 

146 f"Uploaded file exceeds the configured max_file_size_mb of {limit_mb} MB " 

147 f"(read stopped at {size_bytes / MB:.1f} MB). The file was not forwarded to the provider." 

148 ), 

149 type="invalid_request_error", 

150 param="file", 

151 code=413, 

152 ) 

153 case UploadedFileExtensionNotAllowed(extension=extension): 

154 raise ProxyException( 

155 message=( 

156 (f"File extension '{extension}'" if extension else "A file without an extension") 

157 + " is not in this proxy's allowed_file_extensions setting. " 

158 "The file was not forwarded to the provider." 

159 ), 

160 type="invalid_request_error", 

161 param="file", 

162 code=400, 

163 ) 

164 case UploadedFileBlockedExtension(extension=extension): 

165 raise ProxyException( 

166 message=( 

167 f"File extension '{extension}' is blocked by this proxy's blocked_file_extensions " 

168 "setting. The file was not forwarded to the provider." 

169 ), 

170 type="invalid_request_error", 

171 param="file", 

172 code=400, 

173 ) 

174 case UploadedFileUnsafeFilename(filename=filename): 

175 raise ProxyException( 

176 message=( 

177 f"Filename '{filename}' is not allowed: directory traversal sequences are not " 

178 "permitted in uploaded file names. The file was not forwarded to the provider." 

179 ), 

180 type="invalid_request_error", 

181 param="file", 

182 code=400, 

183 ) 

184 case _: 

185 assert_never(failure)