[NDB PERMISSION_DENIED] Defer Blackbox Fuzzer Log Uploads

src/clusterfuzz/_internal/base/utils.py

@@ -23,6 +23,7 @@
 import random
 import re
 import sys
+import tempfile
 import time
 import urllib.parse
 import urllib.request
@@ -1100,3 +1101,10 @@ def batched(iterator, batch_size):
 
   if batch:
     yield batch
+
+
+def create_temp_file(directory: str, prefix: str, suffix: str) -> str:
+  """Creates a temporary file and returns its path."""
+  fd, file_path = tempfile.mkstemp(dir=directory, prefix=prefix, suffix=suffix)
+  os.close(fd)
+  return file_path

src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py

@@ -18,6 +18,7 @@
 import itertools
 import json
 import os
+import queue
 import random
 import re
 import time
@@ -1738,6 +1739,19 @@ def _emit_testcase_generation_time_metric(self, start_time, testcase_count,
               'platform': environment.platform(),
           })
 
+  def _drain_temp_queue(self, temp_queue: queue.Queue,
+                        crashes: list[testcase_manager.Crash]) -> None:
+    """Pops items from temp_queue and processes crashes and run outputs."""
+    while not temp_queue.empty():
+      result: testcase_manager.TestcaseRunResult = temp_queue.get()
+
+      if result.crash:
+        crashes.append(result.crash)
+      if result.fuzzer_run_output_data:
+        fuzzer_run_output = _to_fuzzer_run_output(result.fuzzer_run_output_data)
+        if fuzzer_run_output:
+          self.fuzz_task_output.fuzzer_run_outputs.append(fuzzer_run_output)
+
   def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
     """Run blackbox fuzzing. Currently also used for engine fuzzing."""
     # Set the thread timeout values.
@@ -1847,7 +1861,7 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
         thread = process_handler.get_process()(
             target=testcase_manager.run_testcase_and_return_result_in_queue,
             args=(temp_queue, thread_index, testcase_file_path, gestures,
-                  env_copy, not environment.is_uworker()))
+                  env_copy))
 
         try:
           thread.start()
@@ -1880,9 +1894,7 @@ def do_blackbox_fuzzing(self, fuzzer, fuzzer_directory, job_type):
         process_handler.terminate_stale_application_instances()
         needs_stale_process_cleanup = False
 
-      while not temp_queue.empty():
-        crashes.append(temp_queue.get())
-
+      self._drain_temp_queue(temp_queue, crashes)
       process_handler.close_queue(temp_queue)
       logs.info(f'Upto {test_number}')
       if thread_error_occurred:

src/clusterfuzz/_internal/bot/testcase_manager.py

@@ -18,6 +18,7 @@
 import dataclasses
 import datetime
 import os
+import queue
 import re
 import zlib
 
@@ -481,7 +482,7 @@ def _get_testcase_time(testcase_path):
   if stats:
     return datetime.datetime.utcfromtimestamp(float(stats.timestamp))
 
-  return None
+  return datetime.datetime.utcnow()
 
 
 def upload_testcase(testcase_path, testcase_data, log_time, fuzzer_name=None):
@@ -522,12 +523,9 @@ def _get_crash_output(output):
   return output[:crash_stacktrace_end_marker_index]
 
 
-def run_testcase_and_return_result_in_queue(crash_queue,
-                                            thread_index,
-                                            file_path,
-                                            gestures,
-                                            env_copy,
-                                            upload_output=False):
+def run_testcase_and_return_result_in_queue(
+    crash_queue: queue.Queue, thread_index: int, file_path: str,
+    gestures: list[str], env_copy: dict) -> None:
   """Run a single testcase and return crash results in the crash queue."""
   # Since this is running in its own process, initialize the log handler again.
   # This is needed for Windows where instances are not shared across child
