mirror of
https://github.com/vee1e/mitmproxy.git
synced 2026-09-01 18:27:18 +00:00
parent
430833e3d9
commit
7da3a8e871
7 changed files with 72 additions and 35 deletions
|
|
@ -17,6 +17,8 @@
|
|||
* Updating `Request.port` now also updates the Host header if present.
|
||||
This aligns with `Request.host`, which already does this.
|
||||
([#5908](https://github.com/mitmproxy/mitmproxy/pull/5908), @sujaldev)
|
||||
* Fix editing of multipart HTTP requests from the CLI.
|
||||
([#5148](https://github.com/mitmproxy/mitmproxy/issues/5148), @mhils)
|
||||
|
||||
### Breaking Changes
|
||||
|
||||
|
|
|
|||
|
|
@ -16,7 +16,7 @@ class ViewMultipart(base.View):
|
|||
def __call__(self, data: bytes, content_type: Optional[str] = None, **metadata):
|
||||
if content_type is None:
|
||||
return
|
||||
v = multipart.decode(content_type, data)
|
||||
v = multipart.decode_multipart(content_type, data)
|
||||
if v:
|
||||
return "Multipart form", self._format(v)
|
||||
|
||||
|
|
|
|||
|
|
@ -8,6 +8,7 @@ import warnings
|
|||
from collections.abc import Iterable
|
||||
from collections.abc import Iterator
|
||||
from collections.abc import Mapping
|
||||
from collections.abc import Sequence
|
||||
from dataclasses import dataclass
|
||||
from dataclasses import fields
|
||||
from email.utils import formatdate
|
||||
|
|
@ -963,7 +964,7 @@ class Request(Message):
|
|||
return tuple(url.decode(self.get_text(strict=False)))
|
||||
return ()
|
||||
|
||||
def _set_urlencoded_form(self, form_data):
|
||||
def _set_urlencoded_form(self, form_data: Sequence[tuple[str, str]]) -> None:
|
||||
"""
|
||||
Sets the body to the URL-encoded form data, and adds the appropriate content-type header.
|
||||
This will overwrite the existing content if there is one.
|
||||
|
|
@ -989,23 +990,22 @@ class Request(Message):
|
|||
def urlencoded_form(self, value):
|
||||
self._set_urlencoded_form(value)
|
||||
|
||||
def _get_multipart_form(self):
|
||||
def _get_multipart_form(self) -> list[tuple[bytes, bytes]]:
|
||||
is_valid_content_type = (
|
||||
"multipart/form-data" in self.headers.get("content-type", "").lower()
|
||||
)
|
||||
if is_valid_content_type and self.content is not None:
|
||||
try:
|
||||
return multipart.decode(self.headers.get("content-type"), self.content)
|
||||
return multipart.decode_multipart(
|
||||
self.headers.get("content-type"), self.content
|
||||
)
|
||||
except ValueError:
|
||||
pass
|
||||
return ()
|
||||
return []
|
||||
|
||||
def _set_multipart_form(self, value):
|
||||
is_valid_content_type = (
|
||||
self.headers.get("content-type", "")
|
||||
.lower()
|
||||
.startswith("multipart/form-data")
|
||||
)
|
||||
def _set_multipart_form(self, value: list[tuple[bytes, bytes]]) -> None:
|
||||
ct = self.headers.get("content-type", "")
|
||||
is_valid_content_type = ct.lower().startswith("multipart/form-data")
|
||||
if not is_valid_content_type:
|
||||
"""
|
||||
Generate a random boundary here.
|
||||
|
|
@ -1014,8 +1014,10 @@ class Request(Message):
|
|||
on generating the boundary.
|
||||
"""
|
||||
boundary = "-" * 20 + binascii.hexlify(os.urandom(16)).decode()
|
||||
self.headers["content-type"] = f"multipart/form-data; boundary={boundary}"
|
||||
self.content = multipart.encode(self.headers, value)
|
||||
self.headers[
|
||||
"content-type"
|
||||
] = ct = f"multipart/form-data; boundary={boundary}"
|
||||
self.content = multipart.encode_multipart(ct, value)
|
||||
|
||||
@property
|
||||
def multipart_form(self) -> multidict.MultiDictView[bytes, bytes]:
|
||||
|
|
@ -1032,7 +1034,7 @@ class Request(Message):
|
|||
)
|
||||
|
||||
@multipart_form.setter
|
||||
def multipart_form(self, value):
|
||||
def multipart_form(self, value: list[tuple[bytes, bytes]]) -> None:
|
||||
self._set_multipart_form(value)
|
||||
|
||||
|
||||
|
|
|
|||
|
|
@ -1,23 +1,25 @@
|
|||
from __future__ import annotations
|
||||
|
||||
import mimetypes
|
||||
import re
|
||||
import warnings
|
||||
from typing import Optional
|
||||
from urllib.parse import quote
|
||||
|
||||
from mitmproxy.net.http import headers
|
||||
|
||||
|
||||
def encode(head, l):
|
||||
k = head.get("content-type")
|
||||
if k:
|
||||
k = headers.parse_content_type(k)
|
||||
if k is not None:
|
||||
def encode_multipart(content_type: str, parts: list[tuple[bytes, bytes]]) -> bytes:
|
||||
if content_type:
|
||||
ct = headers.parse_content_type(content_type)
|
||||
if ct is not None:
|
||||
try:
|
||||
boundary = k[2]["boundary"].encode("ascii")
|
||||
boundary = quote(boundary)
|
||||
raw_boundary = ct[2]["boundary"].encode("ascii")
|
||||
boundary = quote(raw_boundary)
|
||||
except (KeyError, UnicodeError):
|
||||
return b""
|
||||
hdrs = []
|
||||
for key, value in l:
|
||||
for key, value in parts:
|
||||
file_type = (
|
||||
mimetypes.guess_type(str(key))[0] or "text/plain; charset=utf-8"
|
||||
)
|
||||
|
|
@ -41,9 +43,12 @@ def encode(head, l):
|
|||
hdrs.append(b"--%b--\r\n" % boundary.encode("utf-8"))
|
||||
temp = b"\r\n".join(hdrs)
|
||||
return temp
|
||||
return b""
|
||||
|
||||
|
||||
def decode(content_type: Optional[str], content: bytes) -> list[tuple[bytes, bytes]]:
|
||||
def decode_multipart(
|
||||
content_type: Optional[str], content: bytes
|
||||
) -> list[tuple[bytes, bytes]]:
|
||||
"""
|
||||
Takes a multipart boundary encoded string and returns list of (key, value) tuples.
|
||||
"""
|
||||
|
|
@ -69,3 +74,23 @@ def decode(content_type: Optional[str], content: bytes) -> list[tuple[bytes, byt
|
|||
r.append((key, value))
|
||||
return r
|
||||
return []
|
||||
|
||||
|
||||
def encode(ct, parts): # pragma: no cover
|
||||
# 2023-02
|
||||
warnings.warn(
|
||||
"multipart.encode is deprecated, use multipart.encode_multipart instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return encode_multipart(ct, parts)
|
||||
|
||||
|
||||
def decode(ct, content): # pragma: no cover
|
||||
# 2023-02
|
||||
warnings.warn(
|
||||
"multipart.decode is deprecated, use multipart.decode_multipart instead.",
|
||||
DeprecationWarning,
|
||||
stacklevel=2,
|
||||
)
|
||||
return encode_multipart(ct, content)
|
||||
|
|
|
|||
|
|
@ -51,7 +51,7 @@ class ResponseHeaderEditor(HeaderEditor):
|
|||
|
||||
class RequestMultipartEditor(base.FocusEditor):
|
||||
title = "Edit Multipart Form"
|
||||
columns = [col_text.Column("Key"), col_text.Column("Value")]
|
||||
columns = [col_bytes.Column("Key"), col_bytes.Column("Value")]
|
||||
|
||||
def get_data(self, flow):
|
||||
return flow.request.multipart_form.items(multi=True)
|
||||
|
|
|
|||
|
|
@ -1,6 +1,5 @@
|
|||
import pytest
|
||||
|
||||
from mitmproxy.http import Headers
|
||||
from mitmproxy.net.http import multipart
|
||||
|
||||
|
||||
|
|
@ -15,23 +14,28 @@ def test_decode():
|
|||
"value2\n"
|
||||
"--{0}--".format(boundary).encode()
|
||||
)
|
||||
form = multipart.decode(f"multipart/form-data; boundary={boundary}", content)
|
||||
form = multipart.decode_multipart(
|
||||
f"multipart/form-data; boundary={boundary}", content
|
||||
)
|
||||
|
||||
assert len(form) == 2
|
||||
assert form[0] == (b"field1", b"value1")
|
||||
assert form[1] == (b"field2", b"value2")
|
||||
|
||||
boundary = "boundary茅莽"
|
||||
result = multipart.decode(f"multipart/form-data; boundary={boundary}", content)
|
||||
result = multipart.decode_multipart(
|
||||
f"multipart/form-data; boundary={boundary}", content
|
||||
)
|
||||
assert result == []
|
||||
|
||||
assert multipart.decode("", content) == []
|
||||
assert multipart.decode_multipart("", content) == []
|
||||
|
||||
|
||||
def test_encode():
|
||||
data = [(b"file", b"shell.jpg"), (b"file_size", b"1000")]
|
||||
headers = Headers(content_type="multipart/form-data; boundary=127824672498")
|
||||
content = multipart.encode(headers, data)
|
||||
content = multipart.encode_multipart(
|
||||
"multipart/form-data; boundary=127824672498", data
|
||||
)
|
||||
|
||||
assert b'Content-Disposition: form-data; name="file"' in content
|
||||
assert (
|
||||
|
|
@ -42,9 +46,13 @@ def test_encode():
|
|||
assert len(content) == 252
|
||||
|
||||
with pytest.raises(ValueError, match=r"boundary found in encoded string"):
|
||||
multipart.encode(headers, [(b"key", b"--127824672498")])
|
||||
multipart.encode_multipart(
|
||||
"multipart/form-data; boundary=127824672498", [(b"key", b"--127824672498")]
|
||||
)
|
||||
|
||||
boundary = "boundary茅莽"
|
||||
headers = Headers(content_type="multipart/form-data; boundary=" + boundary)
|
||||
result = multipart.encode(headers, data)
|
||||
result = multipart.encode_multipart(
|
||||
"multipart/form-data; boundary=boundary茅莽", data
|
||||
)
|
||||
assert result == b""
|
||||
|
||||
assert multipart.encode_multipart("", data) == b""
|
||||
|
|
|
|||
|
|
@ -434,7 +434,7 @@ class TestRequestUtils:
|
|||
request.headers["Content-Type"] = "multipart/form-data"
|
||||
assert list(request.multipart_form.items()) == []
|
||||
|
||||
with mock.patch("mitmproxy.net.http.multipart.decode") as m:
|
||||
with mock.patch("mitmproxy.net.http.multipart.decode_multipart") as m:
|
||||
m.side_effect = ValueError
|
||||
assert list(request.multipart_form.items()) == []
|
||||
|
||||
|
|
|
|||
Loading…
Add table
Add a link
Reference in a new issue