codekingpro/portable-devtools

1from __future__ import annotations2 3import typing as t4from io import BytesIO5from urllib.parse import parse_qsl6 7from ._internal import _plain_int8from .datastructures import FileStorage9from .datastructures import Headers10from .datastructures import MultiDict11from .exceptions import RequestEntityTooLarge12from .http import parse_options_header13from .sansio.multipart import Data14from .sansio.multipart import Epilogue15from .sansio.multipart import Field16from .sansio.multipart import File17from .sansio.multipart import MultipartDecoder18from .sansio.multipart import NeedData19from .wsgi import get_content_length20from .wsgi import get_input_stream21 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 SpooledTemporaryFile26except ImportError:27    from tempfile import TemporaryFile28 29    SpooledTemporaryFile = None  # type: ignore30 31if t.TYPE_CHECKING:32    import typing as te33 34    from _typeshed.wsgi import WSGIEnvironment35 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 * 50060 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 form80    ``(stream, form, files)``.  You should only call this method if the81    transport method is `POST`, `PUT`, or `PATCH`.82 83    If the mimetype of the data transmitted is `multipart/form-data` the84    files multidict will be filled with `FileStorage` objects.  If the85    mimetype is unknown the input stream is wrapped and returned as first86    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 and92                           writeable file descriptor.  This callable works93                           the same as :meth:`Response._get_file_stream`.94    :param max_form_memory_size: the maximum number of bytes to be accepted for95                           in-memory stored form data.  If the data96                           exceeds the value specified an97                           :exc:`~exceptions.RequestEntityTooLarge`98                           exception is raised.99    :param max_content_length: If this is provided and the transmitted data100                               is longer than this value an101                               :exc:`~exceptions.RequestEntityTooLarge`102                               exception is raised.103    :param cls: an optional dict class to use.  If this is not specified104                       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 this107        is exceeded, a :exc:`~exceptions.RequestEntityTooLarge` exception is raised.108    :return: A tuple in the form ``(stream, form, files)``.109 110    .. versionchanged:: 3.0111        The ``charset`` and ``errors`` parameters were removed.112 113    .. versionchanged:: 2.3114        Added the ``max_form_parts`` parameter.115 116    .. versionadded:: 0.5.1117       Added the ``silent`` parameter.118 119    .. versionadded:: 0.5120       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 itself135    it can parse multipart and url encoded form data.  It can be subclassed136    and extended but for most mimetypes it is a better idea to use the137    untouched stream and expose it as separate attributes on a request138    object.139 140    :param stream_factory: An optional callable that returns a new read and141                           writeable file descriptor.  This callable works142                           the same as :meth:`Response._get_file_stream`.143    :param max_form_memory_size: the maximum number of bytes to be accepted for144                           in-memory stored form data.  If the data145                           exceeds the value specified an146                           :exc:`~exceptions.RequestEntityTooLarge`147                           exception is raised.148    :param max_content_length: If this is provided and the transmitted data149                               is longer than this value an150                               :exc:`~exceptions.RequestEntityTooLarge`151                               exception is raised.152    :param cls: an optional dict class to use.  If this is not specified153                       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 this156        is exceeded, a :exc:`~exceptions.RequestEntityTooLarge` exception is raised.157 158    .. versionchanged:: 3.0159        The ``charset`` and ``errors`` parameters were removed.160 161    .. versionchanged:: 3.0162        The ``parse_functions`` attribute and ``get_parse_func`` methods were removed.163 164    .. versionchanged:: 2.2.3165        Added the ``max_form_parts`` parameter.166 167    .. versionadded:: 0.8168    """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_factory182 183        self.stream_factory = stream_factory184        self.max_form_memory_size = max_form_memory_size185        self.max_content_length = max_content_length186        self.max_form_parts = max_form_parts187 188        if cls is None:189            cls = t.cast("type[MultiDict[str, t.Any]]", MultiDict)190 191        self.cls = cls192        self.silent = silent193 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 stream221        :param mimetype: the mimetype of the data222        :param content_length: the content length of the incoming data223        :param options: optional mimetype parameters (used for224                        the multipart boundary for instance)225        :return: A tuple in the form ``(stream, form, files)``.226 227        .. versionchanged:: 3.0228            The invalid ``application/x-url-encoded`` content type is not229            treated as ``application/x-www-form-urlencoded``.230        """231        if mimetype == "multipart/form-data":232            parse_func = self._parse_multipart233        elif mimetype == "application/x-www-form-urlencoded":234            parse_func = self._parse_urlencoded235        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                raise246 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, files269 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 None279            and content_length is not None280            and content_length > self.max_form_memory_size281        ):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 e292 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_size306        self.max_form_parts = max_form_parts307 308        if stream_factory is None:309            stream_factory = default_stream_factory310 311        self.stream_factory = stream_factory312 313        if cls is None:314            cls = t.cast("type[MultiDict[str, t.Any]]", MultiDict)315 316        self.cls = cls317        self.buffer_size = buffer_size318 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 part324        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_charset334 335        return "utf-8"336 337    def start_file_streaming(338        self, event: File, total_content_length: int | None339    ) -> 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 = 0346 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 container354 355    def parse(356        self, stream: t.IO[bytes], boundary: bytes, content_length: int | None357    ) -> tuple[MultiDict[str, str], MultiDict[str, FileStorage]]:358        current_part: Field | File359        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 = event377                    container = []378                    _write = container.append379                elif isinstance(event, File):380                    current_part = event381                    container = self.start_file_streaming(event, content_length)382                    _write = container.write383                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            break420 421        yield data422 423    yield None424