You can not select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
bazarr/libs/werkzeug/formparser.py

496 lines
17 KiB

import typing as t
import warnings
5 years ago
from functools import update_wrapper
from io import BytesIO
5 years ago
from itertools import chain
from typing import Union
5 years ago
from . import exceptions
from ._internal import _to_str
5 years ago
from .datastructures import FileStorage
from .datastructures import Headers
from .datastructures import MultiDict
from .http import parse_options_header
from .sansio.multipart import Data
from .sansio.multipart import Epilogue
from .sansio.multipart import Field
from .sansio.multipart import File
from .sansio.multipart import MultipartDecoder
from .sansio.multipart import NeedData
5 years ago
from .urls import url_decode_stream
from .wsgi import _make_chunk_iter
5 years ago
from .wsgi import get_content_length
from .wsgi import get_input_stream
# there are some platforms where SpooledTemporaryFile is not available.
# In that case we need to provide a fallback.
try:
from tempfile import SpooledTemporaryFile
except ImportError:
from tempfile import TemporaryFile
SpooledTemporaryFile = None # type: ignore
5 years ago
if t.TYPE_CHECKING:
import typing as te
from _typeshed.wsgi import WSGIEnvironment
5 years ago
t_parse_result = t.Tuple[t.IO[bytes], MultiDict, MultiDict]
5 years ago
class TStreamFactory(te.Protocol):
def __call__(
self,
total_content_length: t.Optional[int],
content_type: t.Optional[str],
filename: t.Optional[str],
content_length: t.Optional[int] = None,
) -> t.IO[bytes]:
...
5 years ago
F = t.TypeVar("F", bound=t.Callable[..., t.Any])
def _exhaust(stream: t.IO[bytes]) -> None:
bts = stream.read(64 * 1024)
while bts:
bts = stream.read(64 * 1024)
5 years ago
def default_stream_factory(
total_content_length: t.Optional[int],
content_type: t.Optional[str],
filename: t.Optional[str],
content_length: t.Optional[int] = None,
) -> t.IO[bytes]:
5 years ago
max_size = 1024 * 500
5 years ago
if SpooledTemporaryFile is not None:
return t.cast(t.IO[bytes], SpooledTemporaryFile(max_size=max_size, mode="rb+"))
elif total_content_length is None or total_content_length > max_size:
return t.cast(t.IO[bytes], TemporaryFile("rb+"))
5 years ago
return BytesIO()
def parse_form_data(
environ: "WSGIEnvironment",
stream_factory: t.Optional["TStreamFactory"] = None,
charset: str = "utf-8",
errors: str = "replace",
max_form_memory_size: t.Optional[int] = None,
max_content_length: t.Optional[int] = None,
cls: t.Optional[t.Type[MultiDict]] = None,
silent: bool = True,
) -> "t_parse_result":
5 years ago
"""Parse the form data in the environ and return it as tuple in the form
``(stream, form, files)``. You should only call this method if the
transport method is `POST`, `PUT`, or `PATCH`.
If the mimetype of the data transmitted is `multipart/form-data` the
files multidict will be filled with `FileStorage` objects. If the
mimetype is unknown the input stream is wrapped and returned as first
argument, else the stream is empty.
This is a shortcut for the common usage of :class:`FormDataParser`.
Have a look at :doc:`/request_data` for more details.
5 years ago
.. versionadded:: 0.5
The `max_form_memory_size`, `max_content_length` and
`cls` parameters were added.
.. versionadded:: 0.5.1
The optional `silent` flag was added.
:param environ: the WSGI environment to be used for parsing.
:param stream_factory: An optional callable that returns a new read and
writeable file descriptor. This callable works
the same as :meth:`Response._get_file_stream`.
5 years ago
:param charset: The character set for URL and url encoded form data.
:param errors: The encoding error behavior.
:param max_form_memory_size: the maximum number of bytes to be accepted for
in-memory stored form data. If the data
exceeds the value specified an
:exc:`~exceptions.RequestEntityTooLarge`
exception is raised.
:param max_content_length: If this is provided and the transmitted data
is longer than this value an
:exc:`~exceptions.RequestEntityTooLarge`
exception is raised.
:param cls: an optional dict class to use. If this is not specified
or `None` the default :class:`MultiDict` is used.
:param silent: If set to False parsing errors will not be caught.
:return: A tuple in the form ``(stream, form, files)``.
"""
return FormDataParser(
stream_factory,
charset,
errors,
max_form_memory_size,
max_content_length,
cls,
silent,
).parse_from_environ(environ)
def exhaust_stream(f: F) -> F:
5 years ago
"""Helper decorator for methods that exhausts the stream on return."""
def wrapper(self, stream, *args, **kwargs): # type: ignore
5 years ago
try:
return f(self, stream, *args, **kwargs)
finally:
exhaust = getattr(stream, "exhaust", None)
5 years ago
if exhaust is not None:
exhaust()
else:
while True:
5 years ago
chunk = stream.read(1024 * 64)
5 years ago
if not chunk:
break
return update_wrapper(t.cast(F, wrapper), f)
5 years ago
class FormDataParser:
5 years ago
"""This class implements parsing of form data for Werkzeug. By itself
it can parse multipart and url encoded form data. It can be subclassed
and extended but for most mimetypes it is a better idea to use the
untouched stream and expose it as separate attributes on a request
object.
.. versionadded:: 0.8
:param stream_factory: An optional callable that returns a new read and
writeable file descriptor. This callable works
the same as :meth:`Response._get_file_stream`.
5 years ago
:param charset: The character set for URL and url encoded form data.
:param errors: The encoding error behavior.
:param max_form_memory_size: the maximum number of bytes to be accepted for
in-memory stored form data. If the data
exceeds the value specified an
:exc:`~exceptions.RequestEntityTooLarge`
exception is raised.
:param max_content_length: If this is provided and the transmitted data
is longer than this value an
:exc:`~exceptions.RequestEntityTooLarge`
exception is raised.
:param cls: an optional dict class to use. If this is not specified
or `None` the default :class:`MultiDict` is used.
:param silent: If set to False parsing errors will not be caught.
"""
def __init__(
self,
stream_factory: t.Optional["TStreamFactory"] = None,
charset: str = "utf-8",
errors: str = "replace",
max_form_memory_size: t.Optional[int] = None,
max_content_length: t.Optional[int] = None,
cls: t.Optional[t.Type[MultiDict]] = None,
silent: bool = True,
) -> None:
5 years ago
if stream_factory is None:
stream_factory = default_stream_factory
5 years ago
self.stream_factory = stream_factory
self.charset = charset
self.errors = errors
self.max_form_memory_size = max_form_memory_size
self.max_content_length = max_content_length
5 years ago
if cls is None:
cls = MultiDict
5 years ago
self.cls = cls
self.silent = silent
def get_parse_func(
self, mimetype: str, options: t.Dict[str, str]
) -> t.Optional[
t.Callable[
["FormDataParser", t.IO[bytes], str, t.Optional[int], t.Dict[str, str]],
"t_parse_result",
]
]:
5 years ago
return self.parse_functions.get(mimetype)
def parse_from_environ(self, environ: "WSGIEnvironment") -> "t_parse_result":
5 years ago
"""Parses the information from the environment as form data.
:param environ: the WSGI environment to be used for parsing.
:return: A tuple in the form ``(stream, form, files)``.
"""
content_type = environ.get("CONTENT_TYPE", "")
content_length = get_content_length(environ)
mimetype, options = parse_options_header(content_type)
return self.parse(get_input_stream(environ), mimetype, content_length, options)
def parse(
self,
stream: t.IO[bytes],
mimetype: str,
content_length: t.Optional[int],
options: t.Optional[t.Dict[str, str]] = None,
) -> "t_parse_result":
5 years ago
"""Parses the information from the given stream, mimetype,
content length and mimetype parameters.
:param stream: an input stream
:param mimetype: the mimetype of the data
:param content_length: the content length of the incoming data
:param options: optional mimetype parameters (used for
the multipart boundary for instance)
:return: A tuple in the form ``(stream, form, files)``.
"""
if (
self.max_content_length is not None
and content_length is not None
and content_length > self.max_content_length
):
# if the input stream is not exhausted, firefox reports Connection Reset
_exhaust(stream)
5 years ago
raise exceptions.RequestEntityTooLarge()
5 years ago
if options is None:
options = {}
parse_func = self.get_parse_func(mimetype, options)
5 years ago
if parse_func is not None:
try:
return parse_func(self, stream, mimetype, content_length, options)
except ValueError:
if not self.silent:
raise
return stream, self.cls(), self.cls()
@exhaust_stream
def _parse_multipart(
self,
stream: t.IO[bytes],
mimetype: str,
content_length: t.Optional[int],
options: t.Dict[str, str],
) -> "t_parse_result":
5 years ago
parser = MultiPartParser(
self.stream_factory,
self.charset,
self.errors,
max_form_memory_size=self.max_form_memory_size,
cls=self.cls,
)
boundary = options.get("boundary", "").encode("ascii")
if not boundary:
5 years ago
raise ValueError("Missing boundary")
5 years ago
form, files = parser.parse(stream, boundary, content_length)
return stream, form, files
@exhaust_stream
def _parse_urlencoded(
self,
stream: t.IO[bytes],
mimetype: str,
content_length: t.Optional[int],
options: t.Dict[str, str],
) -> "t_parse_result":
5 years ago
if (
self.max_form_memory_size is not None
and content_length is not None
and content_length > self.max_form_memory_size
):
# if the input stream is not exhausted, firefox reports Connection Reset
_exhaust(stream)
5 years ago
raise exceptions.RequestEntityTooLarge()
5 years ago
form = url_decode_stream(stream, self.charset, errors=self.errors, cls=self.cls)
return stream, form, self.cls()
#: mapping of mimetypes to parsing functions
parse_functions: t.Dict[
str,
t.Callable[
["FormDataParser", t.IO[bytes], str, t.Optional[int], t.Dict[str, str]],
"t_parse_result",
],
] = {
5 years ago
"multipart/form-data": _parse_multipart,
"application/x-www-form-urlencoded": _parse_urlencoded,
"application/x-url-encoded": _parse_urlencoded,
}
def _line_parse(line: str) -> t.Tuple[str, bool]:
5 years ago
"""Removes line ending characters and returns a tuple (`stripped_line`,
`is_terminated`).
"""
if line[-2:] == "\r\n":
5 years ago
return line[:-2], True
elif line[-1:] in {"\r", "\n"}:
5 years ago
return line[:-1], True
5 years ago
return line, False
def parse_multipart_headers(iterable: t.Iterable[bytes]) -> Headers:
5 years ago
"""Parses multipart headers from an iterable that yields lines (including
the trailing newline symbol). The iterable has to be newline terminated.
The iterable will stop at the line where the headers ended so it can be
further consumed.
:param iterable: iterable of strings that are newline terminated
"""
warnings.warn(
"'parse_multipart_headers' is deprecated and will be removed in"
" Werkzeug 2.1.",
DeprecationWarning,
stacklevel=2,
)
result: t.List[t.Tuple[str, str]] = []
for b_line in iterable:
line = _to_str(b_line)
5 years ago
line, line_terminated = _line_parse(line)
5 years ago
if not line_terminated:
raise ValueError("unexpected end of line in multipart header")
5 years ago
if not line:
break
elif line[0] in " \t" and result:
key, value = result[-1]
result[-1] = (key, f"{value}\n {line[1:]}")
5 years ago
else:
parts = line.split(":", 1)
5 years ago
if len(parts) == 2:
result.append((parts[0].strip(), parts[1].strip()))
# we link the list to the headers, no need to create a copy, the
# list was not shared anyways.
return Headers(result)
class MultiPartParser:
5 years ago
def __init__(
self,
stream_factory: t.Optional["TStreamFactory"] = None,
charset: str = "utf-8",
errors: str = "replace",
max_form_memory_size: t.Optional[int] = None,
cls: t.Optional[t.Type[MultiDict]] = None,
buffer_size: int = 64 * 1024,
) -> None:
5 years ago
self.charset = charset
self.errors = errors
self.max_form_memory_size = max_form_memory_size
if stream_factory is None:
stream_factory = default_stream_factory
self.stream_factory = stream_factory
if cls is None:
cls = MultiDict
self.cls = cls
5 years ago
self.buffer_size = buffer_size
def fail(self, message: str) -> "te.NoReturn":
5 years ago
raise ValueError(message)
def get_part_charset(self, headers: Headers) -> str:
5 years ago
# Figure out input charset for current part
content_type = headers.get("content-type")
5 years ago
if content_type:
mimetype, ct_params = parse_options_header(content_type)
return ct_params.get("charset", self.charset)
5 years ago
return self.charset
def start_file_streaming(
self, event: File, total_content_length: t.Optional[int]
) -> t.IO[bytes]:
content_type = event.headers.get("content-type")
5 years ago
try:
content_length = int(event.headers["content-length"])
5 years ago
except (KeyError, ValueError):
content_length = 0
5 years ago
container = self.stream_factory(
total_content_length=total_content_length,
filename=event.filename,
5 years ago
content_type=content_type,
content_length=content_length,
)
return container
5 years ago
def parse(
self, stream: t.IO[bytes], boundary: bytes, content_length: t.Optional[int]
) -> t.Tuple[MultiDict, MultiDict]:
container: t.Union[t.IO[bytes], t.List[bytes]]
_write: t.Callable[[bytes], t.Any]
5 years ago
iterator = chain(
_make_chunk_iter(
stream,
5 years ago
limit=content_length,
buffer_size=self.buffer_size,
),
[None],
5 years ago
)
parser = MultipartDecoder(boundary, self.max_form_memory_size)
fields = []
files = []
current_part: Union[Field, File]
for data in iterator:
parser.receive_data(data)
event = parser.next_event()
while not isinstance(event, (Epilogue, NeedData)):
if isinstance(event, Field):
current_part = event
container = []
_write = container.append
elif isinstance(event, File):
current_part = event
container = self.start_file_streaming(event, content_length)
_write = container.write
elif isinstance(event, Data):
_write(event.data)
if not event.more_data:
if isinstance(current_part, Field):
value = b"".join(container).decode(
self.get_part_charset(current_part.headers), self.errors
)
fields.append((current_part.name, value))
else:
container = t.cast(t.IO[bytes], container)
container.seek(0)
files.append(
(
current_part.name,
FileStorage(
container,
current_part.filename,
current_part.name,
headers=current_part.headers,
),
)
)
event = parser.next_event()
return self.cls(fields), self.cls(files)