Apply roles for Get Workspace Layer

src/layman/authz/__init__.py

@@ -5,6 +5,7 @@
 from layman import LaymanError, settings, authn, util as layman_util, common
 from layman.common.prime_db_schema import workspaces, users
 from layman.common.rest import parse_request_path
+from . import role_service
 
 
 def authorize(workspace, publication_type, publication_name, request_method, actor_name):
@@ -72,8 +73,11 @@ def authorize_after_multi_get_request(actor_name, response):
 
 
 def is_user_in_access_rule(username, access_rule_names):
-    return settings.RIGHTS_EVERYONE_ROLE in access_rule_names \
-        or (username and username in access_rule_names)
+    usernames, rolenames = split_user_and_role_names(access_rule_names)
+    userroles = role_service.get_user_roles(username)
+    return settings.RIGHTS_EVERYONE_ROLE in rolenames \
+        or (username and username in usernames) \
+        or (set(rolenames).intersection(userroles))
 
 
 def can_user_publish_in_public_workspace(username):

src/layman/authz/role_service.py

@@ -22,3 +22,9 @@ def ensure_admin_roles():
     select %s, 'ADMIN'
     ;"""
     db_util.run_statement(create_admin_user_roles_view, (settings.LAYMAN_GS_USER, settings.LAYMAN_GS_ROLE, settings.LAYMAN_GS_USER))
+
+
+def get_user_roles(username):
+    query = f"""select rolename from {ROLE_SERVICE_SCHEMA}.user_roles where username = %s"""
+    roles = db_util.run_query(query, (username, ))
+    return {role[0] for role in roles}

test_tools/process.py

@@ -75,6 +75,8 @@ def oauth2_provider_mock():
                 'test_adjust_db_for_roles_ws': None,
                 'test_adjust_db_for_roles_ws2': None,
                 'test_access_rights_role_user1': None,
+                'test_role_application_user': None,
+                'test_role_application_role_user': None,
             },
         },
         'host': '0.0.0.0',

test_tools/role_service.py

@@ -0,0 +1,23 @@
+from db import util as db_util
+from layman import settings
+
+
+def ensure_role(rolename):
+    insert_role_statement = f'''insert into {settings.LAYMAN_INTERNAL_ROLE_SERVICE_SCHEMA}.bussiness_roles(name) values (%s) ON CONFLICT (name) DO nothing;'''
+    db_util.run_statement(insert_role_statement, (rolename,))
+
+
+def delete_role(rolename):
+    delete_statement = f"""delete from {settings.LAYMAN_INTERNAL_ROLE_SERVICE_SCHEMA}.bussiness_roles where name = %s;"""
+    db_util.run_statement(delete_statement, (rolename,))
+
+
+def ensure_user_role(username, rolename):
+    ensure_role(rolename)
+    insert_user_role_statement = f'''insert into {settings.LAYMAN_INTERNAL_ROLE_SERVICE_SCHEMA}.bussiness_user_roles(username, rolename) values (%s, %s) ON CONFLICT (username, rolename) DO nothing;'''
+    db_util.run_statement(insert_user_role_statement, (username, rolename,))
+
+
+def delete_user_role(username, rolename):
+    delete_statement = f"""delete from {settings.LAYMAN_INTERNAL_ROLE_SERVICE_SCHEMA}.bussiness_user_roles where username = %s and rolename = %s;"""
+    db_util.run_statement(delete_statement, (username, rolename,))

tests/dynamic_data/publications/access_rights/test_role_application.py

@@ -0,0 +1,62 @@
+import pytest
+
+from layman import LaymanError
+from test_tools import process_client, role_service as role_service_util
+from tests import EnumTestTypes
+from tests.asserts.final.publication import util as assert_util
+from tests.dynamic_data import base_test, base_test_classes
+from tests.dynamic_data.publications import common_publications
+
+pytest_generate_tests = base_test.pytest_generate_tests
+
+
+class PublicationTypes(base_test_classes.PublicationByDefinitionBase):
+    LAYER = (common_publications.LAYER_VECTOR_SLD, 'layer')
+
+
+OWNER = 'test_role_application_user'
+ROLE = 'TEST_ROLE_APPLICATION_ROLE'
+ROLE_USER = 'test_role_application_role_user'
+USERS_AND_ROLES = {OWNER, ROLE}
+
+
+@pytest.mark.usefixtures('oauth2_provider_mock')
+class TestPublication(base_test.TestSingleRestPublication):
+    workspace = OWNER
+    publication_type = process_client.LAYER_TYPE
+
+    rest_parametrization = [
+    ]
+
+    usernames_to_reserve = [
+        OWNER,
+        ROLE_USER,
+    ]
+    role_user_headers = process_client.get_authz_headers(ROLE_USER)
+
+    test_cases = [base_test.TestCaseType(key='role_test',
+                                         rest_args={
+                                             'access_rights': {
+                                                 'read': ','.join(USERS_AND_ROLES),
+                                             },
+                                             'actor_name': OWNER,
+                                         },
+                                         type=EnumTestTypes.MANDATORY,
+                                         )]
+
+    def test_publication(self, layer, rest_method, rest_args):
+        rest_method.fn(layer, args=rest_args)
+        assert_util.is_publication_valid_and_complete(layer)
+
+        with pytest.raises(LaymanError) as exc_info:
+            process_client.get_workspace_publication(layer.type, layer.workspace, layer.name, headers=self.role_user_headers)
+        assert exc_info.value.http_code == 404
+        assert exc_info.value.code == 15
+        assert exc_info.value.message == 'Layer was not found'
+
+        role_service_util.ensure_user_role(ROLE_USER, ROLE)
+        info = process_client.get_workspace_publication(layer.type, layer.workspace, layer.name, headers=self.role_user_headers)
+        assert set(info['access_rights']['read']) == USERS_AND_ROLES
+
+        role_service_util.delete_user_role(ROLE_USER, ROLE)
+        role_service_util.delete_role(ROLE)