Show Okta Enforced Repos when Impersonating (#786)

Co-authored-by: Michelle Tran <[email protected]>

Author: nora-shap

codecov_auth/middleware.py

@@ -82,6 +82,7 @@ def process_request(self, request: HttpRequest) -> None:
         if current_user and not current_user.is_anonymous:
             impersonating_ownerid = request.COOKIES.get("staff_user")
             if impersonating_ownerid is None:
+                request.impersonation = False
                 return
 
             log.info(
@@ -129,6 +130,9 @@ def process_request(self, request: HttpRequest) -> None:
                     impersonating_ownerid=impersonating_ownerid,
                 ),
             )
+            request.impersonation = True
+        else:
+            request.impersonation = False
 
 
 class CorsMiddleware(BaseCorsMiddleware):

codecov_auth/tests/unit/test_authentication.py

@@ -1,12 +1,14 @@
 from datetime import datetime, timedelta
 from http.cookies import SimpleCookie
+from unittest.mock import call, patch
 
 import pytest
 from django.conf import settings
 from django.test import TestCase, TransactionTestCase, override_settings
 from django.urls import ResolverMatch
 from rest_framework.exceptions import AuthenticationFailed, PermissionDenied
 from rest_framework.test import APIRequestFactory
+from shared.django_apps.core.tests.factories import RepositoryFactory
 
 from codecov_auth.authentication import (
     InternalTokenAuthentication,
@@ -168,7 +170,7 @@ def test_bearer_token_default_token_envar_and_same_string_as_header(self):
 class ImpersonationTests(TransactionTestCase):
     def setUp(self):
         self.owner_to_impersonate = OwnerFactory(
-            username="impersonateme", service="github"
+            username="impersonateme", service="github", user=UserFactory(is_staff=False)
         )
         self.staff_user = UserFactory(is_staff=True)
         self.non_staff_user = UserFactory(is_staff=False)
@@ -184,6 +186,47 @@ def test_impersonation(self):
         )
         assert res.json()["data"]["me"] == {"user": {"username": "impersonateme"}}
 
+    @patch("core.commands.repository.repository.RepositoryCommands.fetch_repository")
+    def test_impersonation_with_okta(self, mock_call_to_fetch_repository):
+        repo = RepositoryFactory(author=self.owner_to_impersonate, private=True)
+        query_repositories = """{ owner(username: "%s") { repository(name: "%s") { ... on Repository { name } } } }"""
+        query = query_repositories % (repo.author.username, repo.name)
+
+        # not impersonating
+        del self.client.cookies["staff_user"]
+        self.client.force_login(user=self.owner_to_impersonate.user)
+        self.client.post(
+            "/graphql/gh",
+            {"query": query},
+            content_type="application/json",
+        )
+
+        # impersonating, same query
+        self.client.cookies = SimpleCookie({"staff_user": self.owner_to_impersonate.pk})
+        self.client.force_login(user=self.staff_user)
+        self.client.post(
+            "/graphql/gh",
+            {"query": query},
+            content_type="application/json",
+        )
+
+        mock_call_to_fetch_repository.assert_has_calls(
+            [
+                call(
+                    self.owner_to_impersonate,
+                    repo.name,
+                    [],
+                    exclude_okta_enforced_repos=True,
+                ),
+                call(
+                    self.owner_to_impersonate,
+                    repo.name,
+                    [],
+                    exclude_okta_enforced_repos=False,
+                ),
+            ]
+        )
+
     def test_impersonation_non_staff(self):
         self.client.force_login(user=self.non_staff_user)
         with pytest.raises(PermissionDenied):

graphql_api/actions/repository.py

@@ -35,26 +35,35 @@ def list_repository_for_owner(
     owner: Owner,
     filters: dict[str, Any] | None,
     okta_account_auths: list[int],
+    exclude_okta_enforced_repos: bool = True,
 ) -> QuerySet:
+    queryset = Repository.objects.viewable_repos(current_owner)
+
+    if exclude_okta_enforced_repos:
+        queryset = queryset.exclude_accounts_enforced_okta(okta_account_auths)
+
     queryset = (
-        Repository.objects.viewable_repos(current_owner)
-        .exclude_accounts_enforced_okta(okta_account_auths)
-        .with_recent_coverage()
-        .with_latest_commit_at()
-        .filter(author=owner)
+        queryset.with_recent_coverage().with_latest_commit_at().filter(author=owner)
     )
+
     queryset = apply_filters_to_queryset(queryset, filters)
     return queryset
 
 
 def search_repos(
-    current_owner: Owner, filters: dict[str, Any] | None, okta_account_auths: list[int]
+    current_owner: Owner,
+    filters: dict[str, Any] | None,
+    okta_account_auths: list[int],
+    exclude_okta_enforced_repos: bool = True,
 ) -> QuerySet:
     authors_from = [current_owner.ownerid] + (current_owner.organizations or [])
+    queryset = Repository.objects.viewable_repos(current_owner)
+
+    if exclude_okta_enforced_repos:
+        queryset = queryset.exclude_accounts_enforced_okta(okta_account_auths)
+
     queryset = (
-        Repository.objects.viewable_repos(current_owner)
-        .exclude_accounts_enforced_okta(okta_account_auths)
-        .with_recent_coverage()
+        queryset.with_recent_coverage()
         .with_latest_commit_at()
         .filter(author__ownerid__in=authors_from)
     )

graphql_api/tests/helper.py

@@ -1,3 +1,7 @@
+from http.cookies import SimpleCookie
+
+from shared.django_apps.codecov_auth.tests.factories import OwnerFactory, UserFactory
+
 from codecov_auth.views.okta_cloud import OKTA_SIGNED_IN_ACCOUNTS_SESSION_KEY
 from utils.test_utils import Client
 
