blob: 1509a3549d7b810d0da558f6b29ee9bfa1c77921 (
plain)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
|
import sys
import atheris
with atheris.instrument_imports():
from websockets.exceptions import PayloadTooBig, ProtocolError
from websockets.frames import Frame
from websockets.streams import StreamReader
def test_one_input(data):
fdp = atheris.FuzzedDataProvider(data)
mask = fdp.ConsumeBool()
max_size_enabled = fdp.ConsumeBool()
max_size = fdp.ConsumeInt(4)
payload = fdp.ConsumeBytes(atheris.ALL_REMAINING)
reader = StreamReader()
reader.feed_data(payload)
reader.feed_eof()
parser = Frame.parse(
reader.read_exact,
mask=mask,
max_size=max_size if max_size_enabled else None,
)
try:
next(parser)
except StopIteration as exc:
assert isinstance(exc.value, Frame)
return # input accepted
except (
EOFError, # connection is closed without a full WebSocket frame
UnicodeDecodeError, # frame contains invalid UTF-8
PayloadTooBig, # frame's payload size exceeds ``max_size``
ProtocolError, # frame contains incorrect values
):
return # input rejected with a documented exception
raise RuntimeError("parsing didn't complete")
def main():
atheris.Setup(sys.argv, test_one_input)
atheris.Fuzz()
if __name__ == "__main__":
main()
|
