Coverage for /pythoncovmergedfiles/medio/medio/src/python-multipart/fuzz/fuzz_querystring.py: 62%

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 logging 

13import sys 

14 

15logging.disable(logging.CRITICAL) 

16 

17import atheris 

18from helpers import EnhancedDataProvider 

19 

20with atheris.instrument_imports(): 

21 from python_multipart.exceptions import QuerystringParseError 

22 from python_multipart.multipart import QuerystringParser 

23 

24 

25def _noop_data(data: bytes, start: int, end: int) -> None: 

26 pass 

27 

28 

29def _noop() -> None: 

30 pass 

31 

32 

33def fuzz_single_write(fdp: EnhancedDataProvider) -> None: 

34 strict = fdp.ConsumeBool() 

35 parser = QuerystringParser( 

36 callbacks={ 

37 "on_field_start": _noop, 

38 "on_field_name": _noop_data, 

39 "on_field_data": _noop_data, 

40 "on_field_end": _noop, 

41 "on_end": _noop, 

42 }, 

43 strict_parsing=strict, 

44 ) 

45 parser.write(fdp.ConsumeRandomBytes()) 

46 parser.finalize() 

47 

48 

49def fuzz_chunked_write(fdp: EnhancedDataProvider) -> None: 

50 strict = fdp.ConsumeBool() 

51 num_chunks = fdp.ConsumeIntInRange(1, 8) 

52 parser = QuerystringParser( 

53 callbacks={ 

54 "on_field_start": _noop, 

55 "on_field_name": _noop_data, 

56 "on_field_data": _noop_data, 

57 "on_field_end": _noop, 

58 "on_end": _noop, 

59 }, 

60 strict_parsing=strict, 

61 ) 

62 body = fdp.ConsumeRandomBytes() 

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

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

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

66 parser.finalize() 

67 

68 

69def fuzz_max_size(fdp: EnhancedDataProvider) -> None: 

70 body = fdp.ConsumeRandomBytes() 

71 body_len = max(1, len(body)) 

72 # Pick max_size anywhere from 1 byte up to 2× the body — covers both 

73 # "truncate heavily" and "allow everything through" branches. 

74 max_size = fdp.ConsumeIntInRange(1, body_len * 2) 

75 parser = QuerystringParser( 

76 callbacks={ 

77 "on_field_start": _noop, 

78 "on_field_name": _noop_data, 

79 "on_field_data": _noop_data, 

80 "on_field_end": _noop, 

81 "on_end": _noop, 

82 }, 

83 max_size=max_size, 

84 ) 

85 parser.write(body) 

86 parser.finalize() 

87 

88 

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

90 fdp = EnhancedDataProvider(data) 

91 targets = [fuzz_single_write, fuzz_chunked_write, fuzz_max_size] 

92 target = fdp.PickValueInList(targets) 

93 

94 try: 

95 target(fdp) 

96 except QuerystringParseError: 

97 return 

98 

99 

100def main(): 

101 atheris.Setup(sys.argv, TestOneInput) 

102 atheris.Fuzz() 

103 

104 

105if __name__ == "__main__": 

106 main()