@@ -11,12 +15,21 @@ def gql_request(
         variables=None,
         with_errors=False,
         okta_signed_in_accounts=[],
+        impersonate_owner=False,
     ):
         url = f"/graphql/{provider}"
 
         if owner:
             self.client = Client()
-            self.client.force_login_owner(owner)
+
+            if impersonate_owner:
+                staff_owner = OwnerFactory(
+                    name="staff_user", service="github", user=UserFactory(is_staff=True)
+                )
+                self.client.cookies = SimpleCookie({"staff_user": owner.pk})
+                self.client.force_login_owner(staff_owner)
+            else:
+                self.client.force_login_owner(owner)
 
             if okta_signed_in_accounts:
                 session = self.client.session

graphql_api/tests/test_current_user_ariadne.py

@@ -187,7 +187,7 @@ def test_fetch_terms_agreement_and_business_email_when_owner_profile_is_null(sel
         assert data == {"me": {"businessEmail": None, "termsAgreement": False}}
 
     def test_fetch_null_terms_agreement_for_user_without_owner(self):
-        # There is an edge where a owner without user can call the me endpoint
+        # There is an edge where an owner without user can call the "me" endpoint
         # via impersonation, in that case return null for terms agreement
         owner_to_impersonate = OwnerFactory()
         owner_to_impersonate.user.delete()
@@ -281,6 +281,22 @@ def test_fetching_viewable_repositories(self):
             ]
         )
 
+        # Test with impersonation
+        data = self.gql_request(query, owner=current_user, impersonate_owner=True)
+        repos = paginate_connection(data["me"]["viewableRepositories"])
+        repos_name = [repo["name"] for repo in repos]
+        assert (
+            sorted(repos_name)
+            == [
+                "1",  # public repo in org of user
+                "2",  # private repo in org of user and in user permission
+                "6",  # personal private repo
+                "7",  # personal public repo
+                "okta_enforced_repo_authed",  # Okta repo should show up for impersonated users
+                "okta_enforced_repo_unauthed",  # Okta repo should show up for impersonated users
+            ]
+        )
+
     def test_fetching_viewable_repositories_ordering(self):
         current_user = OwnerFactory()
         query = """

graphql_api/tests/test_owner.py

@@ -61,7 +61,7 @@ def setUp(self):
             author=self.owner, active=True, activated=True, private=True, name="a"
         )
         RepositoryFactory(
-            author=self.owner, active=False, private=False, activated=False, name="b"
+            author=self.owner, active=False, activated=False, private=False, name="b"
         )
         RepositoryFactory(
             author=random_user, active=True, activated=False, private=True, name="not"
@@ -247,6 +247,12 @@ def test_fetching_repositories_filter_out_okta_enforced(self):
         repos = paginate_connection(data["owner"]["repositories"])
         assert repos == []
 
+    def test_fetching_repositories_impersonation_show_okta_enforced(self):
+        query = query_repositories % (self.owner.username, "", "")
+        data = self.gql_request(query, owner=self.owner, impersonate_owner=True)
+        repos = paginate_connection(data["owner"]["repositories"])
+        assert repos == [{"name": "a"}, {"name": "b"}]
+
     def test_is_part_of_org_when_unauthenticated(self):
         query = query_repositories % (self.owner.username, "", "")
         data = self.gql_request(query)

graphql_api/types/me/me.py

@@ -53,7 +53,14 @@ def resolve_viewable_repositories(
     okta_authenticated_accounts: list[int] = info.context["request"].session.get(
         OKTA_SIGNED_IN_ACCOUNTS_SESSION_KEY, []
     )
-    queryset = search_repos(current_user, filters, okta_authenticated_accounts)
+    is_impersonation = info.context["request"].impersonation
+    # If the user is impersonating another user, we want to show all the Okta repos.
+    # This means we do not want to filter out the Okta enforced repos
+    exclude_okta_enforced_repos = not is_impersonation
+
+    queryset = search_repos(
+        current_user, filters, okta_authenticated_accounts, exclude_okta_enforced_repos
+    )
     return queryset_to_connection(
         queryset,
         ordering=(ordering, RepositoryOrdering.ID),

graphql_api/types/owner/owner.py

@@ -55,8 +55,14 @@ def resolve_repositories(
     okta_account_auths: list[int] = info.context["request"].session.get(
         OKTA_SIGNED_IN_ACCOUNTS_SESSION_KEY, []
     )
+
+    is_impersonation = info.context["request"].impersonation
+    # If the user is impersonating another user, we want to show all the Okta repos.
+    # This means we do not want to filter out the Okta enforced repos
+    exclude_okta_enforced_repos = not is_impersonation
+
     queryset = list_repository_for_owner(
-        current_owner, owner, filters, okta_account_auths
+        current_owner, owner, filters, okta_account_auths, exclude_okta_enforced_repos
     )
     return queryset_to_connection(
         queryset,
@@ -126,8 +132,17 @@ async def resolve_repository(owner: Owner, info, name):
     okta_authenticated_accounts: list[int] = info.context["request"].session.get(
         OKTA_SIGNED_IN_ACCOUNTS_SESSION_KEY, []
     )
+
+    is_impersonation = info.context["request"].impersonation
+    # If the user is impersonating another user, we want to show all the Okta repos.
+    # This means we do not want to filter out the Okta enforced repos
+    exclude_okta_enforced_repos = not is_impersonation
+
     repository: Optional[Repository] = await command.fetch_repository(
-        owner, name, okta_authenticated_accounts
+        owner,
+        name,
+        okta_authenticated_accounts,
+        exclude_okta_enforced_repos=exclude_okta_enforced_repos,
     )
 
     if repository is None: