fix: activities on private object not visible by owner

udata/core/activity/api.py

@@ -101,18 +101,16 @@ def get(self):
         # Always return a result even not complete
         # But log the error (ie. visible in sentry, silent for user)
         # Can happen when someone manually delete an object in DB (ie. without proper purge)
-        # - Filter out private items (except for sysadmin users)
+        # - Filter out items not visible to the current user
         safe_items = []
         for item in qs.queryset.items:
             try:
                 item.related_to
             except DoesNotExist as e:
                 log.error(e, exc_info=True)
             else:
-                if hasattr(item.related_to, "private") and (
-                    current_user.is_anonymous or not current_user.sysadmin
-                ):
-                    if item.related_to.private:
+                if hasattr(item.related_to, "is_visible_by"):
+                    if not item.related_to.is_visible_by(current_user):
                         continue
                 safe_items.append(item)
         qs.queryset.items = safe_items

udata/core/owned.py

@@ -1,6 +1,7 @@
 import logging
 
 from blinker import signal
+from flask_login import AnonymousUserMixin
 from mongoengine import NULLIFY, Q, post_save
 from mongoengine.fields import ReferenceField
 
@@ -110,6 +111,33 @@ class Owned(object):
         "queryset_class": OwnedQuerySet,
     }
 
+    def is_visible_by(self, user: User | AnonymousUserMixin) -> bool:
+        """
+        Check if this object is visible to the given user.
+        An object is visible if:
+        - It's not private, OR
+        - The user is a sysadmin, OR
+        - The user owns it or is a member of the owning organization
+
+        Models using this method must have a `private` attribute.
+        """
+        if not getattr(self, "private", False):
+            return True
+
+        if user.is_anonymous:
+            return False
+
+        if user.sysadmin:
+            return True
+
+        if self.owner and self.owner.id == user.id:
+            return True
+
+        if self.organization and self.organization in user.organizations:
+            return True
+
+        return False
+
     def clean(self):
         """
         Verify owner consistency and fetch original owner before the new one erase it.

udata/tests/api/test_activities_api.py

@@ -4,6 +4,7 @@
 from udata.core.activity.models import Activity
 from udata.core.dataset.factories import DatasetFactory
 from udata.core.dataset.models import Dataset
+from udata.core.organization.factories import OrganizationFactory
 from udata.core.reuse.factories import ReuseFactory
 from udata.core.reuse.models import Reuse
 from udata.core.topic.factories import TopicFactory
@@ -111,3 +112,38 @@ def test_activity_api_with_topic(self) -> None:
         assert activity_data["related_to_id"] == str(topic.id)
         assert activity_data["related_to_kind"] == "Topic"
         assert activity_data["related_to_url"] == topic.self_api_url()
+
+    def test_activity_api_list_with_private_visible_to_owner(self) -> None:
+        """Owner should see activities about their own private objects."""
+        owner = UserFactory()
+        dataset = DatasetFactory(private=True, owner=owner)
+        FakeDatasetActivity.objects.create(actor=UserFactory(), related_to=dataset)
+
+        # Anonymous user won't see it
+        response = self.get(url_for("api.activity"))
+        assert200(response)
+        assert len(response.json["data"]) == 0
+
+        # Owner should see their own private dataset activity
+        self.login(owner)
+        response = self.get(url_for("api.activity"))
+        assert200(response)
+        assert len(response.json["data"]) == 1
+
+    def test_activity_api_list_with_private_visible_to_org_member(self) -> None:
+        """Organization members should see activities about their org's private objects."""
+        member = UserFactory()
+        org = OrganizationFactory(admins=[member])
+        dataset = DatasetFactory(private=True, organization=org)
+        FakeDatasetActivity.objects.create(actor=UserFactory(), related_to=dataset)
+
+        # Anonymous user won't see it
+        response = self.get(url_for("api.activity"))
+        assert200(response)
+        assert len(response.json["data"]) == 0
+
+        # Org member should see the private dataset activity
+        self.login(member)
+        response = self.get(url_for("api.activity"))
+        assert200(response)
+        assert len(response.json["data"]) == 1