Require verified email for all authorization gates, not just domain trust

src/appengine/libs/access.py

@@ -138,15 +138,22 @@ def get_access(need_privileged_access=False, job_type=None, fuzzer_name=None):
   if _is_blacklisted_user(email):
     return UserAccess.Denied
 
-  if _is_privileged_user(email):
-    return UserAccess.Allowed
-
-  if job_type and external_users.is_job_allowed_for_user(email, job_type):
-    return UserAccess.Allowed
-
-  if (fuzzer_name and
-      external_users.is_fuzzer_allowed_for_user(email, fuzzer_name)):
-    return UserAccess.Allowed
+  # Only trust the email for identity-based authorization when the identity
+  # provider has verified it.  An unverified email (e.g. a GitHub Enterprise
+  # Managed User address) can be asserted arbitrarily and must not be matched
+  # against privileged_users, external-user job/fuzzer ACLs, or whitelisted
+  # domains.  _is_trusted_domain_user already enforces this; extend the same
+  # guard to every other email-identity gate.
+  if user.email_verified:
+    if _is_privileged_user(email):
+      return UserAccess.Allowed
+
+    if job_type and external_users.is_job_allowed_for_user(email, job_type):
+      return UserAccess.Allowed
+
+    if (fuzzer_name and
+        external_users.is_fuzzer_allowed_for_user(email, fuzzer_name)):
+      return UserAccess.Allowed
 
   if not need_privileged_access and _is_trusted_domain_user(user):
     return UserAccess.Allowed
@@ -166,7 +173,15 @@ def can_user_access_testcase(testcase):
       need_privileged_access=need_privileged_access):
     return True
 
-  user_email = helpers.get_user_email()
+  # Email-equality checks below (uploader, assignee, CC, reporter) rely on
+  # the email being verified.  An unverified federated email can be set to any
+  # value (e.g. GitHub EMU) and must not grant access to another user's
+  # test cases or embargoed reproducers.
+  user = auth.get_current_user()
+  if not user or not user.email_verified:
+    return False
+
+  user_email = user.email
   if testcase.uploader_email and testcase.uploader_email == user_email:
     return True
 
@@ -189,8 +204,7 @@ def can_user_access_testcase(testcase):
       issues_to_check.append(original_issue)
 
   relaxed_restrictions = (
-      config.relax_testcase_restrictions or
-      _is_trusted_domain_user(auth.get_current_user()))
+      config.relax_testcase_restrictions or _is_trusted_domain_user(user))
   for issue in issues_to_check:
     if relaxed_restrictions:
       if (any(utils.emails_equal(user_email, cc) for cc in issue.ccs) or

src/clusterfuzz/_internal/tests/appengine/libs/access_test.py

@@ -272,6 +272,47 @@ def test_get_access_unverified_domain(self):
         access.get_access(need_privileged_access=False),
         access.UserAccess.Denied)
 
+  def test_get_access_unverified_privileged(self):
+    """An unverified email matching privileged_users must NOT be granted
+    privileged access (GitHub EMU can assert any email)."""
+    self.mock.get_current_user.return_value = auth.User(
+        'test@test.com', email_verified=False)
+    self.mock.is_current_user_admin.return_value = False
+    self.mock._is_privileged_user.return_value = True
+    self.mock._is_domain_allowed.return_value = False
+    self.mock.is_fuzzer_allowed_for_user.return_value = False
+    self.mock.is_job_allowed_for_user.return_value = False
+    self.assertEqual(
+        access.get_access(need_privileged_access=True),
+        access.UserAccess.Denied)
+    self.assertEqual(
+        access.get_access(need_privileged_access=False),
+        access.UserAccess.Denied)
+
+  def test_get_access_unverified_external_job(self):
+    """An unverified email must not gain job-scoped access."""
+    self.mock.get_current_user.return_value = auth.User(
+        'test@test.com', email_verified=False)
+    self.mock.is_current_user_admin.return_value = False
+    self.mock._is_privileged_user.return_value = False
+    self.mock._is_domain_allowed.return_value = False
+    self.mock.is_fuzzer_allowed_for_user.return_value = False
+    self.mock.is_job_allowed_for_user.return_value = True
+    self.assertEqual(
+        access.get_access(job_type='test'), access.UserAccess.Denied)
+
+  def test_get_access_unverified_external_fuzzer(self):
+    """An unverified email must not gain fuzzer-scoped access."""
+    self.mock.get_current_user.return_value = auth.User(
+        'test@test.com', email_verified=False)
+    self.mock.is_current_user_admin.return_value = False
+    self.mock._is_privileged_user.return_value = False
+    self.mock._is_domain_allowed.return_value = False
+    self.mock.is_fuzzer_allowed_for_user.return_value = True
+    self.mock.is_job_allowed_for_user.return_value = False
+    self.assertEqual(
+        access.get_access(fuzzer_name='test'), access.UserAccess.Denied)
+
   def test_get_access_external_fuzzer(self):
     """For a fuzzer, ensure it allows when a user is allowed."""
     self.mock.get_current_user.return_value = self.user
@@ -479,6 +520,23 @@ def test_allowed_because_of_uploader(self):
 
     self.assertTrue(access.can_user_access_testcase(self.testcase))
 
+  def test_denied_unverified_uploader(self):
+    """An unverified email matching the uploader must NOT grant access."""
+    self.mock.get_current_user.return_value = auth.User(
+        self.email, email_verified=False)
+    self.testcase.uploader_email = self.email
+    self.testcase.security_flag = True
+    self.assertFalse(access.can_user_access_testcase(self.testcase))
+
+  def test_denied_unverified_bug_owner(self):
+    """An unverified email matching a bug assignee must NOT grant access."""
+    self.mock.get_current_user.return_value = auth.User(
+        self.email, email_verified=False)
+    self.bug.owner = self.email
+    self.get_issue.return_value = self.bug
+    self.testcase.bug_information = '1234'
+    self.assertFalse(access.can_user_access_testcase(self.testcase))
+
   def test_allowed_because_of_owner_in_original_issue(self):
     """Ensure it is allowed because the user is the owner of original issue."""
     self.bug.merged_into = 5678