feat: Common sharing permissions (#20261)

ee/clickhouse/views/groups.py

@@ -12,14 +12,10 @@
 from ee.clickhouse.queries.related_actors_query import RelatedActorsQuery
 from posthog.api.documentation import extend_schema
 from posthog.api.routing import TeamAndOrgViewSetMixin
-from posthog.auth import SharingAccessTokenAuthentication
 from posthog.clickhouse.kafka_engine import trim_quotes_expr
 from posthog.client import sync_execute
 from posthog.models.group import Group
 from posthog.models.group_type_mapping import GroupTypeMapping
-from posthog.permissions import (
-    SharingTokenPermission,
-)
 
 
 class GroupTypeSerializer(serializers.ModelSerializer):
@@ -35,14 +31,6 @@ class ClickhouseGroupsTypesView(TeamAndOrgViewSetMixin, mixins.ListModelMixin, v
     pagination_class = None
     sharing_enabled_actions = ["list"]
 
-    def get_permissions(self):
-        if isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication):
-            return [SharingTokenPermission()]
-        return super().get_permissions()
-
-    def get_authenticators(self):
-        return [SharingAccessTokenAuthentication(), *super().get_authenticators()]
-
     @action(detail=False, methods=["PATCH"], name="Update group types metadata")
     def update_metadata(self, request: request.Request, *args, **kwargs):
         for row in cast(List[Dict], request.data):

ee/clickhouse/views/test/test_clickhouse_groups.py

@@ -434,6 +434,26 @@ def test_cannot_list_group_types_of_another_org(self):
             self.permission_denied_response("You don't have access to the project."),
         )
 
+    def test_cannot_list_group_types_of_another_org_with_sharing_token(self):
+        sharing_configuration = SharingConfiguration.objects.create(team=self.team, enabled=True)
+
+        other_org = Organization.objects.create(name="other org")
+        other_team = Team.objects.create(organization=other_org, name="other project")
+
+        GroupTypeMapping.objects.create(team=other_team, group_type="organization", group_type_index=0)
+        GroupTypeMapping.objects.create(team=other_team, group_type="playlist", group_type_index=1)
+        GroupTypeMapping.objects.create(team=other_team, group_type="another", group_type_index=2)
+
+        response = self.client.get(
+            f"/api/projects/{other_team.id}/groups_types/?sharing_access_token={sharing_configuration.access_token}"
+        )
+
+        self.assertEqual(response.status_code, 403, response.json())
+        self.assertEqual(
+            response.json(),
+            self.permission_denied_response("You do not have permission to perform this action."),
+        )
+
     def test_can_list_group_types_of_another_org_with_sharing_access_token(self):
         other_org = Organization.objects.create(name="other org")
         other_team = Team.objects.create(organization=other_org, name="other project")

posthog/api/insight.py

@@ -574,6 +574,7 @@ class InsightViewSet(
     filter_backends = [DjangoFilterBackend]
     filterset_fields = ["short_id", "created_by"]
     include_in_docs = True
+    sharing_enabled_actions = ["retrieve", "list"]
 
     retention_query_class = Retention
     stickiness_query_class = Stickiness
@@ -588,23 +589,11 @@ def get_serializer_class(self) -> Type[serializers.BaseSerializer]:
             return InsightBasicSerializer
         return super().get_serializer_class()
 
-    def get_authenticators(self):
-        return [SharingAccessTokenAuthentication(), *super().get_authenticators()]
-
     def get_serializer_context(self) -> Dict[str, Any]:
         context = super().get_serializer_context()
         context["is_shared"] = isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication)
         return context
 
-    def get_permissions(self):
-        if isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication) and self.action in (
-            "retrieve",
-            "list",
-        ):
-            # Anonymous users authenticated via SharingAccessTokenAuthentication get read-only access to insights
-            return []
-        return super().get_permissions()
-
     def get_queryset(self) -> QuerySet:
         queryset: QuerySet
         if isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication):

posthog/api/routing.py

@@ -9,11 +9,11 @@
 from rest_framework_extensions.settings import extensions_api_settings
 
 from posthog.api.utils import get_token
-from posthog.auth import JwtAuthentication, PersonalAPIKeyAuthentication
+from posthog.auth import JwtAuthentication, PersonalAPIKeyAuthentication, SharingAccessTokenAuthentication
 from posthog.models.organization import Organization
 from posthog.models.team import Team
 from posthog.models.user import User
