[DO NOT MERGE] Make chromium testcases public when bug is made public

src/appengine/handlers/download.py

@@ -91,6 +91,31 @@ def check_public_testcase(self, blob_info, testcase):
 
     return True
 
+  def check_derestricted_testcase(self, blob_info, testcase):
+    """Check if a testcase's associated bug has been derestricted (made public).
+
+    For Chromium deployments, checks if the corresponding bug tracker issue has
+    no view restrictions, indicating it has been derestricted. Only allows
+    access to the minimized testcase. Gated on is_chromium() to avoid exposing
+    testcases on internal deployments where LIMIT_NONE doesn't mean truly
+    public.
+    """
+    if not utils.is_chromium():
+      return False
+
+    if blob_info.key() != testcase.minimized_keys:
+      return False
+
+    if not testcase.bug_information:
+      return False
+
+    issue_tracker = issue_tracker_utils.get_issue_tracker_for_testcase(testcase)
+    issue = issue_tracker.get_issue(testcase.bug_information)
+    if not issue:
+      return False
+
+    return issue.is_unrestricted
+
   def get(self, resource=None):
     """Handle a get request with resource."""
     testcase = None
@@ -132,6 +157,14 @@ def get(self, resource=None):
           is_minimized=True,
           fuzzer_binary_name=fuzzer_binary_name)
 
+    if testcase and self.check_derestricted_testcase(blob_info, testcase):
+      # Testcase for a derestricted (public) Chromium bug.
+      return self._send_blob(
+          blob_info,
+          testcase.key.id(),
+          is_minimized=True,
+          fuzzer_binary_name=fuzzer_binary_name)
+
     is_minimized = testcase and blob_info.key() == testcase.minimized_keys
 
     # Testcase blobs require testcase-level access (which enforces the

src/clusterfuzz/_internal/issue_management/google_issue_tracker/issue_tracker.py

@@ -240,6 +240,17 @@ def __init__(self, data, is_new, tracker):
     self._changed = set()
     self._issue_access_limit = IssueAccessLevel.LIMIT_NONE
 
+  @property
+  def is_unrestricted(self):
+    """Whether the issue has no view restrictions (i.e. is public).
+
+    Checks the accessLimit field from the Buganizer API response. An issue
+    with accessLevel == LIMIT_NONE has no view restrictions.
+    """
+    access_limit = self._data['issueState'].get('accessLimit',
+                                                {}).get('accessLevel')
+    return access_limit == IssueAccessLevel.LIMIT_NONE
+
   def _get_component_tags(self):
     """Returns the value of the Component Tags custom field."""
     custom_fields = self._data['issueState'].get('customFields', [])

src/clusterfuzz/_internal/issue_management/issue_tracker.py

@@ -202,6 +202,11 @@ def save(self, new_comment=None, notify=True):
     """Save the issue."""
     raise NotImplementedError
 
+  @property
+  def is_unrestricted(self):
+    """Whether the issue has no view restrictions (i.e. is public)."""
+    return False
+
   # pylint: disable=unused-argument
   def apply_extension_fields(self, extension_fields):
     """Applies _ext_ prefixed extension fields to the issue."""

src/clusterfuzz/_internal/tests/appengine/handlers/download_test.py

@@ -44,6 +44,7 @@ def setUp(self):
     self.testcase.put()
 
     test_helpers.patch(self, [
+        'clusterfuzz._internal.base.utils.is_chromium',
         'clusterfuzz._internal.base.utils.is_oss_fuzz',
         'clusterfuzz._internal.google_cloud_utils.blobs.get_blob_info',
         'libs.access.can_user_access_testcase',
@@ -53,6 +54,7 @@ def setUp(self):
         'clusterfuzz._internal.issue_management.issue_tracker_utils.'
         'get_issue_tracker_for_testcase',
     ])
+    self.mock.is_chromium.return_value = False
     self.mock.is_oss_fuzz.return_value = False
     self.mock.can_user_access_testcase.return_value = False
     self.mock.has_access.return_value = False
@@ -218,6 +220,66 @@ def test_public_download_chromium(self):
         expect_status=302,
         expect_blob=False)
 
