fix(api): Apply org/team key scoping more narrowly (#25634)

posthog/api/organization.py

@@ -10,6 +10,7 @@
 from posthog import settings
 from posthog.api.routing import TeamAndOrgViewSetMixin
 from posthog.api.shared import ProjectBasicSerializer, TeamBasicSerializer
+from posthog.auth import PersonalAPIKeyAuthentication
 from posthog.cloud_utils import is_cloud
 from posthog.constants import INTERNAL_BOT_EMAIL_SUFFIX, AvailableFeature
 from posthog.event_usage import report_organization_deleted
@@ -178,7 +179,12 @@ def dangerously_get_permissions(self):
         raise NotImplementedError()
 
     def safely_get_queryset(self, queryset) -> QuerySet:
-        return cast(User, self.request.user).organizations.all()
+        user = cast(User, self.request.user)
+        queryset = user.organizations.all()
+        if isinstance(self.request.successful_authenticator, PersonalAPIKeyAuthentication):
+            if scoped_organizations := self.request.successful_authenticator.personal_api_key.scoped_organizations:
+                queryset = queryset.filter(id__in=scoped_organizations)
+        return queryset
 
     def safely_get_object(self, queryset):
         return self.organization

posthog/api/project.py

@@ -15,6 +15,7 @@
     TeamSerializer,
     validate_team_attrs,
 )
+from posthog.auth import PersonalAPIKeyAuthentication
 from posthog.event_usage import report_user_action
 from posthog.geoip import get_geoip_properties
 from posthog.jwt import PosthogJwtAudience, encode_jwt
@@ -397,6 +398,10 @@ class ProjectViewSet(TeamAndOrgViewSetMixin, viewsets.ModelViewSet):
     def safely_get_queryset(self, queryset):
         # IMPORTANT: This is actually what ensures that a user cannot read/update a project for which they don't have permission
         visible_teams_ids = UserPermissions(cast(User, self.request.user)).team_ids_visible_for_user
+        queryset = queryset.filter(id__in=visible_teams_ids)
+        if isinstance(self.request.successful_authenticator, PersonalAPIKeyAuthentication):
+            if scoped_organizations := self.request.successful_authenticator.personal_api_key.scoped_organizations:
+                queryset = queryset.filter(organization_id__in=scoped_organizations)
         return queryset.filter(id__in=visible_teams_ids)
 
     def get_serializer_class(self) -> type[serializers.BaseSerializer]:

posthog/api/team.py

@@ -6,6 +6,8 @@
 
 from django.shortcuts import get_object_or_404
 from loginas.utils import is_impersonated_session
+from posthog.auth import PersonalAPIKeyAuthentication
+from posthog.jwt import PosthogJwtAudience, encode_jwt
 from rest_framework import exceptions, request, response, serializers, viewsets
 from rest_framework.permissions import BasePermission, IsAuthenticated
 
@@ -15,7 +17,6 @@
 from posthog.constants import AvailableFeature
 from posthog.event_usage import report_user_action
 from posthog.geoip import get_geoip_properties
-from posthog.jwt import PosthogJwtAudience, encode_jwt
 from posthog.models import ProductIntent, Team, User
 from posthog.models.activity_logging.activity_log import (
     Detail,
@@ -423,9 +424,16 @@ class TeamViewSet(TeamAndOrgViewSetMixin, viewsets.ModelViewSet):
     ordering = "-created_by"
 
     def safely_get_queryset(self, queryset):
+        user = cast(User, self.request.user)
         # IMPORTANT: This is actually what ensures that a user cannot read/update a project for which they don't have permission
-        visible_teams_ids = UserPermissions(cast(User, self.request.user)).team_ids_visible_for_user
-        return queryset.filter(id__in=visible_teams_ids)
+        visible_teams_ids = UserPermissions(user).team_ids_visible_for_user
+        queryset = queryset.filter(id__in=visible_teams_ids)
+        if isinstance(self.request.successful_authenticator, PersonalAPIKeyAuthentication):
+            if scoped_organizations := self.request.successful_authenticator.personal_api_key.scoped_organizations:
+                queryset = queryset.filter(project__organization_id__in=scoped_organizations)
+            if scoped_teams := self.request.successful_authenticator.personal_api_key.scoped_teams:
+                queryset = queryset.filter(id__in=scoped_teams)
+        return queryset
 
     def get_serializer_class(self) -> type[serializers.BaseSerializer]:
         if self.action == "list":

posthog/api/test/test_organization.py

@@ -1,7 +1,8 @@
-from asgiref.sync import sync_to_async
 from rest_framework import status
 
 from posthog.models import Organization, OrganizationMembership, Team
+from posthog.models.personal_api_key import PersonalAPIKey, hash_key_value
+from posthog.models.utils import generate_random_token_personal
 from posthog.test.base import APIBaseTest
 
 
@@ -138,20 +139,31 @@ def test_enforce_2fa_for_everyone(self):
         self.organization.refresh_from_db()
         self.assertEqual(self.organization.enforce_2fa, True)
 
+    def test_projects_outside_personal_api_key_scoped_organizations_not_listed(self):
+        other_org, _, _ = Organization.objects.bootstrap(self.user)
+        personal_api_key = generate_random_token_personal()
+        PersonalAPIKey.objects.create(
+            label="X",
+            user=self.user,
+            last_used_at="2021-08-25T21:09:14",
+            secure_value=hash_key_value(personal_api_key),
+            scoped_organizations=[other_org.id],
+        )
 
-def create_organization(name: str) -> Organization:
-    """
-    Helper that just creates an organization. It currently uses the orm, but we
-    could use either the api, or django admin to create, to get better parity
-    with real world scenarios.
-    """
-    return Organization.objects.create(name=name)
+        response = self.client.get("/api/organizations/", HTTP_AUTHORIZATION=f"Bearer {personal_api_key}")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {org["id"] for org in response.json()["results"]},
+            {str(other_org.id)},
+            "Only the scoped organization should be listed, the other one should be excluded",
+        )
 
 
-async def acreate_organization(name: str) -> Organization:
+def create_organization(name: str) -> Organization:
     """
     Helper that just creates an organization. It currently uses the orm, but we
     could use either the api, or django admin to create, to get better parity
     with real world scenarios.
     """
