We are currently solving the following issue within our repository. Here is the issue text:
--- BEGIN ISSUE ---
Make Centipede handle custom ASAN options that libFuzzer supports
See https://bugs.chromium.org/p/chromium/issues/detail?id=1451080#c4

--- END ISSUE ---

Below are some code segments, each from a relevant file. One or more of these files may contain bugs.
--- BEGIN FILES ---
Path: `src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py`
Content:
```
1 # Copyright 2022 Google LLC
2 #
3 # Licensed under the Apache License, Version 2.0 (the "License");
4 # you may not use this file except in compliance with the License.
5 # You may obtain a copy of the License at
6 #
7 #      http://www.apache.org/licenses/LICENSE-2.0
8 #
9 # Unless required by applicable law or agreed to in writing, software
10 # distributed under the License is distributed on an "AS IS" BASIS,
11 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
12 # See the License for the specific language governing permissions and
13 # limitations under the License.
14 """Centipede engine interface."""
15 
16 from collections import namedtuple
17 import os
18 import pathlib
19 import re
20 import shutil
21 
22 from clusterfuzz._internal.bot.fuzzers import dictionary_manager
23 from clusterfuzz._internal.bot.fuzzers import engine_common
24 from clusterfuzz._internal.bot.fuzzers import utils as fuzzer_utils
25 from clusterfuzz._internal.metrics import logs
26 from clusterfuzz._internal.system import environment
27 from clusterfuzz._internal.system import new_process
28 from clusterfuzz.fuzz import engine
29 
30 _CLEAN_EXIT_SECS = 10
31 _SERVER_COUNT = 1
32 _RSS_LIMIT = 4096
33 _ADDRESS_SPACE_LIMIT = 4096
34 _TIMEOUT_PER_INPUT_FUZZ = 25
35 _TIMEOUT_PER_INPUT_REPR = 60
36 _DEFAULT_ARGUMENTS = [
37     '--exit_on_crash=1',
38     f'--fork_server={_SERVER_COUNT}',
39     f'--rss_limit_mb={_RSS_LIMIT}',
40     f'--address_space_limit_mb={_ADDRESS_SPACE_LIMIT}',
41 ]
42 
43 CRASH_REGEX = re.compile(r'[sS]aving input to:?\s*(.*)')
44 _CRASH_LOG_PREFIX = 'CRASH LOG: '
45 TargetBinaries = namedtuple('TargetBinaries', ['unsanitized', 'sanitized'])
46 
47 
48 class CentipedeError(Exception):
49   """Base exception class."""
50 
51 
52 def _get_runner(target_path):
53   """Gets the Centipede runner."""
54   centipede_path = pathlib.Path(target_path).parent / 'centipede'
55   if not centipede_path.exists():
56     raise CentipedeError('Centipede not found in build')
57 
58   centipede_path = str(centipede_path)
59   if environment.get_value('USE_UNSHARE'):
60     return new_process.UnicodeModifierRunner(centipede_path)
61   return new_process.UnicodeProcessRunner(centipede_path)
62 
63 
64 def _get_reproducer_path(log, reproducers_dir):
65   """Gets the reproducer path, if any."""
66   crash_match = CRASH_REGEX.search(log)
67   if not crash_match:
68     return None
69   tmp_crash_path = pathlib.Path(crash_match.group(1))
70   crash_path = pathlib.Path(reproducers_dir) / tmp_crash_path.name
71   shutil.copy(tmp_crash_path, crash_path)
72   return crash_path
73 
74 
75 class Engine(engine.Engine):
76   """Centipede engine implementation."""
77 
78   @property
79   def name(self):
80     return 'centipede'
81 
82   # pylint: disable=unused-argument
83   def prepare(self, corpus_dir, target_path, build_dir):
84     """Prepares for a fuzzing session, by generating options.
85 
86     Args:
87       corpus_dir: The main corpus directory.
88       target_path: Path to the target.
89       build_dir: Path to the build directory.
90 
91     Returns:
92       A FuzzOptions object.
93    """
94     arguments = []
95     dict_path = pathlib.Path(
96         dictionary_manager.get_default_dictionary_path(target_path))
97     if dict_path.exists():
98       arguments.append(f'--dictionary={dict_path}')
99 
100     # Directory workdir saves:
101     # 1. Centipede-readable corpus file;
102     # 2. Centipede-readable feature file;
103     # 3. Crash reproducing inputs.
104     workdir = self._create_temp_dir('workdir')
105     arguments.append(f'--workdir={workdir}')
106 
107     # Directory corpus_dir saves the corpus files required by ClusterFuzz.
108     arguments.append(f'--corpus_dir={corpus_dir}')
109 
110     target_binaries = self._get_binary_paths(target_path)
111     if target_binaries.unsanitized is None:
112       # Assuming the only binary is always sanitized (e.g., from Chrome).
113       arguments.append(f'--binary={target_binaries.sanitized}')
114       logs.log_warn('Unable to find unsanitized target binary.')
115     else:
116       arguments.append(f'--binary={target_binaries.unsanitized}')
117       arguments.append(f'--extra_binaries={target_binaries.sanitized}')
118 
119     arguments.append(f'--timeout_per_input={_TIMEOUT_PER_INPUT_FUZZ}')
120 
121     arguments.extend(_DEFAULT_ARGUMENTS)
122 
123     return engine.FuzzOptions(corpus_dir, arguments, {})
124 
125   def _get_binary_paths(self, target_path):
126     """Gets the paths to the main and auxiliary binaries based on |target_path|
127     Args:
128       target_path: Path to the main target in a string.
129 
130     Returns:
131       A named tuple containing paths to both target binaries as pathlib.Path.
132     """
133     # Centipede expects one or two target binaries:
134     # |-------------------------------------------------------|
135     # |            | main target path | auxiliary target path |
136     # |-------------------------------------------------------|
137     # | 1 binary   | sanitized        | -                     |
138     # |-------------------------------------------------------|
139     # | 2 binaries | unsanitized      | sanitized             |
140     # |-------------------------------------------------------|
141 
142     main_target_path = pathlib.Path(target_path)
143     auxiliary_target_path = self._get_auxiliary_target_path(target_path)
144 
145     if main_target_path.exists() and auxiliary_target_path.exists():
146       # 2 binaries were provided.
147       target_binaries = TargetBinaries(main_target_path, auxiliary_target_path)
148     elif main_target_path.exists():
149       # 1 binary was provided.
150       target_binaries = TargetBinaries(None, main_target_path)
151     else:
152       assert not auxiliary_target_path.exists()
153       raise RuntimeError('No fuzz target: Centipede cannot find main target '
154                          f'{main_target_path}, or auxiliary target '
155                          f'{auxiliary_target_path}.')
156 
157     return target_binaries
158 
159   def _get_auxiliary_target_path(self, target_path):
160     """Gets the auxiliary target path based on the main |target_path|.
161     When exists, it points to the sanitized binary, which is required by fuzzing
162     (as an auxiliary) and crash reproduction.
163 
164     Args:
165       target_path: Path to the main target in a string.
166 
167     Returns:
168       Path to the auxiliary binary as a pathlib.Path.
169     """
170     # Assuming they will be in child dirs named by fuzzer_utils.EXTRA_BUILD_DIR.
171     build_dir = environment.get_value('BUILD_DIR')
172     auxiliary_target_name = pathlib.Path(target_path).name
173     auxiliary_target_path = pathlib.Path(
174         build_dir, fuzzer_utils.EXTRA_BUILD_DIR, auxiliary_target_name)
175     return auxiliary_target_path
176 
177   def fuzz(self, target_path, options, reproducers_dir, max_time):  # pylint: disable=unused-argument
178     """Runs a fuzz session.
179 
180     Args:
181       target_path: Path to the target.
182       options: The FuzzOptions object returned by prepare().
183       reproducers_dir: The directory to put reproducers in when crashes
184           are found.
185       max_time: Maximum allowed time for the fuzzing to run.
186 
187    Returns:
188       A FuzzResult object.
189     """
190     runner = _get_runner(target_path)
191     timeout = max_time + _CLEAN_EXIT_SECS
192     fuzz_result = runner.run_and_wait(
193         additional_args=options.arguments, timeout=timeout)
194     fuzz_result.output = Engine.trim_logs(fuzz_result.output)
195 
196     reproducer_path = _get_reproducer_path(fuzz_result.output, reproducers_dir)
197     crashes = []
198     if reproducer_path:
199       crashes.append(
200           engine.Crash(
201               str(reproducer_path), fuzz_result.output, [],
202               int(fuzz_result.time_executed)))
203 
204     # Stats report is not available in Centipede yet.
205     stats = None
206     return engine.FuzzResult(fuzz_result.output, fuzz_result.command, crashes,
207                              stats, fuzz_result.time_executed)
208 
209   @staticmethod
210   def trim_logs(fuzz_log):
211     """ Strips the 'CRASH LOG:' prefix that breaks stacktrace parsing.
212 
213     Args:
214       fuzz_result: The ProcessResult returned by running fuzzer binary.
215     """
216     trimmed_log_lines = [
217         line[len(_CRASH_LOG_PREFIX):]
218         if line.startswith(_CRASH_LOG_PREFIX) else line
219         for line in fuzz_log.splitlines()
220     ]
221     return '\n'.join(trimmed_log_lines)
222 
223   def reproduce(self, target_path, input_path, arguments, max_time):  # pylint: disable=unused-argument
224     """Reproduces a crash given an input.
225 
226     Args:
227       target_path: Path to the target.
228       input_path: Path to the reproducer input.
229       arguments: Additional arguments needed for reproduction.
230       max_time: Maximum allowed time for the reproduction.
231 
232     Returns:
233       A ReproduceResult.
234     """
235     target_binaries = self._get_binary_paths(target_path)
236     sanitized_target = str(target_binaries.sanitized)
237 
238     existing_runner_flags = os.environ.get('CENTIPEDE_RUNNER_FLAGS')
239     if not existing_runner_flags:
240       os.environ['CENTIPEDE_RUNNER_FLAGS'] = (
241           f':rss_limit_mb={_RSS_LIMIT}'
242           f':timeout_per_input={_TIMEOUT_PER_INPUT_REPR}:')
243 
244     runner = new_process.UnicodeProcessRunner(sanitized_target, [input_path])
245     result = runner.run_and_wait(timeout=max_time)
246 
247     if existing_runner_flags:
248       os.environ['CENTIPEDE_RUNNER_FLAGS'] = existing_runner_flags
249     else:
250       os.unsetenv('CENTIPEDE_RUNNER_FLAGS')
251     result.output = Engine.trim_logs(result.output)
252 
253     return engine.ReproduceResult(result.command, result.return_code,
254                                   result.time_executed, result.output)
255 
256   def _create_temp_dir(self, name):
257     """Creates temporary directory for fuzzing."""
258     new_directory = pathlib.Path(fuzzer_utils.get_temp_dir(), name)
259     engine_common.recreate_directory(new_directory)
260     return new_directory
261 
262   def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,
263                       reproducers_dir, max_time):
264     """Runs corpus minimization.
265     Args:
266       target_path: Path to the target.
267       arguments: Additional arguments needed for corpus minimization.
268       input_dirs: Input corpora.
269       output_dir: Output directory to place minimized corpus.
270       reproducers_dir: The directory to put reproducers in when crashes are
271           found.
272       max_time: Maximum allowed time for the minimization.
273 
274     Returns:
275       A FuzzResult object.
276     """
277     raise NotImplementedError
278 
279   def minimize_testcase(self, target_path, arguments, input_path, output_path,
280                         max_time):
281     """Minimizes a testcase.
282     Args:
283       target_path: Path to the target.
284       arguments: Additional arguments needed for testcase minimization.
285       input_path: Path to the reproducer input.
286       output_path: Path to the minimized output.
287       max_time: Maximum allowed time for the minimization.
288     Returns:
289       A ReproduceResult.
290     Raises:
291       TimeoutError: If the testcase minimization exceeds max_time.
292     """
293     raise NotImplementedError
294 
295   def cleanse(self, target_path, arguments, input_path, output_path, max_time):
296     """Cleanses a testcase.
297     Args:
298       target_path: Path to the target.
299       arguments: Additional arguments needed for testcase cleanse.
300       input_path: Path to the reproducer input.
301       output_path: Path to the cleansed output.
302       max_time: Maximum allowed time for the cleanse.
303     Returns:
304       A ReproduceResult.
305     Raises:
306       TimeoutError: If the cleanse exceeds max_time.
307     """
308     raise NotImplementedError
309
```
--- END FILES ---

