1from __future__ import annotations
2
3import typing as t
4from io import BytesIO
5from urllib.parse import parse_qsl
6
7from ._internal import _plain_int
8from .datastructures import FileStorage
9from .datastructures import Headers
10from .datastructures import MultiDict
11from .exceptions import RequestEntityTooLarge
12from .http import parse_options_header
13from .sansio.multipart import Data
14from .sansio.multipart import Epilogue
15from .sansio.multipart import Field
16from .sansio.multipart import File
17from .sansio.multipart import MultipartDecoder
18from .sansio.multipart import NeedData
19from .wsgi import get_content_length
20from .wsgi import get_input_stream
21
22# there are some platforms where SpooledTemporaryFile is not available.
23# In that case we need to provide a fallback.
24try:
25 from tempfile import SpooledTemporaryFile
26except ImportError:
27 from tempfile import TemporaryFile
28
29 SpooledTemporaryFile = None # type: ignore
30
31if t.TYPE_CHECKING:
32 import typing as te
33
34 from _typeshed.wsgi import WSGIEnvironment
35
36 t_parse_result = t.Tuple[
37 t.IO[bytes], MultiDict[str, str], MultiDict[str, FileStorage]
38 ]
39
40 class TStreamFactory(te.Protocol):
41 def __call__(
42 self,
43 total_content_length: int | None,
44 content_type: str | None,
45 filename: str | None,
46 content_length: int | None = None,
47 ) -> t.IO[bytes]: ...
48
49
50F = t.TypeVar("F", bound=t.Callable[..., t.Any])
51
52
53def default_stream_factory(
54 total_content_length: int | None,
55 content_type: str | None,
56 filename: str | None,
57 content_length: int | None = None,
58) -> t.IO[bytes]:
59 max_size = 1024 * 500
60
61 if SpooledTemporaryFile is not None:
62 return t.cast(t.IO[bytes], SpooledTemporaryFile(max_size=max_size, mode="rb+"))
63 elif total_content_length is None or total_content_length > max_size:
64 return t.cast(t.IO[bytes], TemporaryFile("rb+"))
65
66 return BytesIO()
67
68
69def parse_form_data(
70 environ: WSGIEnvironment,
71 stream_factory: TStreamFactory | None = None,
72 max_form_memory_size: int | None = None,
73 max_content_length: int | None = None,
74 cls: type[MultiDict[str, t.Any]] | None = None,
75 silent: bool = True,
76 *,
77 max_form_parts: int | None = None,
78) -> t_parse_result:
79 """Parse the form data in the environ and return it as tuple in the form
80 ``(stream, form, files)``. You should only call this method if the
81 transport method is `POST`, `PUT`, or `PATCH`.
82
83 If the mimetype of the data transmitted is `multipart/form-data` the
84 files multidict will be filled with `FileStorage` objects. If the
85 mimetype is unknown the input stream is wrapped and returned as first
86 argument, else the stream is empty.
87
88 This is a shortcut for the common usage of :class:`FormDataParser`.
89
90 :param environ: the WSGI environment to be used for parsing.
91 :param stream_factory: An optional callable that returns a new read and
92 writeable file descriptor. This callable works
93 the same as :meth:`Response._get_file_stream`.
94 :param max_form_memory_size: the maximum number of bytes to be accepted for
95 in-memory stored form data. If the data
96 exceeds the value specified an
97 :exc:`~exceptions.RequestEntityTooLarge`
98 exception is raised.
99 :param max_content_length: If this is provided and the transmitted data
100 is longer than this value an
101 :exc:`~exceptions.RequestEntityTooLarge`
102 exception is raised.
103 :param cls: an optional dict class to use. If this is not specified
104 or `None` the default :class:`MultiDict` is used.
105 :param silent: If set to False parsing errors will not be caught.
106 :param max_form_parts: The maximum number of multipart parts to be parsed. If this
107 is exceeded, a :exc:`~exceptions.RequestEntityTooLarge` exception is raised.
108 :return: A tuple in the form ``(stream, form, files)``.
109
110 .. versionchanged:: 3.0
111 The ``charset`` and ``errors`` parameters were removed.
112
113 .. versionchanged:: 2.3
114 Added the ``max_form_parts`` parameter.
115
116 .. versionadded:: 0.5.1
117 Added the ``silent`` parameter.
118
119 .. versionadded:: 0.5
120 Added the ``max_form_memory_size``, ``max_content_length``, and ``cls``
121 parameters.
122 """
123 return FormDataParser(
124 stream_factory=stream_factory,
125 max_form_memory_size=max_form_memory_size,
126 max_content_length=max_content_length,
127 max_form_parts=max_form_parts,
128 silent=silent,
129 cls=cls,
130 ).parse_from_environ(environ)
131
132
133class FormDataParser:
134 """This class implements parsing of form data for Werkzeug. By itself
135 it can parse multipart and url encoded form data. It can be subclassed
136 and extended but for most mimetypes it is a better idea to use the
137 untouched stream and expose it as separate attributes on a request
138 object.
139
140 :param stream_factory: An optional callable that returns a new read and
141 writeable file descriptor. This callable works
142 the same as :meth:`Response._get_file_stream`.
143 :param max_form_memory_size: the maximum number of bytes to be accepted for
144 in-memory stored form data. If the data
145 exceeds the value specified an
146 :exc:`~exceptions.RequestEntityTooLarge`
147 exception is raised.
148 :param max_content_length: If this is provided and the transmitted data
149 is longer than this value an
150 :exc:`~exceptions.RequestEntityTooLarge`
151 exception is raised.
152 :param cls: an optional dict class to use. If this is not specified
153 or `None` the default :class:`MultiDict` is used.
154 :param silent: If set to False parsing errors will not be caught.
155 :param max_form_parts: The maximum number of multipart parts to be parsed. If this
156 is exceeded, a :exc:`~exceptions.RequestEntityTooLarge` exception is raised.
157
158 .. versionchanged:: 3.0
159 The ``charset`` and ``errors`` parameters were removed.
160
161 .. versionchanged:: 3.0
162 The ``parse_functions`` attribute and ``get_parse_func`` methods were removed.
163
164 .. versionchanged:: 2.2.3
165 Added the ``max_form_parts`` parameter.
166
167 .. versionadded:: 0.8
168 """
169
170 def __init__(
171 self,
172 stream_factory: TStreamFactory | None = None,
173 max_form_memory_size: int | None = None,
174 max_content_length: int | None = None,
175 cls: type[MultiDict[str, t.Any]] | None = None,
176 silent: bool = True,
177 *,
178 max_form_parts: int | None = None,
179 ) -> None:
180 if stream_factory is None:
181 stream_factory = default_stream_factory
182
183 self.stream_factory = stream_factory
184 self.max_form_memory_size = max_form_memory_size
185 self.max_content_length = max_content_length
186 self.max_form_parts = max_form_parts
187
188 if cls is None:
189 cls = t.cast("type[MultiDict[str, t.Any]]", MultiDict)
190
191 self.cls = cls
192 self.silent = silent
193
194 def parse_from_environ(self, environ: WSGIEnvironment) -> t_parse_result:
195 """Parses the information from the environment as form data.
196
197 :param environ: the WSGI environment to be used for parsing.
198 :return: A tuple in the form ``(stream, form, files)``.
199 """
200 stream = get_input_stream(environ, max_content_length=self.max_content_length)
201 content_length = get_content_length(environ)
202 mimetype, options = parse_options_header(environ.get("CONTENT_TYPE"))
203 return self.parse(
204 stream,
205 content_length=content_length,
206 mimetype=mimetype,
207 options=options,
208 )
209
210 def parse(
211 self,
212 stream: t.IO[bytes],
213 mimetype: str,
214 content_length: int | None,
215 options: dict[str, str] | None = None,
216 ) -> t_parse_result:
217 """Parses the information from the given stream, mimetype,
218 content length and mimetype parameters.
219
220 :param stream: an input stream
221 :param mimetype: the mimetype of the data
222 :param content_length: the content length of the incoming data
223 :param options: optional mimetype parameters (used for
224 the multipart boundary for instance)
225 :return: A tuple in the form ``(stream, form, files)``.
226
227 .. versionchanged:: 3.0
228 The invalid ``application/x-url-encoded`` content type is not
229 treated as ``application/x-www-form-urlencoded``.
230 """
231 if mimetype == "multipart/form-data":
232 parse_func = self._parse_multipart
233 elif mimetype == "application/x-www-form-urlencoded":
234 parse_func = self._parse_urlencoded
235 else:
236 return stream, self.cls(), self.cls()
237
238 if options is None:
239 options = {}
240
241 try:
242 return parse_func(stream, mimetype, content_length, options)
243 except ValueError:
244 if not self.silent:
245 raise
246
247 return stream, self.cls(), self.cls()
248
249 def _parse_multipart(
250 self,
251 stream: t.IO[bytes],
252 mimetype: str,
253 content_length: int | None,
254 options: dict[str, str],
255 ) -> t_parse_result:
256 parser = MultiPartParser(
257 stream_factory=self.stream_factory,
258 max_form_memory_size=self.max_form_memory_size,
259 max_form_parts=self.max_form_parts,
260 cls=self.cls,
261 )
262 boundary = options.get("boundary", "").encode("ascii")
263
264 if not boundary:
265 raise ValueError("Missing boundary")
266
267 form, files = parser.parse(stream, boundary, content_length)
268 return stream, form, files
269
270 def _parse_urlencoded(
271 self,
272 stream: t.IO[bytes],
273 mimetype: str,
274 content_length: int | None,
275 options: dict[str, str],
276 ) -> t_parse_result:
277 if (
278 self.max_form_memory_size is not None
279 and content_length is not None
280 and content_length > self.max_form_memory_size
281 ):
282 raise RequestEntityTooLarge()
283
284 try:
285 items = parse_qsl(
286 stream.read().decode(),
287 keep_blank_values=True,
288 errors="werkzeug.url_quote",
289 )
290 except ValueError as e:
291 raise RequestEntityTooLarge() from e
292
293 return stream, self.cls(items), self.cls()
294
295
296class MultiPartParser:
297 def __init__(
298 self,
299 stream_factory: TStreamFactory | None = None,
300 max_form_memory_size: int | None = None,
301 cls: type[MultiDict[str, t.Any]] | None = None,
302 buffer_size: int = 64 * 1024,
303 max_form_parts: int | None = None,
304 ) -> None:
305 self.max_form_memory_size = max_form_memory_size
306 self.max_form_parts = max_form_parts
307
308 if stream_factory is None:
309 stream_factory = default_stream_factory
310
311 self.stream_factory = stream_factory
312
313 if cls is None:
314 cls = t.cast("type[MultiDict[str, t.Any]]", MultiDict)
315
316 self.cls = cls
317 self.buffer_size = buffer_size
318
319 def fail(self, message: str) -> te.NoReturn:
320 raise ValueError(message)
321
322 def get_part_charset(self, headers: Headers) -> str:
323 # Figure out input charset for current part
324 content_type = headers.get("content-type")
325
326 if content_type:
327 parameters = parse_options_header(content_type)[1]
328 ct_charset = parameters.get("charset", "").lower()
329
330 # A safe list of encodings. Modern clients should only send ASCII or UTF-8.
331 # This list will not be extended further.
332 if ct_charset in {"ascii", "us-ascii", "utf-8", "iso-8859-1"}:
333 return ct_charset
334
335 return "utf-8"
336
337 def start_file_streaming(
338 self, event: File, total_content_length: int | None
339 ) -> t.IO[bytes]:
340 content_type = event.headers.get("content-type")
341
342 try:
343 content_length = _plain_int(event.headers["content-length"])
344 except (KeyError, ValueError):
345 content_length = 0
346
347 container = self.stream_factory(
348 total_content_length=total_content_length,
349 filename=event.filename,
350 content_type=content_type,
351 content_length=content_length,
352 )
353 return container
354
355 def parse(
356 self, stream: t.IO[bytes], boundary: bytes, content_length: int | None
357 ) -> tuple[MultiDict[str, str], MultiDict[str, FileStorage]]:
358 current_part: Field | File
359 container: t.IO[bytes] | list[bytes]
360 _write: t.Callable[[bytes], t.Any]
361
362 parser = MultipartDecoder(
363 boundary,
364 max_form_memory_size=self.max_form_memory_size,
365 max_parts=self.max_form_parts,
366 )
367
368 fields = []
369 files = []
370
371 for data in _chunk_iter(stream.read, self.buffer_size):
372 parser.receive_data(data)
373 event = parser.next_event()
374 while not isinstance(event, (Epilogue, NeedData)):
375 if isinstance(event, Field):
376 current_part = event
377 container = []
378 _write = container.append
379 elif isinstance(event, File):
380 current_part = event
381 container = self.start_file_streaming(event, content_length)
382 _write = container.write
383 elif isinstance(event, Data):
384 _write(event.data)
385 if not event.more_data:
386 if isinstance(current_part, Field):
387 value = b"".join(container).decode(
388 self.get_part_charset(current_part.headers), "replace"
389 )
390 fields.append((current_part.name, value))
391 else:
392 container = t.cast(t.IO[bytes], container)
393 container.seek(0)
394 files.append(
395 (
396 current_part.name,
397 FileStorage(
398 container,
399 current_part.filename,
400 current_part.name,
401 headers=current_part.headers,
402 ),
403 )
404 )
405
406 event = parser.next_event()
407
408 return self.cls(fields), self.cls(files)
409
410
411def _chunk_iter(read: t.Callable[[int], bytes], size: int) -> t.Iterator[bytes | None]:
412 """Read data in chunks for multipart/form-data parsing. Stop if no data is read.
413 Yield ``None`` at the end to signal end of parsing.
414 """
415 while True:
416 data = read(size)
417
418 if not data:
419 break
420
421 yield data
422
423 yield None