-    return await sync_to_async(create_organization)(name)
+    return Organization.objects.create(name=name)

posthog/api/test/test_personal_api_keys.py

@@ -483,7 +483,8 @@ def test_allows_access_to_scoped_org_teams(self):
 
     def test_denies_access_to_non_scoped_org_and_team(self):
         response = self._do_request(f"/api/organizations/{self.other_organization.id}")
-        assert response.status_code == status.HTTP_403_FORBIDDEN, response.json()
+        # In the organizations endpoint this is a 404s, as we filter out at the queryset level
+        assert response.status_code == status.HTTP_404_NOT_FOUND, response.json()
         response = self._do_request(f"/api/projects/{self.other_team.id}/feature_flags")
         assert response.status_code == status.HTTP_403_FORBIDDEN, response.json()
 

posthog/api/test/test_project.py

@@ -1,4 +1,8 @@
 from posthog.api.test.test_team import EnvironmentToProjectRewriteClient, team_api_test_factory
+from posthog.models.organization import Organization
+from posthog.models.personal_api_key import PersonalAPIKey, hash_key_value
+from posthog.models.utils import generate_random_token_personal
+from rest_framework import status
 
 
 class TestProjectAPI(team_api_test_factory()):  # type: ignore
@@ -9,3 +13,23 @@ class TestProjectAPI(team_api_test_factory()):  # type: ignore
     """
 
     client_class = EnvironmentToProjectRewriteClient
+
+    def test_projects_outside_personal_api_key_scoped_organizations_not_listed(self):
+        other_org, _, team_in_other_org = Organization.objects.bootstrap(self.user)
+        personal_api_key = generate_random_token_personal()
+        PersonalAPIKey.objects.create(
+            label="X",
+            user=self.user,
+            last_used_at="2021-08-25T21:09:14",
+            secure_value=hash_key_value(personal_api_key),
+            scoped_organizations=[other_org.id],
+        )
+
+        response = self.client.get("/api/projects/", HTTP_AUTHORIZATION=f"Bearer {personal_api_key}")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {project["id"] for project in response.json()["results"]},
+            {team_in_other_org.project.id},
+            "Only the project belonging to the scoped organization should be listed, the other one should be excluded",
+        )

posthog/api/test/test_team.py

@@ -18,8 +18,11 @@
 from posthog.models.dashboard import Dashboard
 from posthog.models.instance_setting import get_instance_setting
 from posthog.models.organization import Organization, OrganizationMembership
+from posthog.models.personal_api_key import PersonalAPIKey, hash_key_value
 from posthog.models.product_intent import ProductIntent
+from posthog.models.project import Project
 from posthog.models.team import Team
+from posthog.models.utils import generate_random_token_personal
 from posthog.temporal.common.client import sync_connect
 from posthog.temporal.common.schedule import describe_schedule
 from posthog.test.base import APIBaseTest
@@ -1147,7 +1150,7 @@ def create_team(organization: Organization, name: str = "Test team", timezone: s
     """
     This is a helper that just creates a team. It currently uses the orm, but we
     could use either the api, or django admin to create, to get better parity
-    with real world  scenarios.
+    with real world scenarios.
     """
     return Team.objects.create(
         organization=organization,
@@ -1160,4 +1163,45 @@ def create_team(organization: Organization, name: str = "Test team", timezone: s
 
 
 class TestTeamAPI(team_api_test_factory()):  # type: ignore
-    pass
+    def test_teams_outside_personal_api_key_scoped_teams_not_listed(self):
+        other_team_in_project = Team.objects.create(organization=self.organization, project=self.project)
+        _, team_in_other_project = Project.objects.create_with_team(
+            organization=self.organization, initiating_user=self.user
+        )
+        personal_api_key = generate_random_token_personal()
+        PersonalAPIKey.objects.create(
+            label="X",
+            user=self.user,
+            last_used_at="2021-08-25T21:09:14",
+            secure_value=hash_key_value(personal_api_key),
+            scoped_teams=[other_team_in_project.id],
+        )
+
+        response = self.client.get("/api/environments/", HTTP_AUTHORIZATION=f"Bearer {personal_api_key}")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {team["id"] for team in response.json()["results"]},
+            {other_team_in_project.id},
+            "Only the scoped team listed here, the other two should be excluded",
+        )
+
+    def test_teams_outside_personal_api_key_scoped_organizations_not_listed(self):
+        other_org, __, team_in_other_org = Organization.objects.bootstrap(self.user)
+        personal_api_key = generate_random_token_personal()
+        PersonalAPIKey.objects.create(
+            label="X",
+            user=self.user,
+            last_used_at="2021-08-25T21:09:14",
+            secure_value=hash_key_value(personal_api_key),
+            scoped_organizations=[other_org.id],
+        )
+
+        response = self.client.get("/api/environments/", HTTP_AUTHORIZATION=f"Bearer {personal_api_key}")
+
+        self.assertEqual(response.status_code, status.HTTP_200_OK)
+        self.assertEqual(
+            {team["id"] for team in response.json()["results"]},
+            {team_in_other_org.id},
+            "Only the team belonging to the scoped organization should be listed, the other one should be excluded",
+        )