werkzeug

1
import pytest
2

3
from werkzeug.datastructures import Headers
4
from werkzeug.sansio.multipart import Data
5
from werkzeug.sansio.multipart import Epilogue
6
from werkzeug.sansio.multipart import Field
7
from werkzeug.sansio.multipart import File
8
from werkzeug.sansio.multipart import MultipartDecoder
9
from werkzeug.sansio.multipart import MultipartEncoder
10
from werkzeug.sansio.multipart import NeedData
11
from werkzeug.sansio.multipart import Preamble
12

13

14
def test_decoder_simple() -> None:
15
    boundary = b"---------------------------9704338192090380615194531385$"
16
    decoder = MultipartDecoder(boundary)
17
    data = """
18
-----------------------------9704338192090380615194531385$
19
Content-Disposition: form-data; name="fname"
20

21
ß∑œß∂ƒå∂
22
-----------------------------9704338192090380615194531385$
23
Content-Disposition: form-data; name="lname"; filename="bob"
24

25
asdasd
26
-----------------------------9704338192090380615194531385$--
27
    """.replace("\n", "\r\n").encode()
28
    decoder.receive_data(data)
29
    decoder.receive_data(None)
30
    events = [decoder.next_event()]
31
    while not isinstance(events[-1], Epilogue):
32
        events.append(decoder.next_event())
33
    assert events == [
34
        Preamble(data=b""),
35
        Field(
36
            name="fname",
37
            headers=Headers([("Content-Disposition", 'form-data; name="fname"')]),
38
        ),
39
        Data(data="ß∑œß∂ƒå∂".encode(), more_data=False),
40
        File(
41
            name="lname",
42
            filename="bob",
43
            headers=Headers(
44
                [("Content-Disposition", 'form-data; name="lname"; filename="bob"')]
45
            ),
46
        ),
47
        Data(data=b"asdasd", more_data=False),
48
        Epilogue(data=b"    "),
49
    ]
50
    encoder = MultipartEncoder(boundary)
51
    result = b""
52
    for event in events:
53
        result += encoder.send_event(event)
54
    assert data == result
55

56

57
@pytest.mark.parametrize(
58
    "data_start",
59
    [
60
        b"A",
61
        b"\n",
62
        b"\r",
63
        b"\r\n",
64
        b"\n\r",
65
        b"A\n",
66
        b"A\r",
67
        b"A\r\n",
68
        b"A\n\r",
69
    ],
70
)
71
@pytest.mark.parametrize("data_end", [b"", b"\r\n--foo"])
72
def test_decoder_data_start_with_different_newline_positions(
73
    data_start: bytes, data_end: bytes
74
) -> None:
75
    boundary = b"foo"
76
    data = (
77
        b"\r\n--foo\r\n"
78
        b'Content-Disposition: form-data; name="test"; filename="testfile"\r\n'
79
        b"Content-Type: application/octet-stream\r\n\r\n"
80
        b"" + data_start + b"\r\nBCDE" + data_end
81
    )
82
    decoder = MultipartDecoder(boundary)
83
    decoder.receive_data(data)
84
    events = [decoder.next_event()]
85
    # We want to check up to data start event
86
    while not isinstance(events[-1], Data):
87
        events.append(decoder.next_event())
88
    expected = data_start if data_end == b"" else data_start + b"\r\nBCDE"
89
    assert events == [
90
        Preamble(data=b""),
91
        File(
92
            name="test",
93
            filename="testfile",
94
            headers=Headers(
95
                [
96
                    (
97
                        "Content-Disposition",
98
                        'form-data; name="test"; filename="testfile"',
99
                    ),
100
                    ("Content-Type", "application/octet-stream"),
101
                ]
102
            ),
103
        ),
104
        Data(data=expected, more_data=True),
105
    ]
106

107

108
def test_chunked_boundaries() -> None:
109
    boundary = b"--boundary"
110
    decoder = MultipartDecoder(boundary)
111
    decoder.receive_data(b"--")
112
    assert isinstance(decoder.next_event(), NeedData)
113
    decoder.receive_data(b"--boundary\r\n")
114
    assert isinstance(decoder.next_event(), Preamble)
115
    decoder.receive_data(b"Content-Disposition: form-data;")
116
    assert isinstance(decoder.next_event(), NeedData)
117
    decoder.receive_data(b'name="fname"\r\n\r\n')
118
    assert isinstance(decoder.next_event(), Field)
119
    decoder.receive_data(b"longer than the boundary")
120
    assert isinstance(decoder.next_event(), Data)
121
    decoder.receive_data(b"also longer, but includes a linebreak\r\n--")
122
    assert isinstance(decoder.next_event(), Data)
123
    assert isinstance(decoder.next_event(), NeedData)
124
    decoder.receive_data(b"--boundary--\r\n")
125
    event = decoder.next_event()
126
    assert isinstance(event, Data)
127
    assert not event.more_data
128
    decoder.receive_data(None)
129
    assert isinstance(decoder.next_event(), Epilogue)
130

131

132
def test_empty_field() -> None:
133
    boundary = b"foo"
134
    decoder = MultipartDecoder(boundary)
135
    data = """
136
--foo
137
Content-Disposition: form-data; name="text"
138
Content-Type: text/plain; charset="UTF-8"
139

140
Some Text
141
--foo
142
Content-Disposition: form-data; name="empty"
143
Content-Type: text/plain; charset="UTF-8"
144

145
--foo--
146
    """.replace("\n", "\r\n").encode()
147
    decoder.receive_data(data)
148
    decoder.receive_data(None)
149
    events = [decoder.next_event()]
150
    while not isinstance(events[-1], Epilogue):
151
        events.append(decoder.next_event())
152
    assert events == [
153
        Preamble(data=b""),
154
        Field(
155
            name="text",
156
            headers=Headers(
157
                [
158
                    ("Content-Disposition", 'form-data; name="text"'),
159
                    ("Content-Type", 'text/plain; charset="UTF-8"'),
160
                ]
161
            ),
162
        ),
163
        Data(data=b"Some Text", more_data=False),
164
        Field(
165
            name="empty",
166
            headers=Headers(
167
                [
168
                    ("Content-Disposition", 'form-data; name="empty"'),
169
                    ("Content-Type", 'text/plain; charset="UTF-8"'),
170
                ]
171
            ),
172
        ),
173
        Data(data=b"", more_data=False),
174
        Epilogue(data=b"    "),
175
    ]
176
    encoder = MultipartEncoder(boundary)
177
    result = b""
178
    for event in events:
179
        result += encoder.send_event(event)
180
    assert data == result
181