Support layer short names and WMSUseLayerIDs

docker/qgisserver/geomapfish_qgisserver/accesscontrol.py

@@ -21,7 +21,7 @@
 import zope.event.classhandler
 from c2c.template.config import config
 from enum import Enum
-from qgis.core import QgsMessageLog, QgsDataSourceUri, QgsProject
+from qgis.core import QgsMessageLog, QgsDataSourceUri, QgsProject, QgsLayerTreeLayer, QgsLayerTreeGroup
 from qgis.server import QgsAccessControlFilter
 from shapely import ops
 from sqlalchemy.orm import configure_mappers, scoped_session, sessionmaker
@@ -165,6 +165,13 @@ def handle(event: InvalidateCacheEvent):
             QgsMessageLog.logMessage(''.join(traceback.format_exception(*sys.exc_info())))
             raise
 
+    def ogc_layer_name(self, layer):
+        use_layer_id, _ = QgsProject.instance().readBoolEntry("WMSUseLayerIDs", "/", False)
+        if use_layer_id:
+            return layer.id()
+        else:
+            return layer.shortName() or layer.name()
+
     def get_layers(self):
         """
         Get the list of GMF WMS layers that can give access to each QGIS layer or group.
@@ -186,23 +193,31 @@ def get_layers(self):
             nodes = {}  # dict { node name : list of ancestor node names }
 
             def browse(path, node):
-                nodes[node.name()] = [node.name()]
+                if isinstance(node, QgsLayerTreeLayer):
+                    ogc_name = self.ogc_layer_name(node.layer())
+                elif isinstance(node, QgsLayerTreeGroup):
+                    ogc_name = node.customProperty("wmsShortName") or node.name()
+                else:
+                    ogc_name = node.name()
+
+                nodes[ogc_name] = [ogc_name]
 
                 for name in path:
-                    nodes[name].append(node.name())
+                    nodes[ogc_name].append(name)
 
                 for l in node.children():
-                    browse(path + [node.name()], l)
+                    browse(path + [ogc_name], l)
 
             browse([], self.project.layerTreeRoot())
 
             # Transform ancestor names in LayerWMS instances
             layers = {}  # dict( node name : list of LayerWMS }
             for layer in self.DBSession.query(LayerWMS) \
                     .filter(LayerWMS.ogc_server_id == self.ogcserver.id).all():
-                for node_name in layer.layer.split(','):
-                    for name in nodes.get(node_name, []):
-                        layers.setdefault(name, []).append(layer)
+                for ogc_layer_name, ancestors in nodes.items():
+                    for ancestor in ancestors:
+                        if ancestor in layer.layer.split(','):
+                            layers.setdefault(ogc_layer_name, []).append(layer)
             QgsMessageLog.logMessage('[accesscontrol] layers:\n{}'.format(
                 json.dumps(
                     dict([(k, [l.name for l in v]) for k, v in layers.items()]),
@@ -296,12 +311,16 @@ def get_area(self, layer, rw=False):
         - Access area as WKT or None
         """
         roles = self.get_roles()
-        key = (layer.name(), tuple(sorted(role.id for role in roles)), rw)
+        if roles == 'ROOT':
+            return Access.FULL, None
+
+        ogc_name = self.ogc_layer_name(layer)
+        key = (ogc_name, tuple(sorted(role.id for role in roles)), rw)
 
         if key in self.area_cache:
             return self.area_cache[key]
 
-        gmf_layers = self.get_layers().get(layer.name(), None)
+        gmf_layers = self.get_layers().get(ogc_name, None)
         if gmf_layers is None:
             raise Exception("Access to an unknown layer")
 
@@ -392,10 +411,11 @@ def layerPermissions(self, layer):  # NOQA
             rights = QgsAccessControlFilter.LayerPermissions()
             rights.canRead = rights.canInsert = rights.canUpdate = rights.canDelete = False
 