Please first localize the bug based on the issue statement, and then generate a patch according to the `git diff` format fenced by three backticks.

Here is an example:

```diff
diff --git a/examples/server_async.py b/examples/server_async.py
--- a/examples/server_async.py
+++ b/examples/server_async.py
@@ -313,4 +313,4 @@
 
 
 if __name__ == "__main__":
-    asyncio.run(run_async_server("."), debug=True)
+    asyncio.run(run_async_server(), debug=True)
diff --git a/examples/server_sync.py b/examples/server_sync.py
--- a/examples/server_sync.py
+++ b/examples/server_sync.py
@@ -313,5 +313,5 @@
 
 
 if __name__ == "__main__":
-    server = run_sync_server(".")
+    server = run_sync_server()
     server.shutdown()

```

diff --git a/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py b/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py
--- a/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py
+++ b/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py
@@ -72,6 +72,15 @@
   return crash_path
 
 
+def _set_sanitizer_options(fuzzer_path):
+  """Sets sanitizer options based on .options file overrides."""
+  engine_common.process_sanitizer_options_overrides(fuzzer_path)
+  sanitizer_options_var = environment.get_current_memory_tool_var()
+  sanitizer_options = environment.get_memory_tool_options(
+      sanitizer_options_var, {})
+  environment.set_memory_tool_options(sanitizer_options_var, sanitizer_options)
+
+
 class Engine(engine.Engine):
   """Centipede engine implementation."""
 
