Fix bypass Datastore access in uworker for job definition (#5293)

### Problem

The function `get_value_from_job_definition` in data_handler.py can
cause crashes when executed within a UWORKER (untrusted worker)
environment.

The function originally executed a query against Google Cloud Datastore:

`job = data_types.Job.query(data_types.Job.name == job_type).get()`

Because the UWORKER Docker container uses a restricted IAM Service
Account ( untrusted-worker@... ), it has zero permissions to read from
Datastore. This caused a fatal 403 PermissionDenied exception, crashing
processes like the log upload.

### Fix

When a task runs as a UWORKER, the Trusted Worker (TWORKER) that
scheduled the task has already queried Datastore, extracted all the
Job's environment variables, and injected them directly into the
UWORKER's local environment. Therefore, querying Datastore from the
UWORKER is both forbidden and redundant.

---------

Co-authored-by: Diego Jardon <jardondiego@google.com>

Author: jardondiego

src/clusterfuzz/_internal/datastore/data_handler.py

@@ -1262,6 +1262,11 @@ def get_value_from_job_definition(job_type, variable_pattern, default=None):
   if not job_type:
     return default
 
+  # Short-circuit: UWORKERs do not have Datastore access. The TWORKER has
+  # already injected all job variables into the local environment.
+  if environment.is_uworker():
+    return environment.get_value(variable_pattern, default)
+
   job = data_types.Job.query(data_types.Job.name == job_type).get()
   if not job:
     return default

src/clusterfuzz/_internal/tests/core/datastore/data_handler_test.py

@@ -1223,3 +1223,26 @@ def test_no_entities_found(self):
         self.GetEntitiesTestModel, equality_filters={'value': 5})
     self.assertIsInstance(result, Generator)
     self.assertCountEqual(result, [])
+
+
+class GetValueFromJobDefinitionTest(unittest.TestCase):
+  """Tests for get_value_from_job_definition."""
+
+  def setUp(self):
+    helpers.patch(self, [
+        'clusterfuzz._internal.system.environment.is_uworker',
+        'clusterfuzz._internal.system.environment.get_value',
+        'clusterfuzz._internal.datastore.data_types.Job.query',
+    ])
+
+  def test_uworker_bypass(self):
+    """Test that uworker environment skips Datastore and uses environment."""
+    self.mock.is_uworker.return_value = True
+    self.mock.get_value.return_value = 'env_value'
+
+    result = data_handler.get_value_from_job_definition(
+        'job_name', 'VAR_PATTERN', default='default_val')
+
+    self.assertEqual(result, 'env_value')
+    self.mock.get_value.assert_called_once_with('VAR_PATTERN', 'default_val')
+    self.mock.query.assert_not_called()