+  def test_public_download_derestricted_chromium(self):
+    """Test public downloading of derestricted Chromium testcases."""
+    self.mock.get_user_email.return_value = ''
+    self.mock.is_chromium.return_value = True
+    self.mock.is_oss_fuzz.return_value = False
+    mock_issue = self.mock.get_issue_tracker_for_testcase(None).get_issue()
+    mock_issue.labels = issue_tracker.LabelStore([])
+    mock_issue.is_unrestricted = True
+
+    # Minimized testcase should be downloadable.
+    self._test_download(
+        self.minimized_key,
+        testcase_id=self.testcase.key.id(),
+        expect_filename='clusterfuzz-testcase-minimized-1.ext')
+    # Only minimized testcase, not fuzzed.
+    self._test_download(
+        self.fuzzed_key,
+        testcase_id=self.testcase.key.id(),
+        expect_status=302,
+        expect_blob=False)
+
+  def test_public_download_restricted_chromium(self):
+    """Test public downloading of still-restricted Chromium testcases fails."""
+    self.mock.get_user_email.return_value = ''
+    self.mock.is_chromium.return_value = True
+    self.mock.is_oss_fuzz.return_value = False
+    mock_issue = self.mock.get_issue_tracker_for_testcase(None).get_issue()
+    mock_issue.labels = issue_tracker.LabelStore([])
+    mock_issue.is_unrestricted = False
+
+    self._test_download(
+        self.minimized_key,
+        testcase_id=self.testcase.key.id(),
+        expect_status=302,
+        expect_blob=False)
+    self._test_download(
+        self.fuzzed_key,
+        testcase_id=self.testcase.key.id(),
+        expect_status=302,
+        expect_blob=False)
+
+  def test_public_download_non_chromium_unrestricted_bug(self):
+    """Test that unrestricted bugs on non-Chromium deployments don't expose
+    testcases. This guards against internal deployments where LIMIT_NONE means
+    'accessible to Googlers' rather than 'truly public'."""
+    self.mock.get_user_email.return_value = ''
+    self.mock.is_chromium.return_value = False
+    self.mock.is_oss_fuzz.return_value = False
+    mock_issue = self.mock.get_issue_tracker_for_testcase(None).get_issue()
+    mock_issue.labels = issue_tracker.LabelStore([])
+    mock_issue.is_unrestricted = True
+
+    # Even though the bug is unrestricted, non-Chromium deployments should
+    # not serve the testcase publicly.
+    self._test_download(
+        self.minimized_key,
+        testcase_id=self.testcase.key.id(),
+        expect_status=302,
+        expect_blob=False)
+
   def test_public_download_oss_fuzz(self):
     """Test public downloading OSS-Fuzz testcases."""
     self.mock.get_user_email.return_value = ''

src/clusterfuzz/_internal/tests/appengine/libs/issue_management/google_issue_tracker/google_issue_tracker_test.py

@@ -13,6 +13,7 @@
 # limitations under the License.
 """Tests for issue_tracker."""
 
+import copy
 import datetime
 import unittest
 from unittest import mock
@@ -154,6 +155,30 @@ def test_closed(self):
     self.assertEqual(
         datetime.datetime(2019, 6, 24, 6, 40, 7, 672), issue.closed_time)
 
+  def test_is_unrestricted_limit_none(self):
+    """Test is_unrestricted returns True when accessLevel is LIMIT_NONE."""
+    issue_data = copy.deepcopy(BASIC_ISSUE)
+    issue_data['issueState']['accessLimit'] = {'accessLevel': 'LIMIT_NONE'}
+    self.client.issues().get().execute.return_value = issue_data
+    issue = self.issue_tracker.get_issue(68828938)
+    self.assertTrue(issue.is_unrestricted)
+
+  def test_is_unrestricted_limit_view(self):
+    """Test is_unrestricted returns False when accessLevel is LIMIT_VIEW."""
+    issue_data = copy.deepcopy(BASIC_ISSUE)
+    issue_data['issueState']['accessLimit'] = {'accessLevel': 'LIMIT_VIEW'}
+    self.client.issues().get().execute.return_value = issue_data
+    issue = self.issue_tracker.get_issue(68828938)
+    self.assertFalse(issue.is_unrestricted)
+
+  def test_is_unrestricted_no_access_limit(self):
+    """Test is_unrestricted returns False when no accessLimit field."""
+    issue_data = copy.deepcopy(BASIC_ISSUE)
+    issue_data['issueState'].pop('accessLimit', None)
+    self.client.issues().get().execute.return_value = issue_data
+    issue = self.issue_tracker.get_issue(68828938)
+    self.assertFalse(issue.is_unrestricted)
+
   def test_get_labels(self):
     """Test getting labels."""
     self.client.issues().get().execute.return_value = {