Coverage for /pythoncovmergedfiles/medio/medio/usr/local/lib/python3.11/site-packages/werkzeug/formparser.py: 68%
Shortcuts on this page
r m x toggle line displays
j k next/prev highlighted chunk
0 (zero) top of page
1 (one) first highlighted chunk
1from __future__ import annotations
3import collections.abc as cabc
4import typing as t
5import wsgiref.types as wt
6from tempfile import SpooledTemporaryFile
7from types import TracebackType
8from urllib.parse import parse_qsl
10from ._internal import _plain_int
11from .datastructures.file_storage import FileStorage
12from .datastructures.headers import Headers
13from .datastructures.structures import ImmutableMultiDict
14from .datastructures.structures import MultiDict
15from .exceptions import RequestEntityTooLarge
16from .http import parse_options_header
17from .sansio.multipart import Data
18from .sansio.multipart import Epilogue
19from .sansio.multipart import Field
20from .sansio.multipart import File
21from .sansio.multipart import MultipartDecoder
22from .sansio.multipart import NeedData
24F = t.TypeVar("F", bound=t.Callable[..., t.Any])
25t_parse_result = tuple[t.IO[bytes], MultiDict[str, str], MultiDict[str, FileStorage]]
28class TStreamFactory(t.Protocol):
29 def __call__(
30 self,
31 total_content_length: int | None,
32 content_type: str | None,
33 filename: str | None,
34 content_length: int | None = None,
35 ) -> t.IO[bytes]: ...
38def _make_stream_factory(max_size: int | None) -> TStreamFactory:
39 if max_size is None:
40 # Could return BytesIO for this, keeping 500 kB default for now.
41 max_size = 1024 * 500
43 def stream_factory(
44 total_content_length: int | None,
45 content_type: str | None,
46 filename: str | None,
47 content_length: int | None = None,
48 ) -> t.IO[bytes]:
49 return SpooledTemporaryFile(max_size=max_size, mode="rb+")
51 return stream_factory
54_default_stream_factory = _make_stream_factory(1024 * 500)
57def _parse_form_data(
58 environ: wt.WSGIEnvironment,
59 stream_factory: TStreamFactory | None = None,
60 max_form_memory_size: int | None = None,
61 max_content_length: int | None = None,
62 silent: bool = True,
63 *,
64 max_form_parts: int | None = None,
65 **kwargs: t.Any,
66) -> t_parse_result:
67 """Parse the form data from the request body and return it as a
68 tuple ``(stream, form, files)``.
70 If the mimetype of the data transmitted is `multipart/form-data` the
71 files multidict will be filled with `FileStorage` objects. If the
72 mimetype is unknown the input stream is wrapped and returned as first
73 argument, else the stream is empty.
75 This is a shortcut for the common usage of :class:`FormDataParser`.
77 :param environ: the WSGI environment to be used for parsing.
78 :param stream_factory: A callable that returns a file in ``rb+`` mode.
79 Defaults to creating a :class:`tempfile.SpooledTemporaryFile` with
80 ``max_form_memory_size``.
81 :param max_content_length: If the data is larger than this many bytes, raise
82 :exc:`.RequestEntityTooLarge`. This is used by :meth:`parse_from_environ`
83 to set up a limited stream. When using :meth:`parse`, you must get a
84 limited stream with :func:`.get_input_stream`.
85 :param max_form_memory_size: If a ``multipart/form-data`` text part is larger
86 than this many bytes, raise :exc:`~exceptions.RequestEntityTooLarge`.
87 File parts are written to disk after this size. This is an additional
88 check, it does not replace using a limited stream and
89 ``max_content_length``.
90 :param max_form_parts: If more than this number of ``multipart/form-data``
91 parts are received, raise :exc:`.RequestEntityTooLarge`. This is an
92 additional check, it does not replace using a limited stream and
93 ``max_content_length``.
94 :param cls: an optional dict class to use. If this is not specified
95 or `None` the default :class:`MultiDict` is used.
96 :param silent: If set to False parsing errors will not be caught.
97 :return: A tuple in the form ``(stream, form, files)``.
99 .. deprecated:: 3.2
100 Will be removed in Werkzeug 4.0. Use ``Request.form`` and ``files``
101 instead.
103 .. versionchanged:: 3.2
104 The ``cls`` parameter is deprecated and will be removed in Werkzeug 4.0. It will
105 always be ``ImmutableMultiDict``.
107 .. versionchanged:: 3.1.9
108 ``max_form_memory_size`` is not applied to
109 ``application/x-www-form-urlencoded``.
111 .. versionchanged:: 3.0
112 The ``charset`` and ``errors`` parameters were removed.
114 .. versionchanged:: 2.3
115 Added the ``max_form_parts`` parameter.
117 .. versionadded:: 0.5.1
118 Added the ``silent`` parameter.
120 .. versionadded:: 0.5
121 Added the ``max_form_memory_size``, ``max_content_length``, and ``cls``
122 parameters.
123 """
124 parser_kwargs: dict[str, t.Any] = dict(
125 stream_factory=stream_factory,
126 max_form_memory_size=max_form_memory_size,
127 max_content_length=max_content_length,
128 max_form_parts=max_form_parts,
129 silent=silent,
130 )
132 if "cls" in kwargs:
133 import warnings
135 warnings.warn(
136 "The 'cls' parameter is deprecated and will be removed in Werkzeug 4.0."
137 " It will always be 'ImmutableMultiDict'.",
138 DeprecationWarning,
139 stacklevel=2,
140 )
141 parser_kwargs["cls"] = kwargs["cls"]
143 return _FormDataParser(**parser_kwargs).parse_from_environ(environ)
146class _FormDataParser:
147 """This class implements parsing of form data for Werkzeug. By itself
148 it can parse multipart and url encoded form data. It can be subclassed
149 and extended but for most mimetypes it is a better idea to use the
150 untouched stream and expose it as separate attributes on a request
151 object.
153 :param max_content_length: If the data is larger than this many bytes, raise
154 :exc:`.RequestEntityTooLarge`. This is used by :meth:`parse_from_environ`
155 to set up a limited stream. When using :meth:`parse`, you must get a
156 limited stream with :func:`.get_input_stream`.
157 :param max_form_memory_size: If a ``multipart/form-data`` text part is larger
158 than this many bytes, raise :exc:`~exceptions.RequestEntityTooLarge`.
159 File parts are written to disk after this size. This is an additional
160 check, it does not replace using a limited stream and
161 ``max_content_length``.
162 :param max_form_parts: If more than this number of ``multipart/form-data``
163 parts are received, raise :exc:`.RequestEntityTooLarge`. This is an
164 additional check, it does not replace using a limited stream and
165 ``max_content_length``.
166 :param cls: an optional dict class to use. If this is not specified
167 or `None` the default :class:`MultiDict` is used.
168 :param silent: If set to False parsing errors will not be caught.
170 .. deprecated:: 3.2
171 Will be removed in Werkzeug 4.0. Use ``Request.form`` and ``files``
172 instead.
174 .. versionchanged:: 3.2
175 The ``cls`` parameter and attribute are deprecated and will be removed
176 in Werkzeug 4.0. They will always be ``ImmutableMultiDict``.
178 .. versionchanged:: 3.1.9
179 ``max_form_memory_size`` is not applied to
180 ``application/x-www-form-urlencoded``.
182 .. versionchanged:: 3.0
183 The ``charset`` and ``errors`` parameters were removed.
185 .. versionchanged:: 3.0
186 The ``parse_functions`` attribute and ``get_parse_func`` methods were removed.
188 .. versionchanged:: 2.2.3
189 Added the ``max_form_parts`` parameter.
191 .. versionadded:: 0.8
192 """
194 def __init__(
195 self,
196 stream_factory: TStreamFactory | None = None,
197 max_form_memory_size: int | None = None,
198 max_content_length: int | None = None,
199 silent: bool = True,
200 *,
201 max_form_parts: int | None = None,
202 **kwargs: t.Any,
203 ) -> None:
204 self.stream_factory = stream_factory
205 self.max_content_length = max_content_length
206 self.max_form_memory_size = max_form_memory_size
207 self.max_form_parts = max_form_parts
209 if "cls" in kwargs:
210 import warnings
212 warnings.warn(
213 "The 'cls' parameter is deprecated and will be removed in Werkzeug 4.0."
214 " It will always be 'ImmutableMultiDict'.",
215 DeprecationWarning,
216 stacklevel=2,
217 )
219 self.cls: type[ImmutableMultiDict[str, t.Any]] | None = kwargs.get("cls")
220 self.silent = silent
222 def parse_from_environ(self, environ: wt.WSGIEnvironment) -> t_parse_result:
223 """Parses the information from the environment as form data.
225 :param environ: the WSGI environment to be used for parsing.
226 :return: A tuple in the form ``(stream, form, files)``.
227 """
228 from .wrappers.request import Request
230 request = Request(environ)
231 request.max_content_length = self.max_content_length
232 return self.parse(
233 request.stream,
234 content_length=request.content_length,
235 mimetype=request.mimetype,
236 options=request.mimetype_params,
237 )
239 def parse(
240 self,
241 stream: t.IO[bytes],
242 mimetype: str,
243 content_length: int | None,
244 options: cabc.Mapping[str, str] | None = None,
245 ) -> t_parse_result:
246 """Parses the information from the given stream, mimetype,
247 content length and mimetype parameters.
249 :param stream: A limited input stream, from :meth:`.get_input_stream`.
250 :param mimetype: The mimetype used to choose how to parse the form.
251 ``multipart/form-data`` and ``application/x-www-form-urlencoded``
252 are supported.
253 :param content_length: the content length of the incoming data
254 :param options: optional mimetype parameters (used for
255 the multipart boundary for instance)
256 :return: A tuple in the form ``(stream, form, files)``.
258 .. versionchanged:: 3.0
259 The invalid ``application/x-url-encoded`` content type is not
260 treated as ``application/x-www-form-urlencoded``.
261 """
262 if mimetype == "multipart/form-data":
263 parse_func = self._parse_multipart
264 elif mimetype == "application/x-www-form-urlencoded":
265 parse_func = self._parse_urlencoded
266 else:
267 if self.cls is not None:
268 return stream, self.cls(), self.cls()
270 return stream, ImmutableMultiDict(), ImmutableMultiDict()
272 if options is None:
273 options = {}
275 try:
276 return parse_func(stream, mimetype, content_length, options)
277 except ValueError:
278 if not self.silent:
279 raise
281 if self.cls is not None:
282 return stream, self.cls(), self.cls()
284 return stream, ImmutableMultiDict(), ImmutableMultiDict()
286 def _parse_multipart(
287 self,
288 stream: t.IO[bytes],
289 mimetype: str,
290 content_length: int | None,
291 options: cabc.Mapping[str, str],
292 ) -> t_parse_result:
293 boundary = options.get("boundary", "").encode("ascii")
295 if not boundary:
296 raise ValueError("Missing boundary")
298 kwargs: dict[str, t.Any] = dict(
299 stream_factory=self.stream_factory,
300 max_form_memory_size=self.max_form_memory_size,
301 max_form_parts=self.max_form_parts,
302 )
304 if self.cls is not None:
305 kwargs["cls"] = self.cls
307 with MultiPartParser(**kwargs) as parser:
308 form, files = parser.parse(stream, boundary, content_length)
310 return stream, form, files
312 def _parse_urlencoded(
313 self,
314 stream: t.IO[bytes],
315 mimetype: str,
316 content_length: int | None,
317 options: cabc.Mapping[str, str],
318 ) -> t_parse_result:
319 # The stream must already be limited to max_content_length.
320 # max_form_memory_size can't apply, since the entire stream is read at
321 # once instead of incrementally parsed.
322 # max_form_parts could be applied here, but doesn't impose a useful
323 # limit given the previous points and because the parser is much simpler
324 # and faster already.
325 items = parse_qsl(
326 stream.read().decode(),
327 keep_blank_values=True,
328 errors="werkzeug.url_quote",
329 )
331 if self.cls is not None:
332 return stream, self.cls(items), self.cls()
334 return stream, ImmutableMultiDict(items), ImmutableMultiDict()
337class MultiPartParser:
338 def __init__(
339 self,
340 stream_factory: TStreamFactory | None = None,
341 max_form_memory_size: int | None = None,
342 buffer_size: int = 64 * 1024,
343 max_form_parts: int | None = None,
344 **kwargs: t.Any,
345 ) -> None:
346 self.max_form_memory_size = max_form_memory_size
347 self.max_form_parts = max_form_parts
349 if stream_factory is None:
350 stream_factory = _make_stream_factory(max_form_memory_size)
352 self.stream_factory = stream_factory
353 self._files: list[t.IO[bytes]] = []
355 if "cls" in kwargs:
356 import warnings
358 warnings.warn(
359 "The 'cls' parameter is deprecated and will be removed in Werkzeug 4.0."
360 " It will always be 'ImmutableMultiDict'.",
361 DeprecationWarning,
362 stacklevel=2,
363 )
365 self.cls: type[ImmutableMultiDict[str, t.Any]] | None = kwargs.get("cls")
366 self.buffer_size = buffer_size
368 def __enter__(self) -> t.Self:
369 return self
371 def __exit__(
372 self,
373 exc_type: type[BaseException] | None,
374 exc_val: BaseException | None,
375 exc_tb: TracebackType | None,
376 ) -> None:
377 if exc_val is not None:
378 for file in self._files:
379 file.close()
381 def get_part_charset(self, headers: Headers) -> str:
382 # Figure out input charset for current part
383 content_type = headers.get("Content-Type")
385 if content_type:
386 parameters = parse_options_header(content_type)[1]
387 ct_charset = parameters.get("charset", "").lower()
389 # A safe list of encodings. Modern clients should only send ASCII or UTF-8.
390 # This list will not be extended further.
391 if ct_charset in {"ascii", "us-ascii", "utf-8", "iso-8859-1"}:
392 return ct_charset
394 return "utf-8"
396 def start_file_streaming(
397 self, event: File, total_content_length: int | None
398 ) -> t.IO[bytes]:
399 content_type = event.headers.get("Content-Type")
401 try:
402 content_length = _plain_int(event.headers["Content-Length"])
403 except (KeyError, ValueError):
404 content_length = 0
406 container = self.stream_factory(
407 total_content_length=total_content_length,
408 filename=event.filename,
409 content_type=content_type,
410 content_length=content_length,
411 )
412 self._files.append(container)
413 return container
415 def parse(
416 self, stream: t.IO[bytes], boundary: bytes, content_length: int | None
417 ) -> tuple[MultiDict[str, str], MultiDict[str, FileStorage]]:
418 current_part: Field | File
419 field_size: int | None = None
420 container: t.IO[bytes] | list[bytes]
421 _write: t.Callable[[bytes], t.Any]
423 parser = MultipartDecoder(
424 boundary,
425 max_form_memory_size=self.max_form_memory_size,
426 max_parts=self.max_form_parts,
427 )
429 fields = []
430 files = []
432 for data in _chunk_iter(stream.read, self.buffer_size):
433 parser.receive_data(data)
434 event = parser.next_event()
435 while not isinstance(event, (Epilogue, NeedData)):
436 if isinstance(event, Field):
437 current_part = event
438 field_size = 0
439 container = []
440 _write = container.append
441 elif isinstance(event, File):
442 current_part = event
443 field_size = None
444 container = self.start_file_streaming(event, content_length)
445 _write = container.write
446 elif isinstance(event, Data):
447 if self.max_form_memory_size is not None and field_size is not None:
448 # Ensure that accumulated data events do not exceed limit.
449 # Also checked within single event in MultipartDecoder.
450 field_size += len(event.data)
452 if field_size > self.max_form_memory_size:
453 raise RequestEntityTooLarge()
455 _write(event.data)
456 if not event.more_data:
457 if isinstance(current_part, Field):
458 value = b"".join(container).decode(
459 self.get_part_charset(current_part.headers), "replace"
460 )
461 fields.append((current_part.name, value))
462 else:
463 container = t.cast(t.IO[bytes], container)
464 container.seek(0)
465 files.append(
466 (
467 current_part.name,
468 FileStorage(
469 container,
470 current_part.filename,
471 current_part.name,
472 headers=current_part.headers,
473 ),
474 )
475 )
477 event = parser.next_event()
479 if self.cls is not None:
480 return self.cls(fields), self.cls(files)
482 return ImmutableMultiDict(fields), ImmutableMultiDict(files)
485def _chunk_iter(read: t.Callable[[int], bytes], size: int) -> t.Iterator[bytes | None]:
486 """Read data in chunks for multipart/form-data parsing. Stop if no data is read.
487 Yield ``None`` at the end to signal end of parsing.
488 """
489 while True:
490 data = read(size)
492 if not data:
493 break
495 yield data
497 yield None
500if not t.TYPE_CHECKING:
502 def __getattr__(name: str) -> t.Any:
503 import warnings
505 if name == "default_stream_factory":
506 warnings.warn(
507 "'default_stream_factory' is deprecated and will be removed in Werkzeug"
508 " 4.0. If not passed, 'FormDataParser' will use 'SpooledTemporaryFile'"
509 " with 'max_form_memory_size'.",
510 DeprecationWarning,
511 stacklevel=2,
512 )
513 return _default_stream_factory
515 if name == "parse_form_data":
516 warnings.warn(
517 "'parse_form_data' is deprecated and will be removed in Werkzeug 4.0."
518 " Use 'Request.form' and 'files' instead.",
519 DeprecationWarning,
520 stacklevel=2,
521 )
522 return _parse_form_data
524 if name == "FormDataParser":
525 warnings.warn(
526 "The 'FormDataParser' class is deprecated and will be removed in"
527 " Werkzeug 4.0. Use 'Request.form' and 'files' instead.",
528 DeprecationWarning,
529 stacklevel=2,
530 )
531 return _FormDataParser
533 raise AttributeError(name)