Some fixes for viewer role

admin_site/wagtail_hooks.py

@@ -53,7 +53,7 @@ class InstanceItem(MenuItem):
 class InstanceChooserMenu(Menu):
     def menu_items_for_request(self, request: PathsAdminRequest):
         user = request.user
-        instances = InstanceConfig.permission_policy().instances_user_has_any_permission_for(user, ['change'])
+        instances = user.get_adminable_instances()
         if len(instances) < 2:
             return []
         items = []

datasets/tests/test_dataset_authorization.py

@@ -7,6 +7,7 @@
 from kausal_common.datasets.models import Dataset, DatasetSchema
 
 from paths.admin_context import set_admin_instance
+from paths.context import RealmContext, realm_context
 
 from admin_site.dataset_admin import DatasetSchemaViewSet
 from nodes.models import InstanceConfig
@@ -24,7 +25,6 @@ def get(user: User, view, url: str, instance_config: InstanceConfig, kwargs: dic
         if not user.selected_instance:
             user.selected_instance = instance_config
             user.save()
-        from paths.context import RealmContext, realm_context
         set_admin_instance(instance_config, request=request)
         ctx = RealmContext(realm = instance_config, user=request.user)
         with realm_context.activate(ctx):
@@ -142,7 +142,10 @@ def test_dataset_schema_index_view(self, client, setup_test_data, user_key, expe
                 response = get_in_admin_context(user, view, url, data['instance1'])
             return
 
-        response.render()
+        ctx = RealmContext(realm = data['instance1'], user=user)
+        with realm_context.activate(ctx):
+            response.render()
+
         assert response.status_code == 200
         content = response.content.decode('utf-8')
         for schema_name in expected_schemas:
@@ -236,7 +239,10 @@ def test_dataset_index_view(self, client, setup_test_data, user_key, expected_da
 
         for instance_key, dataset_key in expected_datasets:
             response = get_in_admin_context(user, view, url, data[instance_key])
-            response.render()
+            ctx = RealmContext(realm = data[instance_key], user=user)
+            with realm_context.activate(ctx):
+                response.render()
+
             assert response.status_code == 200
             content = response.content.decode('utf-8')
 

nodes/models.py

@@ -151,13 +151,16 @@ def is_framework_viewer(self, user: User, obj: InstanceConfig) -> bool:
             return False
         return user.has_instance_role(self.fw_viewer_role, obj.framework_config.framework)
 
-    def construct_perm_q(self, user: User, action: ObjectSpecificAction) -> models.Q | None:
+    def construct_perm_q(self, user: User, action: ObjectSpecificAction, include_implicit_public: bool = True) -> models.Q | None:
         is_admin = self.admin_role.role_q(user)
         is_viewer = self.viewer_role.role_q(user)
         is_fw_admin = self.fw_admin_role.role_q(user, prefix='framework_config__framework')
         is_fw_viewer = self.fw_viewer_role.role_q(user, prefix='framework_config__framework')
         if action == 'view':
-            return is_viewer | is_admin | is_fw_admin | is_fw_viewer | Q(framework_config__isnull=True)
+            q = is_viewer | is_admin | is_fw_admin | is_fw_viewer
+            if include_implicit_public:
+                q |= Q(framework_config__isnull=True)
+            return q
         return is_admin | is_fw_admin
 
     def construct_perm_q_anon(self, action: BaseObjectAction) -> Q | None:
@@ -167,7 +170,16 @@ def construct_perm_q_anon(self, action: BaseObjectAction) -> Q | None:
         return None
 
     def adminable_instances(self, user: User) -> InstanceConfigQuerySet:
-        return self.instances_user_has_any_permission_for(user, ['change'])
+        """
+        Return instances that the user has been explicitly granted access to.
+
+        We can't simply use the 'view' permission on InstanceConfig, because
+        most instances will be public by default.
+        """
+        qs = self.get_queryset()
+        if user.is_superuser:
+            return qs
+        return qs.filter(self.construct_perm_q(user, 'view', include_implicit_public=False))
 
     def user_has_perm(self, user: User, action: ObjectSpecificAction, obj: InstanceConfig) -> bool:
         if action == 'delete':

nodes/roles.py

@@ -60,6 +60,7 @@ class InstanceViewerRole(InstanceGroupMembershipRole, InstanceSpecificRole['Inst
     instance_group_field_name = 'viewer_group'
 
     model_perms = [
+        ('wagtailadmin', 'admin', ('access',)),
         ('nodes', ('instanceconfig', 'nodeconfig'), ('view',)),
         ('frameworks', (
             'framework', 'frameworkconfig', 'measure', 'measuredatapoint',

paths/dataset_permission_policy.py

@@ -26,7 +26,12 @@
 from nodes.roles import instance_admin_role, instance_viewer_role
 
 if TYPE_CHECKING:
+    from collections.abc import Sequence
+
+    from django.contrib.auth.models import AnonymousUser
+
     from kausal_common.models.permissions import PermissionedModel
+    from kausal_common.models.roles import InstanceSpecificRole
 
     from nodes.models import InstanceConfig
     from users.models import User
@@ -35,7 +40,7 @@
 _QS = TypeVar('_QS', bound=QuerySet, default=QuerySet[_M])
 
 
-class InstanceConfigScopedPermissionPolicy(ModelPermissionPolicy[_M, _QS]):
+class InstanceConfigScopedPermissionPolicy(ModelPermissionPolicy[_M, 'InstanceConfig', _QS]):
     """Permission policy for models that have one or many InstanceConfig objects as scope."""
 
     def __init__(self, model: type[_M]):
@@ -83,6 +88,17 @@ def user_can_create(self, user: User, context: InstanceConfig) -> bool:
     def construct_perm_q_anon(self, action: BaseObjectAction) -> Q | None:
         return None
 
+    def user_has_any_role_in_active_instance(self, user: User, roles: Sequence[InstanceSpecificRole[InstanceConfig]]) -> bool:
+        if user.is_superuser:
+            return True
+
+        active_instance = realm_context.get().realm
+
+        if active_instance is None:
+            return False
+
+        return any(user.has_instance_role(role, active_instance) for role in roles)
+
 
 class DatasetSchemaPermissionPolicy(InstanceConfigScopedPermissionPolicy[DatasetSchema]):
     """Permission policy for DatasetSchema, based on its scope (InstanceConfig)."""
@@ -119,6 +135,19 @@ def construct_perm_q(self, user: User, action: BaseObjectAction) -> Q | None:
 
         return admin_q
 
+    def user_has_permission(self, user: User | AnonymousUser, action: str) -> bool:
+        if not self.user_is_authenticated(user):
+            return False
+
+        allowed_roles: list[InstanceSpecificRole[InstanceConfig]] = [self.instance_admin_role]
+        if action == 'view':
+            allowed_roles.append(self.instance_viewer_role)
+
+        return self.user_has_any_role_in_active_instance(user, allowed_roles)
+
+    def user_has_any_permission(self, user: User | AnonymousUser, actions: Sequence[str]) -> bool:
+        return any(self.user_has_permission(user, action) for action in actions)
+
 
 class DatasetPermissionPolicy(ParentInheritedPolicy[Dataset, DatasetSchema, DatasetQuerySet]):
     """Permission policy for Dataset, inheriting from its schema."""

paths/middleware.py

@@ -67,8 +67,7 @@ async def get_admin_instance(self, request: PathsAdminRequest) -> None | Instanc
             instance_config = await InstanceConfig.objects.filter(id=admin_instance_id).afirst()
 
         adminable_instances = [
-            ic async for ic in InstanceConfig.permission_policy().instances_user_has_any_permission_for(user, ['change'])
-            .filter(site__isnull=False)
+            ic async for ic in user.get_adminable_instances().filter(site__isnull=False)
         ]
         if instance_config is not None and instance_config not in adminable_instances:
             instance_config = None

users/models.py

@@ -72,7 +72,7 @@ def get_active_instance(self) -> InstanceConfig | None:
 
     def get_adminable_instances(self) -> InstanceConfigQuerySet:
         from nodes.models import InstanceConfig
-        return InstanceConfig.permission_policy().instances_user_has_permission_for(self, 'change')
+        return InstanceConfig.permission_policy().adminable_instances(self)
 
     @cached_property
     def cgroups(self) -> QS[Group]: