diff --git a/src/clusterfuzz/_internal/base/tasks/__init__.py b/src/clusterfuzz/_internal/base/tasks/__init__.py
index acfb8f5de87..98709dcbb21 100644
--- a/src/clusterfuzz/_internal/base/tasks/__init__.py
+++ b/src/clusterfuzz/_internal/base/tasks/__init__.py
@@ -234,8 +234,23 @@ def get_regular_task(queue=None):
       return None
 
     task = get_task_from_message(messages[0], queue)
-    if task:
-      return task
+    if not task:
+      continue
+
+    if task.command == 'fuzz' and not environment.is_tworker():
+      fuzzer = data_types.Fuzzer.query(
+          data_types.Fuzzer.name == task.argument).get()
+      if not fuzzer:
+        logs.error(
+            f'Fuzzer {task.argument} not found. Discarding invalid task.')
+        task.dont_retry()
+        continue
+      if not fuzzer.trusted:
+        logs.info(
+            f'Skipping untrusted fuzzer {task.argument} on long-lived bot.')
+        continue
+
+    return task
 
 
 def get_machine_template_for_queue(queue_name):
diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py
index 3b6f9ac4447..ea0a5cda1b7 100644
--- a/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py
+++ b/src/clusterfuzz/_internal/bot/tasks/utasks/fuzz_task.py
@@ -1169,6 +1169,13 @@ def create_testcase(group: uworker_msg_pb2.FuzzTaskCrashGroup,
   comment = (f'Fuzzer {fully_qualified_fuzzer_name} generated testcase crashed '
              f'in {crash.crash_time} seconds '
              f'(r{fuzz_task_output.crash_revision})')
+
+  is_trusted = True
+  if uworker_input.setup_input.HasField('fuzzer'):
+    fuzzer = uworker_io.entity_from_protobuf(uworker_input.setup_input.fuzzer,
+                                             data_types.Fuzzer)
+    is_trusted = fuzzer.trusted
+
   testcase_id = data_handler.store_testcase(
       crash=crash,
       fuzzed_keys=crash.fuzzed_key or None,
@@ -1194,7 +1201,7 @@ def create_testcase(group: uworker_msg_pb2.FuzzTaskCrashGroup,
       minimized_arguments=crash.arguments,
       # TODO(https://github.com/google/clusterfuzz/issues/4175): Before enabling
       # oss-fuzz-on-demand change this.
-      trusted=True)
+      trusted=is_trusted)
   testcase = data_handler.get_testcase_by_id(testcase_id)
   events.emit(
       events.TestcaseCreationEvent(
@@ -2125,6 +2132,7 @@ def run(self):
       time.sleep(failure_wait_interval)
       return uworker_msg_pb2.Output(  # pylint: disable=no-member
           error_type=uworker_msg_pb2.ErrorType.FUZZ_NO_FUZZER)  # pylint: disable=no-member
+    uworker_io.check_running_fuzzer_safe(self.fuzzer)
 
     # Update the session's test_timeout to use the fuzzer's timeout (if any).
     # When the fuzzer has a specified timeout, `update_fuzzer_and_data_bundles`
diff --git a/src/clusterfuzz/_internal/bot/tasks/utasks/uworker_io.py b/src/clusterfuzz/_internal/bot/tasks/utasks/uworker_io.py
index 7380a512890..72f9ad9ad30 100644
--- a/src/clusterfuzz/_internal/bot/tasks/utasks/uworker_io.py
+++ b/src/clusterfuzz/_internal/bot/tasks/utasks/uworker_io.py
@@ -237,10 +237,25 @@ def check_handling_testcase_safe(testcase):
   safely."""
   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.')
+  if environment.is_uworker():
+    return
+
+  logs.log_fatal_and_exit(
+      f'Security Violation: Cannot handle untrusted testcase '
+      f'{testcase.key.id()} in long-lived bot.')
+
+
+def check_running_fuzzer_safe(fuzzer):
+  """Exits when the fuzzer is untrusted but the execution environment is
+  trusted."""
+  if fuzzer.trusted:
+    return True
+  if environment.is_uworker():
+    return True
+  logs.log_fatal_and_exit(
+      f'Security Violation: Cannot run untrusted fuzzer {fuzzer.name} '
+      f'in trusted environment.')
+  return False
 
 
 def timestamp_to_proto_timestamp(pydt) -> Timestamp:
diff --git a/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py b/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py
index 0d3edfec30b..09be079e2d5 100644
--- a/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py
+++ b/src/clusterfuzz/_internal/fuzzing/fuzzer_selection.py
@@ -170,6 +170,19 @@ def get_fuzz_task_payload(platform=None):
     jobs = get_job_list(jobs_selection)
     selected_mappings = [entity for entity in mappings if entity.job in jobs]
 
+  if not environment.is_uworker():
+    untrusted_fuzzers = {
+        fuzzer.name for fuzzer in ndb_utils.get_all_from_query(
+            data_types.Fuzzer.query(
+                ndb_utils.is_false(data_types.Fuzzer.trusted)))
+    }
+
+    if untrusted_fuzzers:
+      selected_mappings = [
+          mapping for mapping in selected_mappings
+          if mapping.fuzzer not in untrusted_fuzzers
+      ]
+
   if not selected_mappings:
     return None, None
 
diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py
index f0df7d3572a..b1904c2dae5 100644
--- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py
+++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/fuzz_task_test.py
@@ -1245,6 +1245,7 @@ def _setup_utask_env(self):
     os.environ['DATA_BUNDLES_DIR'] = '/tmp/data-bundles'
     os.environ['FUZZ_DATA'] = '/tmp/fuzz-data'
     os.environ['BOT_NAME'] = 'test-bot'
+    os.environ['UWORKER'] = 'True'
 
     self.fs.create_dir('/tmp/fuzz-data')
     self.fs.create_dir('/tmp/data-bundles')
@@ -2110,6 +2111,48 @@ def test_create_testcase_event_emit(self):
             creation_origin=events.TestcaseOrigin.FUZZ_TASK,
             uploader=None))
 
+  def test_create_testcase_trusted_fuzzer(self):
+    """Test creating a testcase with a trusted fuzzer."""
+    fuzzer = data_types.Fuzzer(name='engine', trusted=True)
+    fuzzer.put()
+
+    self.uworker_input.setup_input.CopyFrom(uworker_msg_pb2.SetupInput())
+    self.uworker_input.setup_input.fuzzer.CopyFrom(
+        uworker_io.entity_to_protobuf(fuzzer))
+
+    self.mock.store_testcase.side_effect = _store_generic_testcase
+
+    fuzz_task.create_testcase(
+        group=self.group,
+        uworker_input=self.uworker_input,
+        uworker_output=self.uworker_output,
+        fully_qualified_fuzzer_name='engine')
+
+    self.mock.store_testcase.assert_called_once()
+    kwargs = self.mock.store_testcase.call_args[1]
+    self.assertTrue(kwargs.get('trusted'))
+
+  def test_create_testcase_untrusted_fuzzer(self):
+    """Test creating a testcase with an untrusted fuzzer."""
+    fuzzer = data_types.Fuzzer(name='engine', trusted=False)
+    fuzzer.put()
+
+    self.uworker_input.setup_input.CopyFrom(uworker_msg_pb2.SetupInput())
+    self.uworker_input.setup_input.fuzzer.CopyFrom(
+        uworker_io.entity_to_protobuf(fuzzer))
+
+    self.mock.store_testcase.side_effect = _store_generic_testcase
+
+    fuzz_task.create_testcase(
+        group=self.group,
+        uworker_input=self.uworker_input,
+        uworker_output=self.uworker_output,
+        fully_qualified_fuzzer_name='engine')
+
+    self.mock.store_testcase.assert_called_once()
+    kwargs = self.mock.store_testcase.call_args[1]
+    self.assertFalse(kwargs.get('trusted'))
+
 
 def _store_generic_testcase(*args, **kwargs):  # pylint: disable=unused-argument
   """Store a generic testcase and return its id."""
diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/minimize_task_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/minimize_task_test.py
index 75f22804990..069ba877abb 100644
--- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/minimize_task_test.py
+++ b/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(
diff --git a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/uworker_io_test.py b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/uworker_io_test.py
index 527a821d272..969f7e2277e 100644
--- a/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/uworker_io_test.py
+++ b/src/clusterfuzz/_internal/tests/core/bot/tasks/utasks/uworker_io_test.py
@@ -387,3 +387,32 @@ def test_issue_metadata_field(self):
     wire_format = uworker_io.serialize_uworker_input(output)
     deserialized = uworker_io.deserialize_uworker_output(wire_format)
     self.assertEqual(json.loads(deserialized.issue_metadata), metadata)
+
+
+class TestCheckRunningFuzzerSafe(unittest.TestCase):
+  """Tests check_running_fuzzer_safe."""
+
+  def setUp(self):
+    helpers.patch(self, [
+        'clusterfuzz._internal.system.environment.is_uworker',
+    ])
+    self.fuzzer = mock.MagicMock(spec=data_types.Fuzzer)
+    self.fuzzer.name = 'test_fuzzer'
+
+  def test_trusted_fuzzer(self):
+    """Test that trusted fuzzer passes without checks."""
+    self.fuzzer.trusted = True
+    self.assertTrue(uworker_io.check_running_fuzzer_safe(self.fuzzer))
+
+  def test_untrusted_fuzzer_uworker(self):
+    """Test that untrusted fuzzer on uworker passes."""
+    self.fuzzer.trusted = False
+    self.mock.is_uworker.return_value = True
+    self.assertTrue(uworker_io.check_running_fuzzer_safe(self.fuzzer))
+
+  def test_untrusted_fuzzer_not_uworker_raises(self):
+    """Test that untrusted fuzzer not on uworker raises SystemExit."""
+    self.fuzzer.trusted = False
+    self.mock.is_uworker.return_value = False
+    with self.assertRaises(SystemExit):
+      uworker_io.check_running_fuzzer_safe(self.fuzzer)
diff --git a/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py b/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py
index 4859eb12db6..5421cd9418d 100644
--- a/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py
+++ b/src/clusterfuzz/_internal/tests/core/fuzzing/fuzzer_selection_test.py
@@ -325,6 +325,38 @@ def test_platform_restriction(self, local_development):
       argument, job = fuzzer_selection.get_fuzz_task_payload('ANDROID:PIXEL6')
       self.assertEqual(('pixel_fuzzer', 'job_pixel6'), (argument, job))
 
+  @parameterized.parameterized.expand([
+      ('False', 'trusted_fuzzer', 'job_2'),
+      ('True', 'untrusted_fuzzer', 'job_1'),
+  ])
+  def test_untrusted_fuzzer_exclusion(self, is_uworker, expected_fuzzer,
+                                      expected_job):
+    """Ensure that untrusted fuzzers are filtered out on long-lived bots."""
+    os.environ['UWORKER'] = is_uworker
+
+    data_types.Fuzzer(name='untrusted_fuzzer', trusted=False).put()
+    data_types.Fuzzer(name='trusted_fuzzer', trusted=True).put()
+
+    untrusted_mapping = data_types.FuzzerJob()
+    untrusted_mapping.fuzzer = 'untrusted_fuzzer'
+    untrusted_mapping.job = 'job_1'
+    untrusted_mapping.platform = 'linux'
+    untrusted_mapping.put()
+
+    trusted_mapping = data_types.FuzzerJob()
+    trusted_mapping.fuzzer = 'trusted_fuzzer'
+    trusted_mapping.job = 'job_2'
+    trusted_mapping.platform = 'linux'
+    trusted_mapping.put()
+
+    data_types.FuzzerJobs(
+        platform='linux', fuzzer_jobs=[untrusted_mapping,
+                                       trusted_mapping]).put()
+
+    argument, job = fuzzer_selection.get_fuzz_task_payload('linux')
+
+    self.assertEqual((expected_fuzzer, expected_job), (argument, job))
+
 
 @test_utils.with_cloud_emulators('datastore')
 class UpdatePlatformForJobTest(unittest.TestCase):
diff --git a/src/clusterfuzz/_internal/tests/test_libs/test_utils.py b/src/clusterfuzz/_internal/tests/test_libs/test_utils.py
index 5048122163f..6b0d8e88381 100644
--- a/src/clusterfuzz/_internal/tests/test_libs/test_utils.py
+++ b/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
diff --git a/src/clusterfuzz/_internal/tests/test_libs/utask_helpers.py b/src/clusterfuzz/_internal/tests/test_libs/utask_helpers.py
index 174ffc1d537..d6470620a53 100644
--- a/src/clusterfuzz/_internal/tests/test_libs/utask_helpers.py
+++ b/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'))