@@ -188,6 +197,7 @@
       A FuzzResult object.
     """
     runner = _get_runner(target_path)
+    _set_sanitizer_options(target_path)
     timeout = max_time + _CLEAN_EXIT_SECS
     fuzz_result = runner.run_and_wait(
         additional_args=options.arguments, timeout=timeout)
@@ -232,6 +242,7 @@
     Returns:
       A ReproduceResult.
     """
+    _set_sanitizer_options(target_path)
     target_binaries = self._get_binary_paths(target_path)
     sanitized_target = str(target_binaries.sanitized)

{"golden_diff": "diff --git a/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py b/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py\n--- a/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py\n+++ b/src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py\n@@ -72,6 +72,15 @@\n   return crash_path\n \n \n+def _set_sanitizer_options(fuzzer_path):\n+  \"\"\"Sets sanitizer options based on .options file overrides.\"\"\"\n+  engine_common.process_sanitizer_options_overrides(fuzzer_path)\n+  sanitizer_options_var = environment.get_current_memory_tool_var()\n+  sanitizer_options = environment.get_memory_tool_options(\n+      sanitizer_options_var, {})\n+  environment.set_memory_tool_options(sanitizer_options_var, sanitizer_options)\n+\n+\n class Engine(engine.Engine):\n   \"\"\"Centipede engine implementation.\"\"\"\n \n@@ -188,6 +197,7 @@\n       A FuzzResult object.\n     \"\"\"\n     runner = _get_runner(target_path)\n+    _set_sanitizer_options(target_path)\n     timeout = max_time + _CLEAN_EXIT_SECS\n     fuzz_result = runner.run_and_wait(\n         additional_args=options.arguments, timeout=timeout)\n@@ -232,6 +242,7 @@\n     Returns:\n       A ReproduceResult.\n     \"\"\"\n+    _set_sanitizer_options(target_path)\n     target_binaries = self._get_binary_paths(target_path)\n     sanitized_target = str(target_binaries.sanitized)\n", "issue": "Make Centipede handle custom ASAN options that libFuzzer supports\nSee https://bugs.chromium.org/p/chromium/issues/detail?id=1451080#c4\n", "before_files": [{"content": "# Copyright 2022 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Centipede engine interface.\"\"\"\n\nfrom collections import namedtuple\nimport os\nimport pathlib\nimport re\nimport shutil\n\nfrom clusterfuzz._internal.bot.fuzzers import dictionary_manager\nfrom clusterfuzz._internal.bot.fuzzers import engine_common\nfrom clusterfuzz._internal.bot.fuzzers import utils as fuzzer_utils\nfrom clusterfuzz._internal.metrics import logs\nfrom clusterfuzz._internal.system import environment\nfrom clusterfuzz._internal.system import new_process\nfrom clusterfuzz.fuzz import engine\n\n_CLEAN_EXIT_SECS = 10\n_SERVER_COUNT = 1\n_RSS_LIMIT = 4096\n_ADDRESS_SPACE_LIMIT = 4096\n_TIMEOUT_PER_INPUT_FUZZ = 25\n_TIMEOUT_PER_INPUT_REPR = 60\n_DEFAULT_ARGUMENTS = [\n    '--exit_on_crash=1',\n    f'--fork_server={_SERVER_COUNT}',\n    f'--rss_limit_mb={_RSS_LIMIT}',\n    f'--address_space_limit_mb={_ADDRESS_SPACE_LIMIT}',\n]\n\nCRASH_REGEX = re.compile(r'[sS]aving input to:?\\s*(.*)')\n_CRASH_LOG_PREFIX = 'CRASH LOG: '\nTargetBinaries = namedtuple('TargetBinaries', ['unsanitized', 'sanitized'])\n\n\nclass CentipedeError(Exception):\n  \"\"\"Base exception class.\"\"\"\n\n\ndef _get_runner(target_path):\n  \"\"\"Gets the Centipede runner.\"\"\"\n  centipede_path = pathlib.Path(target_path).parent / 'centipede'\n  if not centipede_path.exists():\n    raise CentipedeError('Centipede not found in build')\n\n  centipede_path = str(centipede_path)\n  if environment.get_value('USE_UNSHARE'):\n    return new_process.UnicodeModifierRunner(centipede_path)\n  return new_process.UnicodeProcessRunner(centipede_path)\n\n\ndef _get_reproducer_path(log, reproducers_dir):\n  \"\"\"Gets the reproducer path, if any.\"\"\"\n  crash_match = CRASH_REGEX.search(log)\n  if not crash_match:\n    return None\n  tmp_crash_path = pathlib.Path(crash_match.group(1))\n  crash_path = pathlib.Path(reproducers_dir) / tmp_crash_path.name\n  shutil.copy(tmp_crash_path, crash_path)\n  return crash_path\n\n\nclass Engine(engine.Engine):\n  \"\"\"Centipede engine implementation.\"\"\"\n\n  @property\n  def name(self):\n    return 'centipede'\n\n  # pylint: disable=unused-argument\n  def prepare(self, corpus_dir, target_path, build_dir):\n    \"\"\"Prepares for a fuzzing session, by generating options.\n\n    Args:\n      corpus_dir: The main corpus directory.\n      target_path: Path to the target.\n      build_dir: Path to the build directory.\n\n    Returns:\n      A FuzzOptions object.\n   \"\"\"\n    arguments = []\n    dict_path = pathlib.Path(\n        dictionary_manager.get_default_dictionary_path(target_path))\n    if dict_path.exists():\n      arguments.append(f'--dictionary={dict_path}')\n\n    # Directory workdir saves:\n    # 1. Centipede-readable corpus file;\n    # 2. Centipede-readable feature file;\n    # 3. Crash reproducing inputs.\n    workdir = self._create_temp_dir('workdir')\n    arguments.append(f'--workdir={workdir}')\n\n    # Directory corpus_dir saves the corpus files required by ClusterFuzz.\n    arguments.append(f'--corpus_dir={corpus_dir}')\n\n    target_binaries = self._get_binary_paths(target_path)\n    if target_binaries.unsanitized is None:\n      # Assuming the only binary is always sanitized (e.g., from Chrome).\n      arguments.append(f'--binary={target_binaries.sanitized}')\n      logs.log_warn('Unable to find unsanitized target binary.')\n    else:\n      arguments.append(f'--binary={target_binaries.unsanitized}')\n      arguments.append(f'--extra_binaries={target_binaries.sanitized}')\n\n    arguments.append(f'--timeout_per_input={_TIMEOUT_PER_INPUT_FUZZ}')\n\n    arguments.extend(_DEFAULT_ARGUMENTS)\n\n    return engine.FuzzOptions(corpus_dir, arguments, {})\n\n  def _get_binary_paths(self, target_path):\n    \"\"\"Gets the paths to the main and auxiliary binaries based on |target_path|\n    Args:\n      target_path: Path to the main target in a string.\n\n    Returns:\n      A named tuple containing paths to both target binaries as pathlib.Path.\n    \"\"\"\n    # Centipede expects one or two target binaries:\n    # |-------------------------------------------------------|\n    # |            | main target path | auxiliary target path |\n    # |-------------------------------------------------------|\n    # | 1 binary   | sanitized        | -                     |\n    # |-------------------------------------------------------|\n    # | 2 binaries | unsanitized      | sanitized             |\n    # |-------------------------------------------------------|\n\n    main_target_path = pathlib.Path(target_path)\n    auxiliary_target_path = self._get_auxiliary_target_path(target_path)\n\n    if main_target_path.exists() and auxiliary_target_path.exists():\n      # 2 binaries were provided.\n      target_binaries = TargetBinaries(main_target_path, auxiliary_target_path)\n    elif main_target_path.exists():\n      # 1 binary was provided.\n      target_binaries = TargetBinaries(None, main_target_path)\n    else:\n      assert not auxiliary_target_path.exists()\n      raise RuntimeError('No fuzz target: Centipede cannot find main target '\n                         f'{main_target_path}, or auxiliary target '\n                         f'{auxiliary_target_path}.')\n\n    return target_binaries\n\n  def _get_auxiliary_target_path(self, target_path):\n    \"\"\"Gets the auxiliary target path based on the main |target_path|.\n    When exists, it points to the sanitized binary, which is required by fuzzing\n    (as an auxiliary) and crash reproduction.\n\n    Args:\n      target_path: Path to the main target in a string.\n\n    Returns:\n      Path to the auxiliary binary as a pathlib.Path.\n    \"\"\"\n    # Assuming they will be in child dirs named by fuzzer_utils.EXTRA_BUILD_DIR.\n    build_dir = environment.get_value('BUILD_DIR')\n    auxiliary_target_name = pathlib.Path(target_path).name\n    auxiliary_target_path = pathlib.Path(\n        build_dir, fuzzer_utils.EXTRA_BUILD_DIR, auxiliary_target_name)\n    return auxiliary_target_path\n\n  def fuzz(self, target_path, options, reproducers_dir, max_time):  # pylint: disable=unused-argument\n    \"\"\"Runs a fuzz session.\n\n    Args:\n      target_path: Path to the target.\n      options: The FuzzOptions object returned by prepare().\n      reproducers_dir: The directory to put reproducers in when crashes\n          are found.\n      max_time: Maximum allowed time for the fuzzing to run.\n\n   Returns:\n      A FuzzResult object.\n    \"\"\"\n    runner = _get_runner(target_path)\n    timeout = max_time + _CLEAN_EXIT_SECS\n    fuzz_result = runner.run_and_wait(\n        additional_args=options.arguments, timeout=timeout)\n    fuzz_result.output = Engine.trim_logs(fuzz_result.output)\n\n    reproducer_path = _get_reproducer_path(fuzz_result.output, reproducers_dir)\n    crashes = []\n    if reproducer_path:\n      crashes.append(\n          engine.Crash(\n              str(reproducer_path), fuzz_result.output, [],\n              int(fuzz_result.time_executed)))\n\n    # Stats report is not available in Centipede yet.\n    stats = None\n    return engine.FuzzResult(fuzz_result.output, fuzz_result.command, crashes,\n                             stats, fuzz_result.time_executed)\n\n  @staticmethod\n  def trim_logs(fuzz_log):\n    \"\"\" Strips the 'CRASH LOG:' prefix that breaks stacktrace parsing.\n\n    Args:\n      fuzz_result: The ProcessResult returned by running fuzzer binary.\n    \"\"\"\n    trimmed_log_lines = [\n        line[len(_CRASH_LOG_PREFIX):]\n        if line.startswith(_CRASH_LOG_PREFIX) else line\n        for line in fuzz_log.splitlines()\n    ]\n    return '\\n'.join(trimmed_log_lines)\n\n  def reproduce(self, target_path, input_path, arguments, max_time):  # pylint: disable=unused-argument\n    \"\"\"Reproduces a crash given an input.\n\n    Args:\n      target_path: Path to the target.\n      input_path: Path to the reproducer input.\n      arguments: Additional arguments needed for reproduction.\n      max_time: Maximum allowed time for the reproduction.\n\n    Returns:\n      A ReproduceResult.\n    \"\"\"\n    target_binaries = self._get_binary_paths(target_path)\n    sanitized_target = str(target_binaries.sanitized)\n\n    existing_runner_flags = os.environ.get('CENTIPEDE_RUNNER_FLAGS')\n    if not existing_runner_flags:\n      os.environ['CENTIPEDE_RUNNER_FLAGS'] = (\n          f':rss_limit_mb={_RSS_LIMIT}'\n          f':timeout_per_input={_TIMEOUT_PER_INPUT_REPR}:')\n\n    runner = new_process.UnicodeProcessRunner(sanitized_target, [input_path])\n    result = runner.run_and_wait(timeout=max_time)\n\n    if existing_runner_flags:\n      os.environ['CENTIPEDE_RUNNER_FLAGS'] = existing_runner_flags\n    else:\n      os.unsetenv('CENTIPEDE_RUNNER_FLAGS')\n    result.output = Engine.trim_logs(result.output)\n\n    return engine.ReproduceResult(result.command, result.return_code,\n                                  result.time_executed, result.output)\n\n  def _create_temp_dir(self, name):\n    \"\"\"Creates temporary directory for fuzzing.\"\"\"\n    new_directory = pathlib.Path(fuzzer_utils.get_temp_dir(), name)\n    engine_common.recreate_directory(new_directory)\n    return new_directory\n\n  def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,\n                      reproducers_dir, max_time):\n    \"\"\"Runs corpus minimization.\n    Args:\n      target_path: Path to the target.\n      arguments: Additional arguments needed for corpus minimization.\n      input_dirs: Input corpora.\n      output_dir: Output directory to place minimized corpus.\n      reproducers_dir: The directory to put reproducers in when crashes are\n          found.\n      max_time: Maximum allowed time for the minimization.\n\n    Returns:\n      A FuzzResult object.\n    \"\"\"\n    raise NotImplementedError\n\n  def minimize_testcase(self, target_path, arguments, input_path, output_path,\n                        max_time):\n    \"\"\"Minimizes a testcase.\n    Args:\n      target_path: Path to the target.\n      arguments: Additional arguments needed for testcase minimization.\n      input_path: Path to the reproducer input.\n      output_path: Path to the minimized output.\n      max_time: Maximum allowed time for the minimization.\n    Returns:\n      A ReproduceResult.\n    Raises:\n      TimeoutError: If the testcase minimization exceeds max_time.\n    \"\"\"\n    raise NotImplementedError\n\n  def cleanse(self, target_path, arguments, input_path, output_path, max_time):\n    \"\"\"Cleanses a testcase.\n    Args:\n      target_path: Path to the target.\n      arguments: Additional arguments needed for testcase cleanse.\n      input_path: Path to the reproducer input.\n      output_path: Path to the cleansed output.\n      max_time: Maximum allowed time for the cleanse.\n    Returns:\n      A ReproduceResult.\n    Raises:\n      TimeoutError: If the cleanse exceeds max_time.\n    \"\"\"\n    raise NotImplementedError\n", "path": "src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py"}], "after_files": [{"content": "# Copyright 2022 Google LLC\n#\n# Licensed under the Apache License, Version 2.0 (the \"License\");\n# you may not use this file except in compliance with the License.\n# You may obtain a copy of the License at\n#\n#      http://www.apache.org/licenses/LICENSE-2.0\n#\n# Unless required by applicable law or agreed to in writing, software\n# distributed under the License is distributed on an \"AS IS\" BASIS,\n# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.\n# See the License for the specific language governing permissions and\n# limitations under the License.\n\"\"\"Centipede engine interface.\"\"\"\n\nfrom collections import namedtuple\nimport os\nimport pathlib\nimport re\nimport shutil\n\nfrom clusterfuzz._internal.bot.fuzzers import dictionary_manager\nfrom clusterfuzz._internal.bot.fuzzers import engine_common\nfrom clusterfuzz._internal.bot.fuzzers import utils as fuzzer_utils\nfrom clusterfuzz._internal.metrics import logs\nfrom clusterfuzz._internal.system import environment\nfrom clusterfuzz._internal.system import new_process\nfrom clusterfuzz.fuzz import engine\n\n_CLEAN_EXIT_SECS = 10\n_SERVER_COUNT = 1\n_RSS_LIMIT = 4096\n_ADDRESS_SPACE_LIMIT = 4096\n_TIMEOUT_PER_INPUT_FUZZ = 25\n_TIMEOUT_PER_INPUT_REPR = 60\n_DEFAULT_ARGUMENTS = [\n    '--exit_on_crash=1',\n    f'--fork_server={_SERVER_COUNT}',\n    f'--rss_limit_mb={_RSS_LIMIT}',\n    f'--address_space_limit_mb={_ADDRESS_SPACE_LIMIT}',\n]\n\nCRASH_REGEX = re.compile(r'[sS]aving input to:?\\s*(.*)')\n_CRASH_LOG_PREFIX = 'CRASH LOG: '\nTargetBinaries = namedtuple('TargetBinaries', ['unsanitized', 'sanitized'])\n\n\nclass CentipedeError(Exception):\n  \"\"\"Base exception class.\"\"\"\n\n\ndef _get_runner(target_path):\n  \"\"\"Gets the Centipede runner.\"\"\"\n  centipede_path = pathlib.Path(target_path).parent / 'centipede'\n  if not centipede_path.exists():\n    raise CentipedeError('Centipede not found in build')\n\n  centipede_path = str(centipede_path)\n  if environment.get_value('USE_UNSHARE'):\n    return new_process.UnicodeModifierRunner(centipede_path)\n  return new_process.UnicodeProcessRunner(centipede_path)\n\n\ndef _get_reproducer_path(log, reproducers_dir):\n  \"\"\"Gets the reproducer path, if any.\"\"\"\n  crash_match = CRASH_REGEX.search(log)\n  if not crash_match:\n    return None\n  tmp_crash_path = pathlib.Path(crash_match.group(1))\n  crash_path = pathlib.Path(reproducers_dir) / tmp_crash_path.name\n  shutil.copy(tmp_crash_path, crash_path)\n  return crash_path\n\n\ndef _set_sanitizer_options(fuzzer_path):\n  \"\"\"Sets sanitizer options based on .options file overrides.\"\"\"\n  engine_common.process_sanitizer_options_overrides(fuzzer_path)\n  sanitizer_options_var = environment.get_current_memory_tool_var()\n  sanitizer_options = environment.get_memory_tool_options(\n      sanitizer_options_var, {})\n  environment.set_memory_tool_options(sanitizer_options_var, sanitizer_options)\n\n\nclass Engine(engine.Engine):\n  \"\"\"Centipede engine implementation.\"\"\"\n\n  @property\n  def name(self):\n    return 'centipede'\n\n  # pylint: disable=unused-argument\n  def prepare(self, corpus_dir, target_path, build_dir):\n    \"\"\"Prepares for a fuzzing session, by generating options.\n\n    Args:\n      corpus_dir: The main corpus directory.\n      target_path: Path to the target.\n      build_dir: Path to the build directory.\n\n    Returns:\n      A FuzzOptions object.\n   \"\"\"\n    arguments = []\n    dict_path = pathlib.Path(\n        dictionary_manager.get_default_dictionary_path(target_path))\n    if dict_path.exists():\n      arguments.append(f'--dictionary={dict_path}')\n\n    # Directory workdir saves:\n    # 1. Centipede-readable corpus file;\n    # 2. Centipede-readable feature file;\n    # 3. Crash reproducing inputs.\n    workdir = self._create_temp_dir('workdir')\n    arguments.append(f'--workdir={workdir}')\n\n    # Directory corpus_dir saves the corpus files required by ClusterFuzz.\n    arguments.append(f'--corpus_dir={corpus_dir}')\n\n    target_binaries = self._get_binary_paths(target_path)\n    if target_binaries.unsanitized is None:\n      # Assuming the only binary is always sanitized (e.g., from Chrome).\n      arguments.append(f'--binary={target_binaries.sanitized}')\n      logs.log_warn('Unable to find unsanitized target binary.')\n    else:\n      arguments.append(f'--binary={target_binaries.unsanitized}')\n      arguments.append(f'--extra_binaries={target_binaries.sanitized}')\n\n    arguments.append(f'--timeout_per_input={_TIMEOUT_PER_INPUT_FUZZ}')\n\n    arguments.extend(_DEFAULT_ARGUMENTS)\n\n    return engine.FuzzOptions(corpus_dir, arguments, {})\n\n  def _get_binary_paths(self, target_path):\n    \"\"\"Gets the paths to the main and auxiliary binaries based on |target_path|\n    Args:\n      target_path: Path to the main target in a string.\n\n    Returns:\n      A named tuple containing paths to both target binaries as pathlib.Path.\n    \"\"\"\n    # Centipede expects one or two target binaries:\n    # |-------------------------------------------------------|\n    # |            | main target path | auxiliary target path |\n    # |-------------------------------------------------------|\n    # | 1 binary   | sanitized        | -                     |\n    # |-------------------------------------------------------|\n    # | 2 binaries | unsanitized      | sanitized             |\n    # |-------------------------------------------------------|\n\n    main_target_path = pathlib.Path(target_path)\n    auxiliary_target_path = self._get_auxiliary_target_path(target_path)\n\n    if main_target_path.exists() and auxiliary_target_path.exists():\n      # 2 binaries were provided.\n      target_binaries = TargetBinaries(main_target_path, auxiliary_target_path)\n    elif main_target_path.exists():\n      # 1 binary was provided.\n      target_binaries = TargetBinaries(None, main_target_path)\n    else:\n      assert not auxiliary_target_path.exists()\n      raise RuntimeError('No fuzz target: Centipede cannot find main target '\n                         f'{main_target_path}, or auxiliary target '\n                         f'{auxiliary_target_path}.')\n\n    return target_binaries\n\n  def _get_auxiliary_target_path(self, target_path):\n    \"\"\"Gets the auxiliary target path based on the main |target_path|.\n    When exists, it points to the sanitized binary, which is required by fuzzing\n    (as an auxiliary) and crash reproduction.\n\n    Args:\n      target_path: Path to the main target in a string.\n\n    Returns:\n      Path to the auxiliary binary as a pathlib.Path.\n    \"\"\"\n    # Assuming they will be in child dirs named by fuzzer_utils.EXTRA_BUILD_DIR.\n    build_dir = environment.get_value('BUILD_DIR')\n    auxiliary_target_name = pathlib.Path(target_path).name\n    auxiliary_target_path = pathlib.Path(\n        build_dir, fuzzer_utils.EXTRA_BUILD_DIR, auxiliary_target_name)\n    return auxiliary_target_path\n\n  def fuzz(self, target_path, options, reproducers_dir, max_time):  # pylint: disable=unused-argument\n    \"\"\"Runs a fuzz session.\n\n    Args:\n      target_path: Path to the target.\n      options: The FuzzOptions object returned by prepare().\n      reproducers_dir: The directory to put reproducers in when crashes\n          are found.\n      max_time: Maximum allowed time for the fuzzing to run.\n\n   Returns:\n      A FuzzResult object.\n    \"\"\"\n    runner = _get_runner(target_path)\n    _set_sanitizer_options(target_path)\n    timeout = max_time + _CLEAN_EXIT_SECS\n    fuzz_result = runner.run_and_wait(\n        additional_args=options.arguments, timeout=timeout)\n    fuzz_result.output = Engine.trim_logs(fuzz_result.output)\n\n    reproducer_path = _get_reproducer_path(fuzz_result.output, reproducers_dir)\n    crashes = []\n    if reproducer_path:\n      crashes.append(\n          engine.Crash(\n              str(reproducer_path), fuzz_result.output, [],\n              int(fuzz_result.time_executed)))\n\n    # Stats report is not available in Centipede yet.\n    stats = None\n    return engine.FuzzResult(fuzz_result.output, fuzz_result.command, crashes,\n                             stats, fuzz_result.time_executed)\n\n  @staticmethod\n  def trim_logs(fuzz_log):\n    \"\"\" Strips the 'CRASH LOG:' prefix that breaks stacktrace parsing.\n\n    Args:\n      fuzz_result: The ProcessResult returned by running fuzzer binary.\n    \"\"\"\n    trimmed_log_lines = [\n        line[len(_CRASH_LOG_PREFIX):]\n        if line.startswith(_CRASH_LOG_PREFIX) else line\n        for line in fuzz_log.splitlines()\n    ]\n    return '\\n'.join(trimmed_log_lines)\n\n  def reproduce(self, target_path, input_path, arguments, max_time):  # pylint: disable=unused-argument\n    \"\"\"Reproduces a crash given an input.\n\n    Args:\n      target_path: Path to the target.\n      input_path: Path to the reproducer input.\n      arguments: Additional arguments needed for reproduction.\n      max_time: Maximum allowed time for the reproduction.\n\n    Returns:\n      A ReproduceResult.\n    \"\"\"\n    _set_sanitizer_options(target_path)\n    target_binaries = self._get_binary_paths(target_path)\n    sanitized_target = str(target_binaries.sanitized)\n\n    existing_runner_flags = os.environ.get('CENTIPEDE_RUNNER_FLAGS')\n    if not existing_runner_flags:\n      os.environ['CENTIPEDE_RUNNER_FLAGS'] = (\n          f':rss_limit_mb={_RSS_LIMIT}'\n          f':timeout_per_input={_TIMEOUT_PER_INPUT_REPR}:')\n\n    runner = new_process.UnicodeProcessRunner(sanitized_target, [input_path])\n    result = runner.run_and_wait(timeout=max_time)\n\n    if existing_runner_flags:\n      os.environ['CENTIPEDE_RUNNER_FLAGS'] = existing_runner_flags\n    else:\n      os.unsetenv('CENTIPEDE_RUNNER_FLAGS')\n    result.output = Engine.trim_logs(result.output)\n\n    return engine.ReproduceResult(result.command, result.return_code,\n                                  result.time_executed, result.output)\n\n  def _create_temp_dir(self, name):\n    \"\"\"Creates temporary directory for fuzzing.\"\"\"\n    new_directory = pathlib.Path(fuzzer_utils.get_temp_dir(), name)\n    engine_common.recreate_directory(new_directory)\n    return new_directory\n\n  def minimize_corpus(self, target_path, arguments, input_dirs, output_dir,\n                      reproducers_dir, max_time):\n    \"\"\"Runs corpus minimization.\n    Args:\n      target_path: Path to the target.\n      arguments: Additional arguments needed for corpus minimization.\n      input_dirs: Input corpora.\n      output_dir: Output directory to place minimized corpus.\n      reproducers_dir: The directory to put reproducers in when crashes are\n          found.\n      max_time: Maximum allowed time for the minimization.\n\n    Returns:\n      A FuzzResult object.\n    \"\"\"\n    raise NotImplementedError\n\n  def minimize_testcase(self, target_path, arguments, input_path, output_path,\n                        max_time):\n    \"\"\"Minimizes a testcase.\n    Args:\n      target_path: Path to the target.\n      arguments: Additional arguments needed for testcase minimization.\n      input_path: Path to the reproducer input.\n      output_path: Path to the minimized output.\n      max_time: Maximum allowed time for the minimization.\n    Returns:\n      A ReproduceResult.\n    Raises:\n      TimeoutError: If the testcase minimization exceeds max_time.\n    \"\"\"\n    raise NotImplementedError\n\n  def cleanse(self, target_path, arguments, input_path, output_path, max_time):\n    \"\"\"Cleanses a testcase.\n    Args:\n      target_path: Path to the target.\n      arguments: Additional arguments needed for testcase cleanse.\n      input_path: Path to the reproducer input.\n      output_path: Path to the cleansed output.\n      max_time: Maximum allowed time for the cleanse.\n    Returns:\n      A ReproduceResult.\n    Raises:\n      TimeoutError: If the cleanse exceeds max_time.\n    \"\"\"\n    raise NotImplementedError\n", "path": "src/clusterfuzz/_internal/bot/fuzzers/centipede/engine.py"}]}