Coverage for /pythoncovmergedfiles/medio/medio/src/gitpython/fuzzing/fuzz-targets/fuzz_submodule.py: 59%

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 atheris 

13import sys 

14import os 

15import tempfile 

16from configparser import ParsingError 

17from utils import ( 

18 setup_git_environment, 

19 handle_exception, 

20 get_max_filename_length, 

21) 

22 

23# Setup the Git environment 

24setup_git_environment() 

25from git import Repo, GitCommandError, InvalidGitRepositoryError 

26 

27 

28def sanitize_input(input_str, max_length=255): 

29 """Sanitize and truncate inputs to avoid invalid Git operations.""" 

30 sanitized = "".join(ch for ch in input_str if ch.isalnum() or ch in ("-", "_", ".")) 

31 return sanitized[:max_length] 

32 

33 

34def TestOneInput(data): 

35 fdp = atheris.FuzzedDataProvider(data) 

36 

37 with tempfile.TemporaryDirectory() as repo_temp_dir: 

38 repo = Repo.init(path=repo_temp_dir) 

39 repo.index.commit("Initial commit") 

40 

41 try: 

42 with tempfile.TemporaryDirectory() as submodule_temp_dir: 

43 sub_repo = Repo.init(submodule_temp_dir, bare=fdp.ConsumeBool()) 

44 commit_message = sanitize_input(fdp.ConsumeUnicodeNoSurrogates(fdp.ConsumeIntInRange(1, 512))) 

45 sub_repo.index.commit(commit_message) 

46 

47 submodule_name = sanitize_input( 

48 fdp.ConsumeUnicodeNoSurrogates( 

49 fdp.ConsumeIntInRange(1, get_max_filename_length(repo.working_tree_dir)) 

50 ) 

51 ) 

52 

53 submodule_path = os.path.relpath( 

54 os.path.join(repo.working_tree_dir, submodule_name), 

55 start=repo.working_tree_dir, 

56 ) 

57 

58 # Ensure submodule_path is valid 

59 if not submodule_name or submodule_name.startswith("/") or ".." in submodule_name: 

60 return -1 # Reject invalid input so they are not added to the corpus 

61 

62 submodule = repo.create_submodule(submodule_name, submodule_path, url=sub_repo.git_dir) 

63 repo.index.commit("Added submodule") 

64 

65 with submodule.config_writer() as writer: 

66 key_length = fdp.ConsumeIntInRange(1, max(1, fdp.remaining_bytes())) 

67 value_length = fdp.ConsumeIntInRange(1, max(1, fdp.remaining_bytes())) 

68 

69 writer.set_value( 

70 sanitize_input(fdp.ConsumeUnicodeNoSurrogates(key_length)), 

71 sanitize_input(fdp.ConsumeUnicodeNoSurrogates(value_length)), 

72 ) 

73 writer.release() 

74 

75 submodule.update( 

76 init=fdp.ConsumeBool(), 

77 dry_run=fdp.ConsumeBool(), 

78 force=fdp.ConsumeBool(), 

79 ) 

80 

81 submodule_repo = submodule.module() 

82 

83 new_file_name = sanitize_input( 

84 fdp.ConsumeUnicodeNoSurrogates( 

85 fdp.ConsumeIntInRange(1, get_max_filename_length(submodule_repo.working_tree_dir)) 

86 ) 

87 ) 

88 new_file_path = os.path.join(submodule_repo.working_tree_dir, new_file_name) 

89 with open(new_file_path, "wb") as new_file: 

90 new_file.write(fdp.ConsumeBytes(fdp.ConsumeIntInRange(1, 512))) 

91 

92 submodule_repo.index.add([new_file_path]) 

93 submodule_repo.index.commit("Added new file to submodule") 

94 

95 repo.submodule_update(recursive=fdp.ConsumeBool()) 

96 submodule_repo.head.reset( 

97 commit="HEAD~1", 

98 working_tree=fdp.ConsumeBool(), 

99 head=fdp.ConsumeBool(), 

100 ) 

101 

102 module_option_value, configuration_option_value = fdp.PickValueInList( 

103 [(True, False), (False, True), (True, True)] 

104 ) 

105 submodule.remove( 

106 module=module_option_value, 

107 configuration=configuration_option_value, 

108 dry_run=fdp.ConsumeBool(), 

109 force=fdp.ConsumeBool(), 

110 ) 

111 repo.index.commit(f"Removed submodule {submodule_name}") 

112 

113 except ( 

114 ParsingError, 

115 GitCommandError, 

116 InvalidGitRepositoryError, 

117 FileNotFoundError, 

118 FileExistsError, 

119 IsADirectoryError, 

120 NotADirectoryError, 

121 BrokenPipeError, 

122 PermissionError, 

123 ): 

124 return -1 

125 except Exception as e: 

126 return handle_exception(e) 

127 

128 

129def main(): 

130 atheris.instrument_all() 

131 atheris.Setup(sys.argv, TestOneInput) 

132 atheris.Fuzz() 

133 

134 

135if __name__ == "__main__": 

136 main()