diff --git a/src/appengine/libs/access.py b/src/appengine/libs/access.py index 8cf8fa534cb..6b1d15fad8a 100644 --- a/src/appengine/libs/access.py +++ b/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 diff --git a/src/clusterfuzz/_internal/tests/appengine/libs/access_test.py b/src/clusterfuzz/_internal/tests/appengine/libs/access_test.py index 3cf21155eac..b439e7500a8 100644 --- a/src/clusterfuzz/_internal/tests/appengine/libs/access_test.py +++ b/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