Fix untrusted remote batch fuzz tasks to pass signed URLs

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

@@ -491,7 +491,21 @@ def update_data_bundle(
   """Updates a data bundle to the latest version."""
   data_bundle = uworker_io.entity_from_protobuf(data_bundle_corpus.data_bundle,
                                                 data_types.DataBundle)
-  logs.info('Setting up data bundle %s.' % data_bundle)
+  num_urls = len(data_bundle_corpus.corpus_urls)
+
+  # If this is executing on an untrusted worker, we require signed URLs
+  # Missing the signed URLs results in attempting to run the fuzzer without its
+  # required data bundle
+  # Search index data bundles are the exception, which don't sync via this path.
+  if (environment.is_uworker() and
+      not _is_search_index_data_bundle(data_bundle.name) and num_urls == 0):
+    logs.error(
+        f'Uworker missing required signed URLs for data bundle'
+        f'{data_bundle.name}, and the fuzzer is configured with'
+        f'{fuzzer.data_bundle_name}. The Tworker failed to generate signed URLs'
+        ' during preprocess.')
+  else:
+    logs.info(f'Setting up data bundle {data_bundle.name}.')
   data_bundle_directory = _prepare_update_data_bundle(fuzzer, data_bundle)
 
   if not _should_update_data_bundle(data_bundle, data_bundle_directory):

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

@@ -105,20 +105,37 @@ def is_no_privilege_workload(command, job):
 
 
 def is_remote_utask(command, job):
-  if not COMMAND_TYPES[command].is_execution_remote(command):
-    return False
+  """Returns True if the task execution is meant to happen remotely.
 
+  Remote execution indicates that the main portion of this task will run on an
+  untrusted worker without privileges, requiring the trusted worker to
+  provision resources via signed URLs."""
   if environment.is_uworker():
     # Return True even if we can't query the db.
     return True
 
+  task_cls = COMMAND_TYPES[command]
+  if command == 'fuzz' and environment.is_tworker():
+    # Fuzz tasks run locally on pooled bots (as UTaskLocalExecutor), but on
+    # orchestration tworkers they are escalated to remote UTasks.
+    is_remote = UTask.is_execution_remote(command)
+  else:
+    is_remote = task_cls.is_execution_remote(command)
+
+  if not is_remote:
+    return False
+
   return batch_service.is_remote_task(command,
                                       job) or swarming.is_swarming_task(job)
 
 
 def task_main_runs_on_uworker():
-  """This returns True if the uworker_main portion of this task is
-  unprivileged."""
+  """Returns True if the uworker_main portion of this task is unprivileged.
+
+  Determining this accurately is critical for data provisioning. If this returns
+  False during prepocess on a tworker, signed GCS URLs for datasets (like data
+  bundles) are skipped. If the task is then executed on a Cloud Batch uworker,
+  bucket downloads will silently fail (resulting in an empty corpus)."""
   command = environment.get_value('TASK_NAME')
   job = environment.get_value('JOB_NAME')
   return is_remote_utask(command, job)

src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py

@@ -23,6 +23,7 @@
 from clusterfuzz._internal.bot.tasks import task_types
 from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.metrics import events
+from clusterfuzz._internal.system import environment
 from clusterfuzz._internal.tests.test_libs import helpers
 from clusterfuzz._internal.tests.test_libs import test_utils
 
@@ -62,6 +63,129 @@ def test_trusted(self):
       data_types.Job(name=job_name, platform='LINUX').put()
       self.assertFalse(task_types.is_remote_utask('impact', job_name))
 
+  def test_uworker(self):
+    """Tests that is_remote_utask returns True when running on a uworker."""
+    environment.set_value('UWORKER', True)
+
+    result = task_types.is_remote_utask('fuzz', 'any_job')
+
+    self.assertTrue(result)
+
+  def test_linux_fuzz_tworker(self):
+    """Tests that on an orchestration tworker, is_remote_utask evaluates 'fuzz'
+    as remote when configured for batch, ensuring signed URLs are generated
+    during preprocess."""
+    mock_remotely_executing = mock.patch(
+        'clusterfuzz._internal.base.tasks.task_utils.is_remotely_executing_utasks',
+        return_value=True)
+    mock_is_remote_task = mock.patch(
+        'clusterfuzz._internal.batch.service.is_remote_task', return_value=True)
+    environment.set_value('TWORKER', True)
+
+    with mock_remotely_executing, mock_is_remote_task:
+      result = task_types.is_remote_utask('fuzz', 'linux_asan_chrome')
+
+    self.assertTrue(result)
+
+  def test_linux_fuzz_tworker_swarming(self):
+    """Tests that on an orchestration tworker, is_remote_utask returns False for
+    fuzz if the task is not targeting remote batch or swarming."""
+    mock_remotely_executing = mock.patch(
+        'clusterfuzz._internal.base.tasks.task_utils.is_remotely_executing_utasks',
+        return_value=True)
+    mock_is_remote_task = mock.patch(
+        'clusterfuzz._internal.batch.service.is_remote_task',
+        return_value=False)
+    mock_is_swarming = mock.patch(
+        'clusterfuzz._internal.swarming.is_swarming_task', return_value=False)
+    environment.set_value('TWORKER', True)
+
+    with mock_remotely_executing, mock_is_remote_task, mock_is_swarming:
+      result = task_types.is_remote_utask('fuzz', 'linux_asan_chrome')
+
+    self.assertFalse(result)
+
+  def test_linux_fuzz_trusted_pooled_bot(self):
+    """Tests that trusted pooled bots (e.g. Windows, Mac, or legacy Linux bots
+    which run fuzz tasks end-to-end locally) evaluate is_remote_utask as False.
+    """
+    mock_remotely_executing = mock.patch(
+        'clusterfuzz._internal.base.tasks.task_utils.is_remotely_executing_utasks',
+        return_value=True)
+    mock_is_remote_task = mock.patch(
+        'clusterfuzz._internal.batch.service.is_remote_task', return_value=True)
+    environment.set_value('TWORKER', False)
+    environment.set_value('UWORKER', False)
+
+    with mock_remotely_executing, mock_is_remote_task:
+      result = task_types.is_remote_utask('fuzz', 'linux_asan_chrome')
+
+    self.assertFalse(result)
+
+  def test_trusted_tworker(self):
+    """Tests that on an orchestration tworker, is_remote_utask returns False non
+    fuzz tasks on batch."""
+    mock_remotely_executing = mock.patch(
+        'clusterfuzz._internal.base.tasks.task_utils.is_remotely_executing_utasks',
+        return_value=True)
+    mock_is_remote_task = mock.patch(
+        'clusterfuzz._internal.batch.service.is_remote_task', return_value=True)
+    environment.set_value('TWORKER', True)
+
+    with mock_remotely_executing, mock_is_remote_task:
+      result = task_types.is_remote_utask('impact', 'linux_asan_chrome')
+
+    self.assertFalse(result)
+
+
+@test_utils.with_cloud_emulators('datastore')
+class TaskMainRunsOnUworkerTest(unittest.TestCase):
+  """Tests for task_main_runs_on_uworker."""
+
+  def setUp(self):
+    helpers.patch_environ(self)
+    helpers.patch(self, [
+        'clusterfuzz._internal.base.tasks.task_utils.is_remotely_executing_utasks',
+        'clusterfuzz._internal.batch.service.is_remote_task',
+    ])
+    self.mock.is_remotely_executing_utasks.return_value = True
+    self.mock.is_remote_task.return_value = True
+
+  def test_fuzz_tworker(self):
+    """Tests that task_main_runs_on_uworker returns True during preprocess on
+    an orchestration tworker for a batch fuzz task, triggering signed URL
+    generation for data bundles."""
+    environment.set_value('TASK_NAME', 'fuzz')
+    environment.set_value('JOB_NAME', 'linux_asan_chrome')
+    environment.set_value('TWORKER', True)
+
+    result = task_types.task_main_runs_on_uworker()
+
+    self.assertTrue(result)
+
+  def test_fuzz_trusted_pooled_bot(self):
+    """Tests that task_main_runs_on_uworker returns False on trusted pooled bots
+    """
+    environment.set_value('TASK_NAME', 'fuzz')
+    environment.set_value('JOB_NAME', 'linux_asan_chrome')
+    environment.set_value('TWORKER', False)
+    environment.set_value('UWORKER', False)
+
+    result = task_types.task_main_runs_on_uworker()
+
+    self.assertFalse(result)
+
+  def test_trusted_task(self):
+    """Tests that task_main_runs_on_uworker returns False for trusted tasks
+    (e.g. impact) even on orchestration tworkers."""
+    environment.set_value('TASK_NAME', 'impact')
+    environment.set_value('JOB_NAME', 'linux_asan_chrome')
+    environment.set_value('TWORKER', True)
+
+    result = task_types.task_main_runs_on_uworker()
+
+    self.assertFalse(result)
+
 
 @test_utils.with_cloud_emulators('datastore')
 class TrustedTaskEventTest(unittest.TestCase):