[VRP] Harden testcase safety check in long-lived bots

src/clusterfuzz/_internal/bot/tasks/task_creation.py

@@ -149,8 +149,6 @@ def create_regression_task_if_needed(testcase):
 
 def create_variant_tasks_if_needed(testcase):
   """Creates a variant task if needed."""
-  # TODO(https://b.corp.google.com/issues/328691756): Allow untrusted
-  # testcases to only run untrusted variants.
   if testcase.duplicate_of:
     # If another testcase exists with same params, no need to spend cycles on
     # calculating variants again.

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

@@ -887,8 +887,8 @@ def _process_corpus_crashes(output: uworker_msg_pb2.Output):  # pylint: disable=
       absolute_testcase_path = os.path.join(
           environment.get_value('FUZZ_INPUTS'), 'testcase')
 
-      # TODO(https://b.corp.google.com/issues/328691756): Set trusted based on
-      # the job when we start doing untrusted fuzzing.
+      # TODO(b/556173460): Set trusted based on
+      # the job when we start doing untrusted engine fuzzing
       testcase_id = data_handler.store_testcase(
           crash=crash,
           fuzzed_keys=key,

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

@@ -234,13 +234,16 @@ def entity_from_protobuf(entity_proto: any_pb2.Any, model_type: Type[T]) -> T:
 def check_handling_testcase_safe(testcase):
   """Exits when the current task execution model is trusted but the testcase is
   untrusted. This will allow uploading testcases to trusted jobs (e.g. Mac) more
-  safely."""
+  safely. Returns True if safe to handle, False otherwise."""
   if testcase.trusted:
-    return
-  if not environment.get_value('UNTRUSTED_UTASK'):
-    # TODO(https://b.corp.google.com/issues/328691756): Change this to
-    # log_fatal_and_exit once we are handling untrusted tasks properly.
-    logs.warning(f'Cannot handle {testcase.key.id()} in trusted task.')
+    return True
+  if environment.is_uworker():
+    return True
+
+  logs.log_fatal_and_exit(
+      f'Security Violation: Cannot handle untrusted testcase '
+      f'{testcase.key.id()} in long-lived bot.')
+  return False
 
 
 def check_running_fuzzer_safe(fuzzer):

src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/minimize_task_test.py

@@ -112,6 +112,7 @@ def setUp(self):
     """Set up."""
     super().setUp()
     environment.set_value('JOB_NAME', 'libfuzzer_asan_job')
+    environment.set_value('UWORKER', True)
 
     patcher = mock.patch(
         'clusterfuzz._internal.bot.fuzzers.libFuzzer.fuzzer.LibFuzzer.fuzzer_directory',
@@ -384,7 +385,7 @@ def test_check_app_path_exit(self, setup_testcase, preprocess_setup_testcase,
     setup_testcase.return_value = ([], '/path', None)
     del setup_build
     del check_app_path
-    testcase = data_types.Testcase()
+    testcase = data_types.Testcase(trusted=True)
     testcase.put()
     environment.set_value('FAIL_WAIT', 10)
     uworker_input = uworker_msg_pb2.Input(

src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/uworker_io_test.py

@@ -416,3 +416,32 @@ def test_untrusted_fuzzer_not_uworker_raises(self):
     self.mock.is_uworker.return_value = False
     with self.assertRaises(SystemExit):
       uworker_io.check_running_fuzzer_safe(self.fuzzer)
+
+
+class TestCheckRunningTestcaseSafe(unittest.TestCase):
+  """Tests check_handling_testcase_safe."""
+
+  def setUp(self):
+    helpers.patch(self, [
+        'clusterfuzz._internal.system.environment.is_uworker',
+    ])
+    self.testcase = mock.MagicMock(spec=data_types.Testcase)
+    self.testcase.name = 'test_testcase'
+
+  def test_trusted_testcase(self):
+    """Test that trusted testcase passes without checks."""
+    self.testcase.trusted = True
+    self.assertTrue(uworker_io.check_handling_testcase_safe(self.testcase))
+
+  def test_untrusted_testcase_uworker(self):
+    """Test that untrusted testcase on uworker passes."""
+    self.testcase.trusted = False
+    self.mock.is_uworker.return_value = True
+    self.assertTrue(uworker_io.check_handling_testcase_safe(self.testcase))
+
+  def test_untrusted_testcase_not_uworker_raises(self):
+    """Test that untrusted testcase not on uworker raises SystemExit."""
+    self.testcase.trusted = False
+    self.mock.is_uworker.return_value = False
+    with self.assertRaises(SystemExit):
+      uworker_io.check_handling_testcase_safe(self.testcase)

src/clusterfuzz/_internal/tests/test_libs/test_utils.py

@@ -63,6 +63,7 @@ def create_generic_testcase(created_days_ago=28):
   testcase.timestamp = CURRENT_TIME - datetime.timedelta(days=created_days_ago)
   testcase.project_name = 'project'
   testcase.platform = 'linux'
+  testcase.trusted = True
   testcase.put()
 
   return testcase

src/clusterfuzz/_internal/tests/test_libs/utask_helpers.py

@@ -64,7 +64,7 @@ def setUp(self):
     self.uworker_env = commands.update_environment_for_job(environment_string)
     job.put()
     self.fuzz_target = 'test_fuzzer'
-    self.testcase = data_types.Testcase(job_type=self.job_type)
+    self.testcase = data_types.Testcase(job_type=self.job_type, trusted=True)
     self.testcase.fuzzed_keys = blobs.write_blob(
         os.path.join(TEST_LIBS_DATA_DIR,
                      'crash-adc83b19e793491b1c6ea0fd8b46cd9f32e592fc'))