Implicitly grant read permissions for data/wfs_layer if any create/update/delete permissions is granted

Author: manisandro

src/config_generator/data_service_config.py

@@ -273,10 +273,14 @@ def dataset_permissions(self, role, session):
             for layer_name in datasets:
 
                 # lookup permissions (dataset restricted by default)
-                dataset_restricted_for_public = layer_name not in \
-                    public_permissions['data'].get(map_name, {})
-                dataset_permitted_for_role = layer_name in \
-                    role_permissions['data'].get(map_name, {})
+                # NOTE: dataset readable is implicit if any of 'data_create', data_update', 'data_delete' is permitted
+                dataset_restricted_for_public = True
+                dataset_permitted_for_role = False
+                for resource_type in ['data', 'data_create', 'data_update', 'data_delete']:
+                    dataset_restricted_for_public &= layer_name not in \
+                        public_permissions[resource_type].get(map_name, {})
+                    dataset_permitted_for_role |= layer_name in \
+                        role_permissions[resource_type].get(map_name, {})
                 if dataset_restricted_for_public and not dataset_permitted_for_role:
                     # dataset not permitted
                     continue

src/config_generator/ogc_service_config.py

@@ -772,10 +772,27 @@ def collect_wfs_layer_permissions(self, service_name, cap, is_public_role,
             layer_permitted_for_role = layer['name'] in \
                 role_permissions['wfs_layer'].get(service_name, {})
 
-            layer_or_service_restricted = layer_restricted_for_public or service_restricted_for_public
+            # lookup permissions
+            if self.permissions_default_allow:
+                # NOTE: restricted if wfs_layer is restricted
+                layer_restricted_for_public = layer['name'] in \
+                    public_restrictions['wfs_layer'].get(service_name, {})
+            else:
+                # NOTE: restricted by default, permitted also if any of 'wfs_layer_create', wfs_layer_update', 'wfs_layer_delete' are permitted
+                layer_restricted_for_public = True
+                for resource_type in ['wfs_layer', 'wfs_layer_create', 'wfs_layer_update', 'wfs_layer_delete']:
+                    layer_restricted_for_public &= layer['name'] not in \
+                        public_permissions[resource_type].get(service_name, {})
+
+            layer_permitted_for_role = False
+            for resource_type in ['wfs_layer', 'wfs_layer_create', 'wfs_layer_update', 'wfs_layer_delete']:
+                layer_permitted_for_role |= layer['name'] in \
+                    role_permissions[resource_type].get(service_name, {})
+
             if layer_restricted_for_public and not layer_permitted_for_role:
                 # wfs_layer not permitted at least for reading
                 continue
+            layer_or_service_restricted = layer_restricted_for_public or service_restricted_for_public
 
             layer_permissions = OrderedDict()
             layer_permissions['name'] = layer['name']

src/config_generator/theme_reader.py

@@ -173,8 +173,8 @@ def __get_edit_datasets(self, service_name):
                 # map not found
                 return []
 
-            # query writable data permissions
-            resource_types = ['data']
+            # query data permissions
+            resource_types = ['data', 'data_create', 'data_update', 'data_delete']
             datasets_query = session.query(Permission) \
                 .join(Permission.resource) \
                 .filter(Resource.parent_id == map_obj.id) \

tests/permissions_tests.py

@@ -314,6 +314,30 @@ def test_data_permissions(self):
         self.assertEqual(len(parse("$.roles[?(@.role=='admin')].permissions.data_datasets[?(@.name=='qwc_demo.edit_polygons' & @.deletable)]").find(perm)), 0)
         self.assertEqual(len(parse("$.roles[?(@.role=='admin')].permissions.data_datasets[?(@.name=='qwc_demo.edit_polygons')].attributes[?(@=='description')]").find(perm)), 1)
 
+
+        # Check that dataset is readable/creatable if only data_create is set
+        self.cursor.execute(f"""
+            DELETE FROM qwc_config.permissions;
+            DELETE FROM qwc_config.resources;
+            INSERT INTO qwc_config.resources (id, parent_id, type, name)
+            VALUES
+            (1, NULL, 'map', 'qwc_demo'),
+            (2, 1, 'data_create', 'edit_points');
+            INSERT INTO qwc_config.permissions (id, role_id, resource_id, priority, write)
+            VALUES
+            (1, {ROLE_PUBLIC}, 1, 0, FALSE),
+            (2, {ROLE_PUBLIC}, 2, 0, FALSE);
+        """)
+        PermissionsTests.conn.commit()
+
+        perm = self.__run_config_generator({})
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.data_datasets[?(@.name=='qwc_demo.edit_points' & @.writable==false)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.data_datasets[?(@.name=='qwc_demo.edit_points' & @.creatable==true)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.data_datasets[?(@.name=='qwc_demo.edit_points' & @.readable==true)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.data_datasets[?(@.name=='qwc_demo.edit_points' & @.updatable==false)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.data_datasets[?(@.name=='qwc_demo.edit_points' & @.deletable==false)]").find(perm)), 1)
+
+
     def test_wfs_permissions(self):
         """ Test WFS permissions. """
 
@@ -386,6 +410,29 @@ def test_wfs_permissions(self):
         self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Linien')].attributes[?(@=='beschreibung')]").find(perm)), 0)
         self.assertEqual(len(parse("$.roles[?(@.role=='admin')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Linien')].attributes[?(@=='beschreibung')]").find(perm)), 1)
 
+        # Check that wfs_layer is readable/creatable if only data_create is set
+        self.cursor.execute(f"""
+            DELETE FROM qwc_config.permissions;
+            DELETE FROM qwc_config.resources;
+            INSERT INTO qwc_config.resources (id, parent_id, type, name)
+            VALUES
+            (1, NULL, 'wfs_service', 'scan/wfs_test'),
+            (2, 1, 'wfs_layer_create', 'ÖV: Haltestellen');
+            INSERT INTO qwc_config.permissions (id, role_id, resource_id, priority, write)
+            VALUES
+            (1, {ROLE_PUBLIC}, 1, 0, FALSE),
+            (2, {ROLE_PUBLIC}, 2, 0, FALSE);
+        """)
+        PermissionsTests.conn.commit()
+
+        perm = self.__run_config_generator({})
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Haltestellen' & @.writable==false)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Haltestellen' & @.creatable==true)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Haltestellen' & @.readable==true)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Haltestellen' & @.updatable==false)]").find(perm)), 1)
+        self.assertEqual(len(parse("$.roles[?(@.role=='public')].permissions.wfs_services[?(@.name=='scan/wfs_test')].layers[?(@.name=='ÖV-_Haltestellen' & @.deletable==false)]").find(perm)), 1)
+
+
     def test_public_permissions_default_restrict_no_permissions(self):
         """ Test permissions_default_allow=false and no permissions. """