-            if layer.name() not in self.get_layers():
+            layers = self.get_layers()
+            ogc_layer_name = self.ogc_layer_name(layer)
+            if ogc_layer_name not in layers:
                 return rights
-
-            gmf_layers = self.get_layers()[layer.name()]
+            gmf_layers = self.get_layers()[ogc_layer_name]
 
             roles = self.get_roles()
             access, _ = self.get_restriction_areas(gmf_layers, roles=roles)

docker/qgisserver/tests/functional/accesscontrol_test.py

@@ -114,9 +114,11 @@ def test_data(dbsession):
     roles = {role.name: role for role in (role1, role2)}
     dbsession.add_all(roles.values())
 
+    root = User('root')
+    root.id = 0
     user1 = User('user1', roles=[role1])
     user2 = User('user12', roles=[role1, role2])
-    users = {user.username: user for user in (user1, user2)}
+    users = {user.username: user for user in (root, user1, user2)}
     dbsession.add_all(users.values())
 
 
@@ -126,6 +128,8 @@ def add_node(parent_node, node_def):
 
         if node_def['type'] == 'layer':
             vlayer = QgsVectorLayer("Point", node_def['name'], 'memory')
+            if 'shortName' in node_def:
+                vlayer.setShortName(node_def['shortName'])
             project.addMapLayer(vlayer)
             node = project.layerTreeRoot().findLayer(vlayer)
             clone = node.clone()
@@ -134,18 +138,27 @@ def add_node(parent_node, node_def):
 
         if node_def['type'] == 'group':
             node = parent_node.addGroup(node_def['name'])
+            if 'shortName' in node_def:
+                node.setCustomProperty("wmsShortName", node_def['shortName'])
             for child_def in node_def['children']:
                 add_node(node, child_def)
 
     for node in [
         {'name': 'root', 'type': 'group', 'children': [
             {'name': 'public_group', 'type': 'group', 'children': [
-                {'name': 'public_layer', 'type': 'layer'}]},
+                {'name': 'public_layer', 'type': 'layer'},
+            ]},
             {'name': 'private_group1', 'type': 'group', 'children': [
-                {'name': 'private_layer1', 'type': 'layer'}]},
+                {'name': 'private_layer1', 'type': 'layer'},
+            ]},
             {'name': 'private_group2', 'type': 'group', 'children': [
-                {'name': 'private_layer2', 'type': 'layer'}]},
-        ]}
+                {'name': 'private_layer2', 'type': 'layer'},
+            ]},
+            # For group and layer short names
+            {'name': 'private_group3', 'type': 'group', 'shortName': 'pg3', 'children': [
+                {'name': 'private_layer3', 'type': 'layer', 'shortName': 'pl3'},
+            ]},
+        ]},
     ]:
         add_node(project.layerTreeRoot(), node)
 
@@ -161,14 +174,20 @@ def add_node(parent_node, node_def):
     private_layer2 = LayerWMS(name='private_layer2', layer='private_layer2', public=False)
     private_layer2.ogc_server = ogc_server1
 
+    private_group3 = LayerWMS(name='private_group3', layer='pg3', public=False)
+    private_group3.ogc_server = ogc_server1
+
+    private_layer3 = LayerWMS(name='private_layer3', layer='pl3', public=False)
+    private_layer3.ogc_server = ogc_server1
+
     layers = {layer.name: layer for layer in (
-        public_group, public_layer, private_layer1, private_layer2
+        public_group, public_layer, private_layer1, private_layer2, private_group3, private_layer3
     )}
     dbsession.add_all(layers.values())
 
 
     ra1 = RestrictionArea('restriction_area1',
-                          layers=[private_layer1],
+                          layers=[private_layer1, private_layer3],
                           roles=[role1],
                           area=from_shape(area1, srid=21781))
     ra2 = RestrictionArea('restriction_area2',
@@ -193,6 +212,19 @@ def add_node(parent_node, node_def):
     t.rollback()
 
 
+@pytest.fixture(scope='function')
+def wms_use_layer_ids():
+    """
+    Activate WMSUseLayerIDs
+    """
+    project = QgsProject.instance()
+    try:
+        project.writeEntry("WMSUseLayerIDs", "/", True)
+        yield
+    finally:
+        project.writeEntry("WMSUseLayerIDs", "/", False)
+
+
 @pytest.mark.usefixtures("server_iface",
                          "qgs_access_control_filter",
                          "test_data")
@@ -207,15 +239,38 @@ def test_init(self, server_iface, dbsession, test_data):
         )
         assert ogcserver_accesscontrol.ogcserver is test_data['ogc_servers']['qgisserver1']
 
