Make live updates honour user_rel restrictions, also allow superusers to see everything.

dddp/api.py

@@ -4,6 +4,7 @@ import collections
 import traceback
 import dbarray
 from django.conf import settings
+from django.contrib.auth import get_user_model
 from django.db import connection, transaction
 from django.db.models import aggregates, Q
 from django.db.models.sql import aggregates as sql_aggregates
@@ -210,7 +211,7 @@ class Collection(APIMixin):
             )
         return qs
 
-    def user_ids_for_object(self, obj, base_qs=None):
+    def user_ids_for_object(self, obj, base_qs=None, include_superusers=True):
         """Find user IDs related to object/pk in queryset."""
         qs = base_qs or self.queryset
         if self.user_rel:
@@ -222,12 +223,21 @@ class Collection(APIMixin):
                 for index, user_rel
                 in enumerate(user_rels)
             }
+
             user_ids = set()
+            if include_superusers:
+                user_ids.update(
+                    get_user_model().objects.filter(
+                        is_superuser=True, is_active=True,
+                    ).values_list('pk', flat=True)
+                )
+
             for rel_user_ids in qs.filter(
                     pk=hasattr(obj, 'pk') and obj.pk or obj,
             ).annotate(**user_rel_map).values_list(*user_rel_map.keys()).get():
                 user_ids.update(rel_user_ids)
-            return sorted(user_ids.difference([None]))
+            user_ids.difference_update([None])
+            return user_ids
         else:
             return None
 
@@ -390,7 +400,7 @@ class DDP(APIMixin):
     @api_endpoint
     def sub(self, id_, name, *params):
         """Create subscription, send matched objects that haven't been sent."""
-        return self._sub( id_, name, *params)
+        return self._sub(id_, name, *params)
 
     @transaction.atomic
     def _sub(self, id_, name, *params):

dddp/notify.py

@@ -22,6 +22,7 @@ _CLS_CACHE = ImportCache()
 
 def send_notify(model, obj, msg, using):
     """Dispatch PostgreSQL async NOTIFY."""
+    user_cache = {}
     col_name = collection_name(model)
     if col_name == 'migrations.migration':
         return  # never send migration models.
@@ -44,6 +45,18 @@ def send_notify(model, obj, msg, using):
             qs = pub_queries[sub_col.name]
             col = _CLS_CACHE[sub_col.collection_class]()
             # filter qs using user_rel paths on collection
+            try:
+                user_ids = user_cache[sub_col.collection_class]
+            except KeyError:
+                user_ids = user_cache.setdefault(
+                    sub_col.collection_class, col.user_ids_for_object(obj)
+                )
+
+            if user_ids is None:
+                pass  # unrestricted collection, anyone can see.
+            elif sub.user_id not in user_ids:
+                continue  # not for this user
+
             qs = col.get_queryset(qs)
             if qs.filter(pk=obj.pk).exists():
                 sub_ids.add(sub.sub_id)