Coverage for /pythoncovmergedfiles/medio/medio/src/python-multipart/fuzz/fuzz_form.py: 57%

1###### Coverage stub 

2import atexit 

3import coverage 

4cov = coverage.coverage(data_file='.coverage', cover_pylib=True) 

5cov.start() 

6# Register an exist handler that will print coverage 

7def exit_handler(): 

8 cov.stop() 

9 cov.save() 

10atexit.register(exit_handler) 

11####### End of coverage stub 

12import io 

13import logging 

14import sys 

15 

16logging.disable(logging.CRITICAL) 

17 

18import atheris 

19from helpers import EnhancedDataProvider 

20 

21with atheris.instrument_imports(): 

22 from python_multipart.exceptions import FormParserError 

23 from python_multipart.multipart import parse_form 

24 

25 

26def _on_field(field) -> None: 

27 pass 

28 

29 

30def _on_file(file) -> None: 

31 pass 

32 

33 

34def parse_octet_stream(fdp: EnhancedDataProvider) -> None: 

35 header = {"Content-Type": "application/octet-stream"} 

36 parse_form(header, io.BytesIO(fdp.ConsumeRandomBytes()), _on_field, _on_file) 

37 

38 

39def parse_url_encoded(fdp: EnhancedDataProvider) -> None: 

40 ct = fdp.PickValueInList(["application/x-url-encoded", "application/x-www-form-urlencoded"]) 

41 header = {"Content-Type": ct} 

42 parse_form(header, io.BytesIO(fdp.ConsumeRandomBytes()), _on_field, _on_file) 

43 

44 

45def parse_multipart_raw(fdp: EnhancedDataProvider) -> None: 

46 # Boundary: 1-70 bytes, no CR/LF (RFC 2046 constraint kept to avoid ValueError). 

47 boundary_len = fdp.ConsumeIntInRange(1, max(1, min(70, fdp.remaining_bytes() // 2))) 

48 boundary = fdp.ConsumeBytes(boundary_len) 

49 boundary = boundary.replace(b"\r", b"-").replace(b"\n", b"-").rstrip(b" \t") or b"B" 

50 header = {"Content-Type": "multipart/form-data; boundary=" + boundary.decode("latin-1")} 

51 body = fdp.ConsumeRandomBytes() 

52 parse_form(header, io.BytesIO(body), _on_field, _on_file) 

53 

54 

55def parse_multipart_with_content_length(fdp: EnhancedDataProvider) -> None: 

56 boundary = b"boundary" 

57 content_length = fdp.ConsumeIntInRange(0, 1024) 

58 header = { 

59 "Content-Type": "multipart/form-data; boundary=boundary", 

60 "Content-Length": str(content_length), 

61 } 

62 body = fdp.ConsumeRandomBytes() 

63 parse_form(header, io.BytesIO(body), _on_field, _on_file) 

64 

65 

66def parse_form_urlencoded_chunked(fdp: EnhancedDataProvider) -> None: 

67 from python_multipart.multipart import create_form_parser 

68 

69 num_chunks = fdp.ConsumeIntInRange(1, 8) 

70 header = {"Content-Type": "application/x-www-form-urlencoded"} 

71 parser = create_form_parser(header, _on_field, _on_file) 

72 body = fdp.ConsumeRandomBytes() 

73 chunk_size = max(1, (len(body) + num_chunks - 1) // num_chunks) 

74 for i in range(0, len(body), chunk_size): 

75 parser.write(body[i : i + chunk_size]) 

76 parser.finalize() 

77 

78def TestOneInput(data: bytes) -> None: 

79 fdp = EnhancedDataProvider(data) 

80 targets = [ 

81 parse_octet_stream, 

82 parse_url_encoded, 

83 parse_multipart_raw, 

84 parse_multipart_with_content_length, 

85 parse_form_urlencoded_chunked, 

86 ] 

87 target = fdp.PickValueInList(targets) 

88 

89 try: 

90 target(fdp) 

91 except FormParserError: 

92 return 

93 

94 

95def main(): 

96 atheris.Setup(sys.argv, TestOneInput) 

97 atheris.Fuzz() 

98 

99 

100if __name__ == "__main__": 

101 main()