Protect testcase task-log downloads

src/appengine/handlers/testcase_detail/show.py

@@ -640,9 +640,18 @@ class TaskLogHandler(base_handler.Handler):
   @handler.get(handler.TEXT)
   def get(self):
     """Serve the task log."""
+    testcase_id = helpers.cast(flask.request.args.get('testcase_id'), int,
+                               "The param 'testcase_id' is not a number.")
+    access.check_access_and_get_testcase(testcase_id)
+
     task_id = flask.request.args.get('task_id')
+    if not task_id:
+      raise helpers.EarlyExitError('No task ID provided.', 400)
+
     task_name = flask.request.args.get('task_name')
-    testcase_id = flask.request.args.get('testcase_id')
+    if not task_name:
+      raise helpers.EarlyExitError('No task name provided.', 400)
+
     log_content = testcase_status_events.get_task_log(testcase_id, task_id,
                                                       task_name)
 

src/appengine/handlers/testcase_detail/testcase_status_events.py

@@ -31,6 +31,11 @@
 _BASE_LOGS_URL = 'https://console.cloud.google.com/logs/viewer'
 
 
+def _quote_logging_filter_value(value: str) -> str:
+  """Formats a string literal for a Cloud Logging query filter."""
+  return json.dumps(value)
+
+
 def _format_timestamp(timestamp: datetime.datetime) -> str:
   """Formats a timestamp."""
   return timestamp.strftime('%Y-%m-%d %H:%M:%S.%f UTC')
@@ -221,9 +226,10 @@ def _get_time_range_filter(self, days: int) -> str:
 
   def _get_task_log_query_filter(self, task_id: str, task_name: str) -> str:
     """Returns the filter string for querying task logs."""
-    query = (f'jsonPayload.extras.task_id="{task_id}" AND '
-             f'jsonPayload.extras.testcase_id="{self._testcase_id}" AND '
-             f'jsonPayload.extras.task_name="{task_name}"')
+    query = (f'jsonPayload.extras.task_id={_quote_logging_filter_value(task_id)} AND '
+             'jsonPayload.extras.testcase_id='
+             f'{_quote_logging_filter_value(str(self._testcase_id))} AND '
+             f'jsonPayload.extras.task_name={_quote_logging_filter_value(task_name)}')
     query += f' AND {self._get_time_range_filter(days=31)}'
     return query
 

src/clusterfuzz/_internal/tests/appengine/handlers/testcase_detail/show_test.py

@@ -18,6 +18,9 @@
 import os
 import unittest
 
+import flask
+import webtest
+
 from clusterfuzz._internal.datastore import data_types
 from clusterfuzz._internal.tests.test_libs import helpers as test_helpers
 from clusterfuzz._internal.tests.test_libs import test_utils
@@ -607,3 +610,39 @@ def test_unreproducible_get(self):
     self.assertDictContainsSubset({
         'lines': [show.Line(1, 'crash_stacktrace', False)]
     }, result['last_tested_crash_stacktrace'])
+
+@test_utils.with_cloud_emulators('datastore')
+class TaskLogHandlerTest(unittest.TestCase):
+  """Tests for TaskLogHandler."""
+
+  def setUp(self):
+    test_helpers.patch(self, [
+        'handlers.testcase_detail.show.access.check_access_and_get_testcase',
+        'handlers.testcase_detail.show.testcase_status_events.get_task_log',
+    ])
+    flaskapp = flask.Flask('testflask')
+    flaskapp.add_url_rule('/', view_func=show.TaskLogHandler.as_view('/'))
+    self.app = webtest.TestApp(flaskapp)
+    self.mock.get_task_log.return_value = 'task log content'
+
+  def test_get(self):
+    """Ensure the handler checks testcase access before returning logs."""
+    response = self.app.get('/?testcase_id=123&task_id=task-1&task_name=minimize')
+
+    self.assertEqual(200, response.status_int)
+    self.assertEqual('task log content', response.text)
+    self.assertEqual('text/plain; charset=utf-8', response.headers['Content-Type'])
+    self.assertEqual('attachment; filename="task_task-1_log.txt"',
+                     response.headers['Content-Disposition'])
+    self.mock.check_access_and_get_testcase.assert_called_once_with(123)
+    self.mock.get_task_log.assert_called_once_with(123, 'task-1', 'minimize')
+
+  def test_invalid_testcase_id(self):
+    """Ensure invalid testcase IDs are rejected before querying logs."""
+    response = self.app.get(
+        '/?testcase_id=abc&task_id=task-1&task_name=minimize',
+        expect_errors=True)
+
+    self.assertEqual(400, response.status_int)
+    self.mock.check_access_and_get_testcase.assert_not_called()
+    self.mock.get_task_log.assert_not_called()

src/clusterfuzz/_internal/tests/appengine/handlers/testcase_detail/testcase_status_events_test.py

@@ -15,6 +15,7 @@
 # pylint: disable=protected-access
 
 import datetime
+import json
 import unittest
 from unittest import mock
 
@@ -746,9 +747,23 @@ def test_get_task_log_query_filter(self):
     """Verify that the task log query filter is generated correctly."""
     result = self.event_history._get_task_log_query_filter(
         'task123', 'minimize')
-    expected = (f'jsonPayload.extras.task_id="task123" AND '
-                f'jsonPayload.extras.testcase_id="{self.testcase_id}" AND '
-                'jsonPayload.extras.task_name="minimize" AND '
+    quoted_testcase_id = json.dumps(str(self.testcase_id))
+    expected = (f'jsonPayload.extras.task_id={json.dumps("task123")} AND '
+                f'jsonPayload.extras.testcase_id={quoted_testcase_id} AND '
+                f'jsonPayload.extras.task_name={json.dumps("minimize")} AND '
+                'timestamp >= "2025-01-01T00:00:00Z"')
+    self.assertEqual(result, expected)
+
+  def test_get_task_log_query_filter_escapes_values(self):
+    """Verify that task values are escaped before building the log filter."""
+    task_id = 'task123" OR severity>="WARNING'
+    task_name = 'minimize\\latest'
+
+    result = self.event_history._get_task_log_query_filter(task_id, task_name)
+    quoted_testcase_id = json.dumps(str(self.testcase_id))
+    expected = (f'jsonPayload.extras.task_id={json.dumps(task_id)} AND '
+                f'jsonPayload.extras.testcase_id={quoted_testcase_id} AND '
+                f'jsonPayload.extras.task_name={json.dumps(task_name)} AND '
                 'timestamp >= "2025-01-01T00:00:00Z"')
     self.assertEqual(result, expected)