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