Upload crashes to logs bucket in permissionless way (#4400)

Author: jonathanmetzman

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

@@ -1530,11 +1530,17 @@ def do_engine_fuzzing(self, engine_impl):
       crash_result_obj = crash_result.CrashResult(
           return_code, result.time_executed, result.logs)
       output = crash_result_obj.get_stacktrace()
-      self.fuzz_task_output.engine_outputs.append(
-          _to_engine_output(output, return_code, log_time))
+      # TODO(metzman): Consider uploading this with a signed URL.
+      if result.crashes:
+        # We only upload the first, because they will clobber each other if we
+        # upload more.
+        result_crash = result.crashes[0].input_path
+      else:
+        result_crash = None
 
-      for crash in result.crashes:
-        testcase_manager.upload_testcase(crash.input_path, log_time)
+      engine_output = _to_engine_output(output, result_crash, return_code,
+                                        log_time)
+      self.fuzz_task_output.engine_outputs.append(engine_output)
 
       add_additional_testcase_run_data(testcase_run,
                                        self.fuzz_target.fully_qualified_name(),
@@ -2051,7 +2057,7 @@ def save_fuzz_targets(output):
                                    output.uworker_input.job_type)
 
 
-def _to_engine_output(output: str, return_code: int,
+def _to_engine_output(output: str, crash_path: str, return_code: int,
                       log_time: datetime.datetime):
   """Returns an EngineOutput proto."""
   truncated_output = truncate_fuzzer_output(output, ENGINE_OUTPUT_LIMIT)
@@ -2063,13 +2069,22 @@ def _to_engine_output(output: str, return_code: int,
       output=bytes(truncated_output, 'utf-8'),
       return_code=return_code,
       timestamp=proto_timestamp)
+
+  if crash_path is None:
+    return engine_output
+  if os.path.getsize(crash_path) > 10 * 1024**2:
+    return engine_output
+  with open(crash_path, 'rb') as fp:
+    engine_output.testcase = fp.read()
+
   return engine_output
 
 
-def _upload_engine_output_log(engine_output):
+def _upload_engine_output(engine_output):
   timestamp = uworker_io.proto_timestamp_to_timestamp(engine_output.timestamp)
   testcase_manager.upload_log(engine_output.output.decode(),
                               engine_output.return_code, timestamp)
+  testcase_manager.upload_testcase(None, engine_output.testcase, timestamp)
 
 
 def utask_postprocess(output):
@@ -2087,4 +2102,4 @@ def utask_postprocess(output):
   # TODO(b/374776013): Refactor this code so the uploads happen during
   # utask_main.
   for engine_output in output.fuzz_task_output.engine_outputs:
-    _upload_engine_output_log(engine_output)
+    _upload_engine_output(engine_output)

src/clusterfuzz/_internal/bot/testcase_manager.py

@@ -424,21 +424,22 @@ def _get_testcase_time(testcase_path):
   return None
 
 
-def upload_testcase(testcase_path, log_time):
+def upload_testcase(testcase_path, testcase_data, log_time):
   """Uploads testcase so that a log file can be matched with it folder."""
   fuzz_logs_bucket = environment.get_value('FUZZ_LOGS_BUCKET')
   if not fuzz_logs_bucket:
     return
 
-  if not os.path.exists(testcase_path):
-    return
-
-  with open(testcase_path, 'rb') as file_handle:
-    testcase_contents = file_handle.read()
+  assert not (testcase_path and testcase_data)
+  if testcase_path:
+    if not os.path.exists(testcase_path):
+      return
+    with open(testcase_path, 'rb') as file_handle:
+      testcase_data = file_handle.read()
 
   fuzzer_logs.upload_to_logs(
       fuzz_logs_bucket,
-      testcase_contents,
+      testcase_data,
       time=log_time,
       file_extension='.testcase')
 
@@ -534,7 +535,7 @@ def _do_run_testcase_and_return_result_in_queue(crash_queue,
       # 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, log_time)
+        upload_testcase(file_path, log_time, None)
 
     if upload_output:
       # Include full output for uploaded logs (crash output, merge output, etc).

src/clusterfuzz/_internal/protos/uworker_msg.proto

@@ -240,6 +240,7 @@ message EngineOutput {
   optional bytes output = 1;
   optional int64 return_code = 2;
   google.protobuf.Timestamp timestamp = 3;
+  optional bytes testcase = 4;
 }
 
 message FuzzTaskOutput {

src/clusterfuzz/_internal/protos/uworker_msg_pb2.py

@@ -1199,23 +1199,28 @@ class EngineOutput(google.protobuf.message.Message):
     OUTPUT_FIELD_NUMBER: builtins.int
     RETURN_CODE_FIELD_NUMBER: builtins.int
     TIMESTAMP_FIELD_NUMBER: builtins.int
+    TESTCASE_FIELD_NUMBER: builtins.int
     output: builtins.bytes
     return_code: builtins.int
     @property
     def timestamp(self) -> google.protobuf.timestamp_pb2.Timestamp: ...
+    testcase: builtins.bytes
     def __init__(
         self,
         *,
         output: builtins.bytes | None = ...,
         return_code: builtins.int | None = ...,
         timestamp: google.protobuf.timestamp_pb2.Timestamp | None = ...,
+        testcase: builtins.bytes | None = ...,
     ) -> None: ...
-    def HasField(self, field_name: typing_extensions.Literal["_output", b"_output", "_return_code", b"_return_code", "output", b"output", "return_code", b"return_code", "timestamp", b"timestamp"]) -> builtins.bool: ...
-    def ClearField(self, field_name: typing_extensions.Literal["_output", b"_output", "_return_code", b"_return_code", "output", b"output", "return_code", b"return_code", "timestamp", b"timestamp"]) -> None: ...
+    def HasField(self, field_name: typing_extensions.Literal["_output", b"_output", "_return_code", b"_return_code", "_testcase", b"_testcase", "output", b"output", "return_code", b"return_code", "testcase", b"testcase", "timestamp", b"timestamp"]) -> builtins.bool: ...
+    def ClearField(self, field_name: typing_extensions.Literal["_output", b"_output", "_return_code", b"_return_code", "_testcase", b"_testcase", "output", b"output", "return_code", b"return_code", "testcase", b"testcase", "timestamp", b"timestamp"]) -> None: ...
     @typing.overload
     def WhichOneof(self, oneof_group: typing_extensions.Literal["_output", b"_output"]) -> typing_extensions.Literal["output"] | None: ...
     @typing.overload
     def WhichOneof(self, oneof_group: typing_extensions.Literal["_return_code", b"_return_code"]) -> typing_extensions.Literal["return_code"] | None: ...
+    @typing.overload
+    def WhichOneof(self, oneof_group: typing_extensions.Literal["_testcase", b"_testcase"]) -> typing_extensions.Literal["testcase"] | None: ...
 
 global___EngineOutput = EngineOutput
 

src/clusterfuzz/_internal/protos/uworker_msg_pb2.pyi

@@ -1356,12 +1356,6 @@ def test_basic(self):
             '[email protected]',
     }, fuzzer_metadata)
 
-    log_time = datetime.datetime(1970, 1, 1, 0, 0)
-    self.mock.upload_testcase.assert_has_calls([
-        mock.call('/input', log_time),
-        mock.call('/input', log_time),
-    ])
-
     self.assertEqual(2, len(crashes))
     for i in range(2):
       self.assertEqual('/input', crashes[i].file_path)
@@ -1383,6 +1377,8 @@ def test_basic(self):
           'strategy_strategy_2': 50,
           'timestamp': 0.0,
       }, testcase_run)
+      # TODO(metzman): We need a test for fuzzing end to end with
+      # preprocess/main/postprocess.
 
 
 class UntrustedRunEngineFuzzerTest(