-    def test_get_layers(self, server_iface, dbsession, test_data):
+    def test_ogc_layer_name(self, server_iface, dbsession):
         ogcserver_accesscontrol = OGCServerAccessControl(
             server_iface,
             'qgisserver1',
             21781,
             dbsession
         )
+        for layer_name, expected in (
+            ('private_layer1', 'private_layer1'),
+            ('private_layer3', 'pl3'),
+        ):
+            layer = QgsProject.instance().mapLayersByName(layer_name)[0]
+            assert expected == ogcserver_accesscontrol.ogc_layer_name(layer)
 
-        layers = ogcserver_accesscontrol.get_layers()
+    @pytest.mark.usefixtures("wms_use_layer_ids")
+    def test_ogc_layer_with_wms_use_layer_ids(self, server_iface, dbsession):
+        ogcserver_accesscontrol = OGCServerAccessControl(
+            server_iface,
+            'qgisserver1',
+            21781,
+            dbsession
+        )
+        layer = QgsProject.instance().mapLayersByName('private_layer1')[0]
+        assert layer.id() == ogcserver_accesscontrol.ogc_layer_name(layer)
+
+    def test_get_layers(self, server_iface, dbsession, test_data):
+        ogcserver_accesscontrol = OGCServerAccessControl(
+            server_iface,
+            'qgisserver1',
+            21781,
+            dbsession
+        )
 
         test_layers = test_data['layers']
         expected = {
@@ -224,7 +279,13 @@ def test_get_layers(self, server_iface, dbsession, test_data):
                              test_layers['public_layer']],
             'private_layer1': [test_layers['private_layer1']],
             'private_layer2': [test_layers['private_layer2']],
+            'pg3': [test_layers['private_group3']],
+            'pl3': [test_layers['private_group3'],
+                    test_layers['private_layer3']],
         }
+
+        layers = ogcserver_accesscontrol.get_layers()
+
         assert set(expected.keys()) == set(layers.keys())
         for key in expected.keys():
             assert set(expected[key]) == set(layers[key])
@@ -272,6 +333,7 @@ def test_get_restriction_areas(self, server_iface, dbsession, test_data):
 
         for layer_names, rw, role_names, expected in (
             (('private_layer1',), False, ('role1',), (Access.AREA, [area1])),
+            (('private_layer3',), False, ('role1',), (Access.AREA, [area1])),
         ):
             layers = [test_layers[layer_name] for layer_name in layer_names]
             roles = [test_roles[role_name] for role_name in role_names]
@@ -288,11 +350,20 @@ def test_get_area(self, server_iface, dbsession, test_data):
             dbsession
         )
 
-        for user_name, layer_name, expected in (
+        for user_name, layer_name, expected in [
+            ('root', layer_name, (Access.FULL, None))
+            for layer_name in (
+                'public_layer',
+                'private_layer1',
+                'private_layer2',
+                'private_layer3',
+            )
+        ] + [
             ('user1', 'public_layer', (Access.FULL, None)),
             ('user1', 'private_layer1', (Access.AREA, area1.wkt)),
             ('user1', 'private_layer2', (Access.NO, None)),
-        ):
+            ('user1', 'private_layer3', (Access.AREA, area1.wkt)),
+        ]:
             user = test_data['users'][user_name]
             set_request_parameters(server_iface, {
                 'USER_ID': str(user.id)