-from posthog.permissions import OrganizationMemberPermissions, TeamMemberAccessPermission
+from posthog.permissions import OrganizationMemberPermissions, SharingTokenPermission, TeamMemberAccessPermission
 from posthog.user_permissions import UserPermissions
 
 if TYPE_CHECKING:
@@ -45,9 +45,14 @@ class TeamAndOrgViewSetMixin(_GenericViewSet):
     authentication_classes = []
     permission_classes = []
 
+    sharing_enabled_actions: list[str] = []
+
     # We want to try and ensure that the base permission and authentication are always used
     # so we offer a way to add additional classes
     def get_permissions(self):
+        if isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication):
+            return [SharingTokenPermission()]
+
         # NOTE: We define these here to make it hard _not_ to use them. If you want to override them, you have to
         # override the entire method.
         permission_classes: list = [IsAuthenticated]
@@ -64,11 +69,19 @@ def get_authenticators(self):
         # NOTE: Custom authentication_classes go first as these typically have extra initial checks
         authentication_classes: list = [
             *self.authentication_classes,
-            JwtAuthentication,
-            PersonalAPIKeyAuthentication,
-            authentication.SessionAuthentication,
         ]
 
+        if self.sharing_enabled_actions:
+            authentication_classes.append(SharingAccessTokenAuthentication)
+
+        authentication_classes.extend(
+            [
+                JwtAuthentication,
+                PersonalAPIKeyAuthentication,
+                authentication.SessionAuthentication,
+            ]
+        )
+
         return [auth() for auth in authentication_classes]
 
     def get_queryset(self):

posthog/models/sharing_configuration.py

@@ -1,8 +1,10 @@
 import secrets
-from typing import List
+from typing import List, cast
 
 from django.db import models
 
+from posthog.models.insight import Insight
+
 
 def get_default_access_token() -> str:
     return secrets.token_urlsafe(22)
@@ -37,6 +39,9 @@ def can_access_object(self, obj: models.Model):
         if obj.team_id != self.team_id:  # type: ignore
             return False
 
+        if obj._meta.object_name == "Insight" and self.dashboard:
+            return cast(Insight, obj).id in self.get_connected_insight_ids()
+
         for comparison in [self.insight, self.dashboard, self.recording]:
             if comparison and comparison == obj:
                 return True

posthog/permissions.py

@@ -1,6 +1,7 @@
 from typing import cast
 
 from django.db.models import Model
+from rest_framework.exceptions import NotFound
 from rest_framework.permissions import SAFE_METHODS, BasePermission, IsAdminUser
 from rest_framework.request import Request
 from rest_framework.views import APIView
@@ -231,6 +232,13 @@ def has_permission(self, request, view) -> bool:
         ), "SharingTokenPermission requires the `sharing_enabled_actions` attribute to be set in the view"
 
         if isinstance(request.successful_authenticator, SharingAccessTokenAuthentication):
+            try:
+                view.team  # noqa: B018
+                if request.successful_authenticator.sharing_configuration.team != view.team:
+                    return False
+            except NotFound:
+                return False
+
             return view.action in view.sharing_enabled_actions
 
         return False

posthog/session_recordings/session_recording_api.py

@@ -27,9 +27,6 @@
 from posthog.models.filters.session_recordings_filter import SessionRecordingsFilter
 from posthog.models.person.person import PersonDistinctId
 from posthog.session_recordings.models.session_recording import SessionRecording
-from posthog.permissions import (
-    SharingTokenPermission,
-)
 from posthog.session_recordings.models.session_recording_event import (
     SessionRecordingViewed,
 )
@@ -186,14 +183,6 @@ class SessionRecordingViewSet(TeamAndOrgViewSetMixin, viewsets.GenericViewSet):
 
     sharing_enabled_actions = ["retrieve", "snapshots", "snapshot_file"]
 
-    def get_permissions(self):
-        if isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication):
-            return [SharingTokenPermission()]
-        return super().get_permissions()
-
-    def get_authenticators(self):
-        return [SharingAccessTokenAuthentication(), *super().get_authenticators()]
-
     def get_serializer_class(self) -> Type[serializers.Serializer]:
         if isinstance(self.request.successful_authenticator, SharingAccessTokenAuthentication):
             return SessionRecordingSharedSerializer