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
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
|
import atheris
import sys
import os
import tempfile
from configparser import ParsingError
from utils import (
setup_git_environment,
handle_exception,
get_max_filename_length,
)
# Setup the Git environment
setup_git_environment()
from git import Repo, GitCommandError, InvalidGitRepositoryError
def sanitize_input(input_str, max_length=255):
"""Sanitize and truncate inputs to avoid invalid Git operations."""
sanitized = "".join(ch for ch in input_str if ch.isalnum() or ch in ("-", "_", "."))
return sanitized[:max_length]
def TestOneInput(data):
fdp = atheris.FuzzedDataProvider(data)
with tempfile.TemporaryDirectory() as repo_temp_dir:
repo = Repo.init(path=repo_temp_dir)
repo.index.commit("Initial commit")
try:
with tempfile.TemporaryDirectory() as submodule_temp_dir:
sub_repo = Repo.init(submodule_temp_dir, bare=fdp.ConsumeBool())
commit_message = sanitize_input(fdp.ConsumeUnicodeNoSurrogates(fdp.ConsumeIntInRange(1, 512)))
sub_repo.index.commit(commit_message)
submodule_name = sanitize_input(
fdp.ConsumeUnicodeNoSurrogates(
fdp.ConsumeIntInRange(1, get_max_filename_length(repo.working_tree_dir))
)
)
submodule_path = os.path.relpath(
os.path.join(repo.working_tree_dir, submodule_name),
start=repo.working_tree_dir,
)
# Ensure submodule_path is valid
if not submodule_name or submodule_name.startswith("/") or ".." in submodule_name:
return -1 # Reject invalid input so they are not added to the corpus
submodule = repo.create_submodule(submodule_name, submodule_path, url=sub_repo.git_dir)
repo.index.commit("Added submodule")
with submodule.config_writer() as writer:
key_length = fdp.ConsumeIntInRange(1, max(1, fdp.remaining_bytes()))
value_length = fdp.ConsumeIntInRange(1, max(1, fdp.remaining_bytes()))
writer.set_value(
sanitize_input(fdp.ConsumeUnicodeNoSurrogates(key_length)),
sanitize_input(fdp.ConsumeUnicodeNoSurrogates(value_length)),
)
writer.release()
submodule.update(
init=fdp.ConsumeBool(),
dry_run=fdp.ConsumeBool(),
force=fdp.ConsumeBool(),
)
submodule_repo = submodule.module()
new_file_name = sanitize_input(
fdp.ConsumeUnicodeNoSurrogates(
fdp.ConsumeIntInRange(1, get_max_filename_length(submodule_repo.working_tree_dir))
)
)
new_file_path = os.path.join(submodule_repo.working_tree_dir, new_file_name)
with open(new_file_path, "wb") as new_file:
new_file.write(fdp.ConsumeBytes(fdp.ConsumeIntInRange(1, 512)))
submodule_repo.index.add([new_file_path])
submodule_repo.index.commit("Added new file to submodule")
repo.submodule_update(recursive=fdp.ConsumeBool())
submodule_repo.head.reset(
commit="HEAD~1",
working_tree=fdp.ConsumeBool(),
head=fdp.ConsumeBool(),
)
module_option_value, configuration_option_value = fdp.PickValueInList(
[(True, False), (False, True), (True, True)]
)
submodule.remove(
module=module_option_value,
configuration=configuration_option_value,
dry_run=fdp.ConsumeBool(),
force=fdp.ConsumeBool(),
)
repo.index.commit(f"Removed submodule {submodule_name}")
except (
ParsingError,
GitCommandError,
InvalidGitRepositoryError,
FileNotFoundError,
FileExistsError,
IsADirectoryError,
NotADirectoryError,
BrokenPipeError,
PermissionError,
):
return -1
except Exception as e:
return handle_exception(e)
def main():
atheris.instrument_all()
atheris.Setup(sys.argv, TestOneInput)
atheris.Fuzz()
if __name__ == "__main__":
main()
|
