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,176 @@ 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_neither_batch_nor_swarming(self):
+    """Tests that on an orchestration tworker (TWORKER=True), is_remote_utask
+    returns False if the job is not configured for Cloud Batch or Swarming.
+
+    Context:
+    - Bot environment: Orchestration tworker VM
+    - Job config: A job targeting neither Cloud Batch nor Swarming (e.g. jobs
+      configured for legacy pooled execution).
+    - Expected behavior: Even though tworkers escalate fuzz tasks to UTask,
+      is_remote_utask must return False because no untrusted remote backend
+      (Batch or Swarming) is targeted, ensuring signed GCS URLs are not
+      unnecessarily generated.
+    """
+    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_android_al_fuzz_tworker_swarming(self):
+    """Tests that on an orchestration tworker (TWORKER=True), is_remote_utask
+    returns True for fuzz when the job targets Swarming.
+
+    Context:
+    - Bot environment: Orchestration tworker VM running preprocess.
+    - Job config: Swarming jobs such as Android Aluminium (AL) emulators
+      configured via IS_SWARMING_JOB or SWARMING_DIMENSIONS (is_swarming_task=True).
+    - Expected behavior: The tworker identifies this as an untrusted remote
+      uworker execution, returning True so signed GCS URLs are generated during
+      preprocess for data bundles and corpus assets.
+    """
+    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=True)
+    environment.set_value('TWORKER', True)
+
+    with mock_remotely_executing, mock_is_remote_task, mock_is_swarming:
+      result = task_types.is_remote_utask('fuzz', 'android_asan_chrome')
+
+    self.assertTrue(result)
+
+  def test_linux_fuzz_trusted_pooled_bot(self):
+    """Tests that on a trusted pooled bot (TWORKER=False, UWORKER=False),
+    is_remote_utask always evaluates to False regardless of job config.
+
+    Context:
+    - Bot environment: Long-lived trusted pooled bot (e.g. Windows GCE VMs
+      like clusterfuzz_windows_pre, bare-metal Mac bots, or legacy Linux VMs).
+    - Job config: Even if the job has batch enabled (is_remote_task=True).
+    - Expected behavior: Because the bot running this task is a pooled bot
+      (not a tworker), fuzz tasks must execute end-to-end locally in-memory
+      via UTaskLocalExecutor with direct GCS access (gsutil/gcloud), rather
+      than treating the task as remote. This prevents the regression from
+      PR #4965 where pooled bots crashed trying to run remote UTasks.
+    """
+    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):