From d3a30f0160ce48567a93ff1c132ab285f5620e1b Mon Sep 17 00:00:00 2001 From: Dylan Jew Date: Thu, 3 Sep 2026 19:53:11 -0400 Subject: [PATCH] Fix untrusted remote batch fuzz tasks to pass signed URLs --- src/clusterfuzz/_internal/bot/tasks/setup.py | 16 ++- .../_internal/bot/tasks/task_types.py | 25 +++- .../tests/core/bot/tasks/task_types_test.py | 124 ++++++++++++++++++ 3 files changed, 160 insertions(+), 5 deletions(-) diff --git a/src/clusterfuzz/_internal/bot/tasks/setup.py b/src/clusterfuzz/_internal/bot/tasks/setup.py index 9e8aab54316..4b68ad09c37 100644 --- a/src/clusterfuzz/_internal/bot/tasks/setup.py +++ b/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): diff --git a/src/clusterfuzz/_internal/bot/tasks/task_types.py b/src/clusterfuzz/_internal/bot/tasks/task_types.py index 9152ec98880..996c2209d2a 100644 --- a/src/clusterfuzz/_internal/bot/tasks/task_types.py +++ b/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) diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py index 240ca27087f..33a95de19fc 100644 --- a/src/clusterfuzz/_internal/tests/core/bot/tasks/task_types_test.py +++ b/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):