@@ -539,21 +537,13 @@ def run_testcase_and_return_result_in_queue(crash_queue,
 
   # Also reinitialize NDB context for the same reason as above.
   with ndb_init.context():
-    _do_run_testcase_and_return_result_in_queue(
-        crash_queue,
-        thread_index,
-        file_path,
-        gestures,
-        env_copy,
-        upload_output=upload_output)
-
-
-def _do_run_testcase_and_return_result_in_queue(crash_queue,
-                                                thread_index,
-                                                file_path,
-                                                gestures,
-                                                env_copy,
-                                                upload_output=False):
+    _do_run_testcase_and_return_result_in_queue(crash_queue, thread_index,
+                                                file_path, gestures, env_copy)
+
+
+def _do_run_testcase_and_return_result_in_queue(
+    crash_queue: queue.Queue, thread_index: int, file_path: str,
+    gestures: list[str], env_copy: dict) -> None:
   """Run a single testcase and return crash results in the crash queue."""
   try:
     # Run testcase and check whether a crash occurred or not.
@@ -571,9 +561,9 @@ def _do_run_testcase_and_return_result_in_queue(crash_queue,
 
     # To provide consistency between stats and logs, we use timestamp taken
     # from stats when uploading logs and testcase.
-    if upload_output:
-      log_time = _get_testcase_time(file_path)
+    log_time = _get_testcase_time(file_path)
 
+    crash = None
     if crash_result.is_crash():
       # Initialize resource list with the testcase path.
       resource_list = [file_path]
@@ -586,28 +576,36 @@ def _do_run_testcase_and_return_result_in_queue(crash_queue,
                                      utils.string_hash(file_path))
       utils.write_data_to_file(crash_output, stack_file_path)
 
-      # Put crash/no-crash results in the crash queue.
-      crash_queue.put(
-          Crash(
-              file_path=file_path,
-              crash_time=crash_time,
-              return_code=return_code,
-              resource_list=resource_list,
-              gestures=gestures,
-              stack_file_path=stack_file_path))
-
-      # Don't upload uninteresting testcases (no crash) or if there is no log to
-      # correlate it with (not upload_output).
-      if upload_output:
-        upload_testcase(file_path, None, log_time)
-
-    if upload_output:
-      # Include full output for uploaded logs (crash output, merge output, etc).
-      crash_result_full = CrashResult(return_code, crash_time, output)
-      upload_log(crash_result_full.get_stacktrace(), return_code, log_time)
-  except Exception:
-    logs.error('Exception occurred while running '
-               'run_testcase_and_return_result_in_queue.')
+      crash = Crash(
+          file_path=file_path,
+          crash_time=crash_time,
+          return_code=return_code,
+          resource_list=resource_list,
+          gestures=gestures,
+          stack_file_path=stack_file_path)
+
+    # Save raw, unsymbolized execution output for deferred postprocessing.
+    crash_result_full = CrashResult(return_code, crash_time, output)
+    unsymbolized_output = crash_result_full.get_stacktrace(symbolized=False)
+    crash_path = file_path if crash else None
+    log_file_path = utils.create_temp_file(
+        directory=environment.get_value('BOT_TMPDIR'),
+        prefix='fuzzer_output_',
+        suffix='.log')
+    utils.write_data_to_file(unsymbolized_output, log_file_path)
+    fuzzer_run_output_data = FuzzerRunOutputData.from_file_path(
+        file_path=log_file_path,
+        crash_path=crash_path,
+        return_code=return_code,
+        log_time=log_time)
+    crash_queue.put(
+        TestcaseRunResult(
+            crash=crash, fuzzer_run_output_data=fuzzer_run_output_data))
+  except Exception as e:
+    logs.error(
+        'Exception occurred while running '
+        'run_testcase_and_return_result_in_queue.',
+        exception=e)
 
 
 def engine_reproduce(engine_impl: engine.Engine, target_name, testcase_path,

src/clusterfuzz/_internal/tests/core/base/utils_test.py

@@ -720,3 +720,16 @@ def test_additional_prefix(self):
     """Test parsing manifest data with an added prefix."""
     file_data = 'prefix123-20250402153042-utc-40773ac0-username_test123-cad6977-prod'
     self.assertIsNone(utils.parse_manifest_data(file_data))
+
+
+class CreateTempFileTest(unittest.TestCase):
+  """Tests for create_temp_file."""
+
+  def test_create_temp_file(self):
+    """Test create_temp_file creates a file and returns its path."""
+    with tempfile.TemporaryDirectory() as tmpdir:
+      file_path = utils.create_temp_file(tmpdir, 'prefix_', '_suffix')
+      self.assertTrue(os.path.exists(file_path))
+      self.assertTrue(file_path.startswith(tmpdir))
+      self.assertIn('prefix_', file_path)
+      self.assertTrue(file_path.endswith('_suffix'))