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
|
# Invariant tested: No matter what random garbage a client throws at us, we
# either successfully parse it, or else throw a RemoteProtocolError, never any
# other error.
import os
import sys
import afl
import h11
def process_all(c):
while True:
event = c.next_event()
if event is h11.NEED_DATA or event is h11.PAUSED:
break
if type(event) is h11.ConnectionClosed:
break
afl.init()
data = sys.stdin.detach().read()
# one big chunk
server1 = h11.Connection(h11.SERVER)
try:
server1.receive_data(data)
process_all(server1)
server1.receive_data(b"")
process_all(server1)
except h11.RemoteProtocolError:
pass
# byte at a time
server2 = h11.Connection(h11.SERVER)
try:
for i in range(len(data)):
server2.receive_data(data[i : i + 1])
process_all(server2)
server2.receive_data(b"")
process_all(server2)
except h11.RemoteProtocolError:
pass
# Suggested by the afl-python docs -- this substantially speeds up fuzzing, at
# the risk of missing bugs that would cause the interpreter to crash on
# exit. h11 is pure python, so I'm pretty sure h11 doesn't have any bugs that
# would cause the interpreter to crash on exit.
os._exit(0)
|