1"""
2Multi-part parsing for file uploads.
3
4Exposes one class, ``MultiPartParser``, which feeds chunks of uploaded data to
5file upload handlers for processing.
6"""
7
8import base64
9import binascii
10import collections
11import html
12from collections.abc import Iterator
13from typing import TYPE_CHECKING, Any
14
15from plain.internal.files.uploadhandler import SkipFile, StopFutureHandlers, StopUpload
16from plain.runtime import settings
17from plain.utils.datastructures import MultiValueDict
18from plain.utils.encoding import force_str
19from plain.utils.http import parse_header_parameters
20from plain.utils.module_loading import import_string
21from plain.utils.regex_helper import _lazy_re_compile
22
23from .exceptions import (
24 BadRequestError400,
25 ContentTooLargeError413,
26 SuspiciousMultipartFormError400,
27 TooManyFieldsSentError400,
28 TooManyFilesSentError400,
29)
30
31if TYPE_CHECKING:
32 from plain.http.request import Request
33 from plain.internal.files.uploadhandler import FileUploadHandler
34
35__all__ = ["InputStreamExhausted", "MultiPartParser", "MultiPartParserError"]
36
37
38class MultiPartParserError(BadRequestError400):
39 pass
40
41
42class InputStreamExhausted(Exception):
43 """
44 No more reads are allowed from this device.
45 """
46
47
48_RAW = "raw"
49_FILE = "file"
50_FIELD = "field"
51_FIELD_TYPES = frozenset([_FIELD, _RAW])
52
53
54class MultiPartParser:
55 """
56 An RFC 7578 multipart/form-data parser.
57
58 ``MultiValueDict.parse()`` reads the input stream in ``chunk_size`` chunks
59 and returns a tuple of ``(MultiValueDict(POST), MultiValueDict(FILES))``.
60 """
61
62 boundary_re = _lazy_re_compile(r"[ -~]{0,200}[!-~]")
63
64 def __init__(self, request: Request):
65 """
66 Initialize the MultiPartParser object.
67
68 :request:
69 The HTTP request object (used for headers and as the input stream).
70 """
71 # Content-Type should contain multipart and the boundary information.
72 content_type = request.content_type or ""
73 if not content_type.startswith("multipart/"):
74 raise MultiPartParserError(f"Invalid Content-Type: {content_type}")
75
76 try:
77 content_type.encode("ascii")
78 except UnicodeEncodeError:
79 raise MultiPartParserError(
80 f"Invalid non-ASCII Content-Type in multipart: {force_str(content_type)}"
81 )
82
83 # Get the boundary from parsed content type parameters.
84 content_params = request.content_params or {}
85 boundary = content_params.get("boundary")
86 if not boundary or not self.boundary_re.fullmatch(boundary):
87 raise MultiPartParserError(
88 f"Invalid boundary in multipart: {force_str(boundary)}"
89 )
90
91 content_length = request.content_length
92 if content_length < 0:
93 # This means we shouldn't continue...raise an error.
94 raise MultiPartParserError(f"Invalid content length: {content_length!r}")
95
96 self._boundary = boundary.encode("ascii")
97 self._request = request
98
99 # Create upload handlers for this parsing session
100 self._upload_handlers: list[FileUploadHandler] = [
101 import_string(handler)(request) for handler in settings.FILE_UPLOAD_HANDLERS
102 ]
103
104 # For compatibility with low-level network APIs (with 32-bit integers),
105 # the chunk size should be < 2^31, but still divisible by 4.
106 possible_sizes = [x.chunk_size for x in self._upload_handlers if x.chunk_size]
107 self._chunk_size = min([2**31 - 4] + possible_sizes)
108
109 self._encoding = request.encoding or "utf-8"
110 self._content_length = content_length
111
112 def parse(self) -> tuple[Any, MultiValueDict]:
113 # Call the actual parse routine and close all open files in case of
114 # errors. This is needed because if exceptions are thrown the
115 # MultiPartParser will not be garbage collected immediately and
116 # resources would be kept alive. This is only needed for errors because
117 # the Request object closes all uploaded files at the end of the
118 # request.
119 try:
120 return self._parse()
121 except Exception:
122 if hasattr(self, "_files"):
123 for _, files in self._files.lists():
124 for fileobj in files:
125 fileobj.close()
126 raise
127
128 def _parse(self) -> tuple[Any, MultiValueDict]:
129 """
130 Parse the POST data and break it into a FILES MultiValueDict and a POST
131 MultiValueDict.
132
133 Return a tuple containing the POST and FILES dictionary, respectively.
134 """
135 from plain.http import QueryDict
136
137 encoding = self._encoding
138 handlers = self._upload_handlers
139
140 # HTTP spec says that Content-Length >= 0 is valid
141 # handling content-length == 0 before continuing
142 if self._content_length == 0:
143 return QueryDict(encoding=self._encoding), MultiValueDict()
144
145 # See if any of the handlers take care of the parsing.
146 # This allows overriding everything if need be.
147 for handler in handlers:
148 result = handler.handle_raw_input(
149 self._request,
150 self._boundary,
151 encoding,
152 )
153 # Check to see if it was handled
154 if result is not None:
155 return result[0], result[1]
156
157 # Create the data structures to be used later.
158 self._post = QueryDict(mutable=True)
159 self._files = MultiValueDict()
160
161 # Instantiate the parser and stream:
162 stream = LazyStream(ChunkIter(self._request, self._chunk_size))
163
164 # Whether or not to signal a file-completion at the beginning of the loop.
165 old_field_name = None
166 counters = [0] * len(handlers)
167
168 # Number of bytes that have been read.
169 num_bytes_read = 0
170 # To count the number of keys in the request.
171 num_post_keys = 0
172 # To count the number of files in the request.
173 num_files = 0
174 # To limit the amount of data read from the request.
175 read_size = None
176 # Whether a file upload is finished.
177 uploaded_file = True
178
179 try:
180 for item_type, meta_data, field_stream in Parser(stream, self._boundary):
181 if old_field_name:
182 # We run this at the beginning of the next loop
183 # since we cannot be sure a file is complete until
184 # we hit the next boundary/part of the multipart content.
185 self.handle_file_complete(old_field_name, counters)
186 old_field_name = None
187 uploaded_file = True
188
189 if (
190 item_type in _FIELD_TYPES
191 and settings.DATA_UPLOAD_MAX_NUMBER_FIELDS is not None
192 ):
193 # Avoid storing more than DATA_UPLOAD_MAX_NUMBER_FIELDS.
194 num_post_keys += 1
195 # 2 accounts for empty raw fields before and after the
196 # last boundary.
197 if settings.DATA_UPLOAD_MAX_NUMBER_FIELDS + 2 < num_post_keys:
198 raise TooManyFieldsSentError400(
199 "The number of GET/POST parameters exceeded "
200 "settings.DATA_UPLOAD_MAX_NUMBER_FIELDS."
201 )
202
203 try:
204 disposition = meta_data["content-disposition"][1]
205 field_name = disposition["name"].strip()
206 except KeyError, IndexError, AttributeError:
207 continue
208
209 transfer_encoding = meta_data.get("content-transfer-encoding")
210 if transfer_encoding is not None:
211 transfer_encoding = transfer_encoding[0].strip()
212 field_name = force_str(field_name, encoding, errors="replace")
213
214 if item_type == _FIELD:
215 # Avoid reading more than DATA_UPLOAD_MAX_MEMORY_SIZE.
216 if settings.DATA_UPLOAD_MAX_MEMORY_SIZE is not None:
217 read_size = (
218 settings.DATA_UPLOAD_MAX_MEMORY_SIZE - num_bytes_read
219 )
220
221 # This is a post field, we can just set it in the post
222 if transfer_encoding == "base64":
223 raw_data = field_stream.read(size=read_size)
224 num_bytes_read += len(raw_data)
225 try:
226 data = base64.b64decode(raw_data)
227 except binascii.Error:
228 data = raw_data
229 else:
230 data = field_stream.read(size=read_size)
231 num_bytes_read += len(data)
232
233 # Add two here to make the check consistent with the
234 # x-www-form-urlencoded check that includes '&='.
235 num_bytes_read += len(field_name) + 2
236 if (
237 settings.DATA_UPLOAD_MAX_MEMORY_SIZE is not None
238 and num_bytes_read > settings.DATA_UPLOAD_MAX_MEMORY_SIZE
239 ):
240 raise ContentTooLargeError413(
241 "Request body exceeded "
242 "settings.DATA_UPLOAD_MAX_MEMORY_SIZE."
243 )
244
245 self._post.appendlist(
246 field_name, force_str(data, encoding, errors="replace")
247 )
248 elif item_type == _FILE:
249 # Avoid storing more than DATA_UPLOAD_MAX_NUMBER_FILES.
250 num_files += 1
251 if (
252 settings.DATA_UPLOAD_MAX_NUMBER_FILES is not None
253 and num_files > settings.DATA_UPLOAD_MAX_NUMBER_FILES
254 ):
255 raise TooManyFilesSentError400(
256 "The number of files exceeded "
257 "settings.DATA_UPLOAD_MAX_NUMBER_FILES."
258 )
259 # This is a file, use the handler...
260 file_name = disposition.get("filename")
261 if file_name:
262 file_name = force_str(file_name, encoding, errors="replace")
263 file_name = self.sanitize_file_name(file_name)
264 if not file_name:
265 continue
266
267 content_type, content_type_extra = meta_data.get(
268 "content-type", ("", {})
269 )
270 content_type = content_type.strip()
271 charset = content_type_extra.get("charset")
272
273 try:
274 content_length_value = meta_data.get("content-length")
275 content_length = (
276 int(content_length_value[0])
277 if content_length_value
278 else None
279 )
280 except IndexError, TypeError, ValueError:
281 content_length = None
282
283 counters = [0] * len(handlers)
284 uploaded_file = False
285 try:
286 for handler in handlers:
287 try:
288 handler.new_file(
289 field_name,
290 file_name,
291 content_type,
292 content_length,
293 charset,
294 content_type_extra,
295 )
296 except StopFutureHandlers:
297 break
298
299 for chunk in field_stream:
300 if transfer_encoding == "base64":
301 # We only special-case base64 transfer encoding
302 # We should always decode base64 chunks by
303 # multiple of 4, ignoring whitespace.
304
305 stripped_chunk = b"".join(chunk.split())
306
307 remaining = len(stripped_chunk) % 4
308 while remaining != 0:
309 over_chunk = field_stream.read(4 - remaining)
310 if not over_chunk:
311 break
312 stripped_chunk += b"".join(over_chunk.split())
313 remaining = len(stripped_chunk) % 4
314
315 try:
316 chunk = base64.b64decode(stripped_chunk)
317 except Exception as exc:
318 # Since this is only a chunk, any error is
319 # an unfixable error.
320 raise MultiPartParserError(
321 "Could not decode base64 data."
322 ) from exc
323
324 for i, handler in enumerate(handlers):
325 chunk_length = len(chunk)
326 chunk = handler.receive_data_chunk(chunk, counters[i])
327 counters[i] += chunk_length
328 if chunk is None:
329 # Don't continue if the chunk received by
330 # the handler is None.
331 break
332
333 except SkipFile:
334 self._close_files()
335 # Just use up the rest of this file...
336 _exhaust(field_stream)
337 else:
338 # Handle file upload completions on next iteration.
339 old_field_name = field_name
340 else:
341 # If this is neither a FIELD nor a FILE, exhaust the field
342 # stream. Note: There could be an error here at some point,
343 # but there will be at least two RAW types (before and
344 # after the other boundaries). This branch is usually not
345 # reached at all, because a missing content-disposition
346 # header will skip the whole boundary.
347 _exhaust(field_stream)
348 except StopUpload as e:
349 self._close_files()
350 if not e.connection_reset:
351 _exhaust(self._request)
352 else:
353 if not uploaded_file:
354 for handler in handlers:
355 handler.upload_interrupted()
356 # Make sure that the request data is all fed
357 _exhaust(self._request)
358
359 # Signal that the upload has completed.
360 # any() shortcircuits if a handler's upload_complete() returns a value.
361 any(handler.upload_complete() for handler in handlers)
362 self._post._mutable = False
363 return self._post, self._files
364
365 def handle_file_complete(self, old_field_name: str, counters: list[int]) -> None:
366 """
367 Handle all the signaling that takes place when a file is complete.
368 """
369 for i, handler in enumerate(self._upload_handlers):
370 file_obj = handler.file_complete(counters[i])
371 if file_obj:
372 # If it returns a file object, then set the files dict.
373 self._files.appendlist(
374 force_str(old_field_name, self._encoding, errors="replace"),
375 file_obj,
376 )
377 break
378
379 def sanitize_file_name(self, file_name: str) -> str | None:
380 """
381 Sanitize the filename of an upload.
382
383 Remove all possible path separators, even though that might remove more
384 than actually required by the target system. Filenames that could
385 potentially cause problems (current/parent dir) are also discarded.
386
387 It should be noted that this function could still return a "filepath"
388 like "C:some_file.txt" which is handled later on by the storage layer.
389 So while this function does sanitize filenames to some extent, the
390 resulting filename should still be considered as untrusted user input.
391 """
392 file_name = html.unescape(file_name)
393 file_name = file_name.rsplit("/")[-1]
394 file_name = file_name.rsplit("\\")[-1]
395 # Remove non-printable characters.
396 file_name = "".join([char for char in file_name if char.isprintable()])
397
398 if file_name in {"", ".", ".."}:
399 return None
400 return file_name
401
402 def _close_files(self) -> None:
403 # Free up all file handles.
404 # FIXME: this currently assumes that upload handlers store the file as 'file'
405 # We should document that...
406 # (Maybe add handler.free_file to complement new_file)
407 for handler in self._upload_handlers:
408 file = getattr(handler, "file", None)
409 if file is not None:
410 file.close()
411
412
413class LazyStream:
414 """
415 The LazyStream wrapper allows one to get and "unget" bytes from a stream.
416
417 Given a producer object (an iterator that yields bytestrings), the
418 LazyStream object will support iteration, reading, and keeping a "look-back"
419 variable in case you need to "unget" some bytes.
420 """
421
422 def __init__(self, producer: Iterator[bytes], length: int | None = None):
423 """
424 Every LazyStream must have a producer when instantiated.
425
426 A producer is an iterable that returns a string each time it
427 is called.
428 """
429 self._producer = producer
430 self._empty = False
431 self._leftover = b""
432 self.length = length
433 self.position = 0
434 self._remaining = length
435 self._unget_history = []
436
437 def tell(self) -> int:
438 return self.position
439
440 def read(self, size: int | None = None) -> bytes:
441 def parts() -> Iterator[bytes]:
442 remaining = self._remaining if size is None else size
443 # do the whole thing in one shot if no limit was provided.
444 if remaining is None:
445 yield b"".join(self)
446 return
447
448 # otherwise do some bookkeeping to return exactly enough
449 # of the stream and stashing any extra content we get from
450 # the producer
451 while remaining != 0:
452 assert remaining > 0, "remaining bytes to read should never go negative"
453
454 try:
455 chunk = next(self)
456 except StopIteration:
457 return
458 else:
459 emitting = chunk[:remaining]
460 self.unget(chunk[remaining:])
461 remaining -= len(emitting)
462 yield emitting
463
464 return b"".join(parts())
465
466 def __next__(self) -> bytes:
467 """
468 Used when the exact number of bytes to read is unimportant.
469
470 Return whatever chunk is conveniently returned from the iterator.
471 Useful to avoid unnecessary bookkeeping if performance is an issue.
472 """
473 if self._leftover:
474 output = self._leftover
475 self._leftover = b""
476 else:
477 output = next(self._producer)
478 self._unget_history = []
479 self.position += len(output)
480 return output
481
482 def close(self) -> None:
483 """
484 Used to invalidate/disable this lazy stream.
485
486 Replace the producer with an empty list. Any leftover bytes that have
487 already been read will still be reported upon read() and/or next().
488 """
489 self._producer = iter([])
490
491 def __iter__(self) -> LazyStream:
492 return self
493
494 def unget(self, bytes: bytes) -> None:
495 """
496 Place bytes back onto the front of the lazy stream.
497
498 Future calls to read() will return those bytes first. The
499 stream position and thus tell() will be rewound.
500 """
501 if not bytes:
502 return
503 self._update_unget_history(len(bytes))
504 self.position -= len(bytes)
505 self._leftover = bytes + self._leftover
506
507 def _update_unget_history(self, num_bytes: int) -> None:
508 """
509 Update the unget history as a sanity check to see if we've pushed
510 back the same number of bytes in one chunk. If we keep ungetting the
511 same number of bytes many times (here, 50), we're mostly likely in an
512 infinite loop of some sort. This is usually caused by a
513 maliciously-malformed MIME request.
514 """
515 self._unget_history = [num_bytes] + self._unget_history[:49]
516 number_equal = len(
517 [
518 current_number
519 for current_number in self._unget_history
520 if current_number == num_bytes
521 ]
522 )
523
524 if number_equal > 40:
525 raise SuspiciousMultipartFormError400(
526 "The multipart parser got stuck, which shouldn't happen with"
527 " normal uploaded files. Check for malicious upload activity;"
528 " if there is none, report this to the Plain developers."
529 )
530
531
532class ChunkIter:
533 """
534 An iterable that will yield chunks of data. Given a file-like object as the
535 constructor, yield chunks of read operations from that object.
536 """
537
538 def __init__(self, flo: Any, chunk_size: int = 64 * 1024):
539 self.flo = flo
540 self.chunk_size = chunk_size
541
542 def __next__(self) -> bytes:
543 try:
544 data = self.flo.read(self.chunk_size)
545 except InputStreamExhausted:
546 raise StopIteration()
547 if data:
548 return data
549 else:
550 raise StopIteration()
551
552 def __iter__(self) -> ChunkIter:
553 return self
554
555
556class InterBoundaryIter:
557 """
558 A Producer that will iterate over boundaries.
559 """
560
561 def __init__(self, stream: LazyStream, boundary: bytes):
562 self._stream = stream
563 self._boundary = boundary
564
565 def __iter__(self) -> InterBoundaryIter:
566 return self
567
568 def __next__(self) -> LazyStream:
569 try:
570 return LazyStream(BoundaryIter(self._stream, self._boundary))
571 except InputStreamExhausted:
572 raise StopIteration()
573
574
575class BoundaryIter:
576 """
577 A Producer that is sensitive to boundaries.
578
579 Will happily yield bytes until a boundary is found. Will yield the bytes
580 before the boundary, throw away the boundary bytes themselves, and push the
581 post-boundary bytes back on the stream.
582
583 The future calls to next() after locating the boundary will raise a
584 StopIteration exception.
585 """
586
587 def __init__(self, stream: LazyStream, boundary: bytes):
588 self._stream = stream
589 self._boundary = boundary
590 self._done = False
591 # rollback an additional six bytes because the format is like
592 # this: CRLF<boundary>[--CRLF]
593 self._rollback = len(boundary) + 6
594
595 # Try to use mx fast string search if available. Otherwise
596 # use Python find. Wrap the latter for consistency.
597 unused_char = self._stream.read(1)
598 if not unused_char:
599 raise InputStreamExhausted()
600 self._stream.unget(unused_char)
601
602 def __iter__(self) -> BoundaryIter:
603 return self
604
605 def __next__(self) -> bytes:
606 if self._done:
607 raise StopIteration()
608
609 stream = self._stream
610 rollback = self._rollback
611
612 bytes_read = 0
613 chunks = []
614 for bytes in stream:
615 bytes_read += len(bytes)
616 chunks.append(bytes)
617 if bytes_read > rollback:
618 break
619 if not bytes:
620 break
621 else:
622 self._done = True
623
624 if not chunks:
625 raise StopIteration()
626
627 chunk = b"".join(chunks)
628 boundary = self._find_boundary(chunk)
629
630 if boundary:
631 end, next = boundary
632 stream.unget(chunk[next:])
633 self._done = True
634 return chunk[:end]
635 else:
636 # make sure we don't treat a partial boundary (and
637 # its separators) as data
638 if not chunk[:-rollback]: # and len(chunk) >= (len(self._boundary) + 6):
639 # There's nothing left, we should just return and mark as done.
640 self._done = True
641 return chunk
642 else:
643 stream.unget(chunk[-rollback:])
644 return chunk[:-rollback]
645
646 def _find_boundary(self, data: bytes) -> tuple[int, int] | None:
647 """
648 Find a multipart boundary in data.
649
650 Should no boundary exist in the data, return None. Otherwise, return
651 a tuple containing the indices of the following:
652 * the end of current encapsulation
653 * the start of the next encapsulation
654 """
655 index = data.find(self._boundary)
656 if index < 0:
657 return None
658 else:
659 end = index
660 next = index + len(self._boundary)
661 # backup over CRLF
662 last = max(0, end - 1)
663 if data[last : last + 1] == b"\n":
664 end -= 1
665 last = max(0, end - 1)
666 if data[last : last + 1] == b"\r":
667 end -= 1
668 return end, next
669
670
671def _exhaust(stream_or_iterable: Any) -> None:
672 """Exhaust an iterator or stream."""
673 try:
674 iterator = iter(stream_or_iterable)
675 except TypeError:
676 iterator = ChunkIter(stream_or_iterable, 16384)
677 collections.deque(iterator, maxlen=0) # consume iterator quickly.
678
679
680def _parse_boundary_stream(
681 stream: LazyStream, max_header_size: int
682) -> tuple[str, dict[str, Any], LazyStream]:
683 """
684 Parse one and exactly one stream that encapsulates a boundary.
685 """
686 # Stream at beginning of header, look for end of header
687 # and parse it if found. The header must fit within one
688 # chunk.
689 chunk = stream.read(max_header_size)
690
691 # 'find' returns the top of these four bytes, so we'll
692 # need to munch them later to prevent them from polluting
693 # the payload.
694 header_end = chunk.find(b"\r\n\r\n")
695
696 if header_end == -1:
697 # we find no header, so we just mark this fact and pass on
698 # the stream verbatim
699 stream.unget(chunk)
700 return (_RAW, {}, stream)
701
702 header = chunk[:header_end]
703
704 # here we place any excess chunk back onto the stream, as
705 # well as throwing away the CRLFCRLF bytes from above.
706 stream.unget(chunk[header_end + 4 :])
707
708 TYPE = _RAW
709 outdict = {}
710
711 # Eliminate blank lines
712 for line in header.split(b"\r\n"):
713 # This terminology ("main value" and "dictionary of
714 # parameters") is from the Python docs.
715 try:
716 main_value_pair, params = parse_header_parameters(line.decode())
717 name, value = main_value_pair.split(":", 1)
718 params = {k: v.encode() for k, v in params.items()}
719 except ValueError: # Invalid header.
720 continue
721
722 if name == "content-disposition":
723 TYPE = _FIELD
724 if params.get("filename"):
725 TYPE = _FILE
726
727 outdict[name] = value, params
728
729 if TYPE == _RAW:
730 stream.unget(chunk)
731
732 return (TYPE, outdict, stream)
733
734
735class Parser:
736 def __init__(self, stream: LazyStream, boundary: bytes):
737 self._stream = stream
738 self._separator = b"--" + boundary
739
740 def __iter__(self) -> Iterator[tuple[str, dict[str, Any], LazyStream]]:
741 boundarystream = InterBoundaryIter(self._stream, self._separator)
742 for sub_stream in boundarystream:
743 # Iterate over each part
744 yield _parse_boundary_stream(sub_stream, 1024)