Access rights are tested against axioms

src/layman/common/prime_db_schema/publications.py

@@ -64,6 +64,79 @@ def get_publication_infos(workspace_name=None, pub_type=None):
     return infos
 
 
+def only_valid_names(users_list):
+    usernames_for_chesk = users_list.copy()
+    usernames_for_chesk.discard(ROLE_EVERYONE)
+    for username in usernames_for_chesk:
+        info = users.get_user_infos(username)
+        if not info:
+            raise LaymanError(43, {f'Not existing username. Username={username}'})
+
+
+def at_least_one_can_write(can_write):
+    if not can_write:
+        raise LaymanError(43, {f'At least one user have to have write rights.'})
+
+
+def who_can_write_can_read(can_read, can_write):
+    if ROLE_EVERYONE not in can_read and set(can_write).difference(can_read):
+        raise LaymanError(43, {f'All users who have write rights have to have also read rights. Who is missing={set(can_write).difference(can_read)}'})
+
+
+def i_can_still_write(actor_name, can_write):
+    if ROLE_EVERYONE not in can_write and actor_name not in can_write:
+        raise LaymanError(43, {f'After the operation, the actor has to have write right.'})
+
+
+def owner_can_still_write(owner,
+                          can_write,
+                          ):
+    if owner and ROLE_EVERYONE not in can_write and owner not in can_write:
+        raise LaymanError(43, {f'Owner of the personal workspace have to keep write right.'})
+
+
+def check_rights_axioms(can_read,
+                        can_write,
+                        actor_name,
+                        owner,
+                        can_read_old=None,
+                        can_write_old=None):
+    if can_read:
+        only_valid_names(can_read)
+    if can_write:
+        only_valid_names(can_write)
+        owner_can_still_write(owner, can_write)
+        at_least_one_can_write(can_write)
+        i_can_still_write(actor_name, can_write)
+    if can_read or can_write:
+        can_read_check = can_read or can_read_old
+        can_write_check = can_write or can_write_old
+        who_can_write_can_read(can_read_check, can_write_check)
+
+
+def check_publication_info(workspace_name, info):
+    owner_info = users.get_user_infos(workspace_name).get(workspace_name)
+    info["owner"] = owner_info and owner_info["username"]
+    try:
+        check_rights_axioms(info['access_rights'].get('read'),
+                            info['access_rights'].get('write'),
+                            info["actor_name"],
+                            info["owner"],
+                            info['access_rights'].get('read_old'),
+                            info['access_rights'].get('write_old'),
+                            )
+    except LaymanError as exc_info:
+        raise LaymanError(43, {'workspace_name': workspace_name,
+                               'publication_name': info.get("name"),
+                               'access_rights': {
+                                   'read': info['access_rights'].get('read'),
+                                   'write': info['access_rights'].get('write'), },
+                               'actor_name': info.get("actor_name"),
+                               'owner': info["owner"],
+                               'message': exc_info.data,
+                               })
+
+
 def clear_roles(users_list, workspace_name):
     result_set = set(users_list)
     result_set.discard(ROLE_EVERYONE)
@@ -75,6 +148,7 @@ def clear_roles(users_list, workspace_name):
 
 def insert_publication(workspace_name, info):
     id_workspace = workspaces.ensure_workspace(workspace_name)
+    check_publication_info(workspace_name, info)
 
     insert_publications_sql = f'''insert into {DB_SCHEMA}.publications as p
         (id_workspace, name, title, type, uuid, everyone_can_read, everyone_can_write) values
@@ -123,6 +197,7 @@ def update_publication(workspace_name, info):
         for right_type in right_type_list:
             access_rights_changes[right_type]['username_list_old'] = info_old["access_rights"][right_type]
             info["access_rights"][right_type + "_old"] = access_rights_changes[right_type]['username_list_old']
+        check_publication_info(workspace_name, info)
 
         for right_type in right_type_list:
             if info['access_rights'].get(right_type):

src/layman/common/prime_db_schema/publications_test.py

@@ -114,6 +114,138 @@ def test_select_publications():
         pubs = publications.get_publication_infos(username)
         assert len(pubs) == 0, pubs
 
+
+def test_only_valid_names():
+    workspace_name = 'test_only_valid_names_workspace'
+    username = 'test_only_valid_names_user'
+
+    with app.app_context():
+        workspaces.ensure_workspace(workspace_name)
+        id_workspace_user = workspaces.ensure_workspace(username)
+        users.ensure_user(id_workspace_user, userinfo)
+
+        publications.only_valid_names(set())
+        publications.only_valid_names({username, })
+        publications.only_valid_names({settings.RIGHTS_EVERYONE_ROLE, })
+        publications.only_valid_names({settings.RIGHTS_EVERYONE_ROLE, username, })
+        publications.only_valid_names({username, settings.RIGHTS_EVERYONE_ROLE, })
+
+        with pytest.raises(LaymanError) as exc_info:
+            publications.only_valid_names({username, workspace_name})
+            assert exc_info.value.code == 43
+
+        with pytest.raises(LaymanError) as exc_info:
+            publications.only_valid_names({workspace_name, username})
+            assert exc_info.value.code == 43
+
+        with pytest.raises(LaymanError) as exc_info:
+            publications.only_valid_names({workspace_name, settings.RIGHTS_EVERYONE_ROLE, })
+            assert exc_info.value.code == 43
+
+        with pytest.raises(LaymanError) as exc_info:
+            publications.only_valid_names({settings.RIGHTS_EVERYONE_ROLE, 'skaljgdalskfglshfgd', })
+            assert exc_info.value.code == 43
+
+
+def test_at_least_one_can_write():
+    workspace_name = 'test_at_least_one_can_write_workspace'
+    username = 'test_at_least_one_can_write_user'
+
+    publications.at_least_one_can_write({username, })
+    publications.at_least_one_can_write({settings.RIGHTS_EVERYONE_ROLE, })
+    publications.at_least_one_can_write({username, settings.RIGHTS_EVERYONE_ROLE, })
+    publications.at_least_one_can_write({workspace_name, })
+    publications.at_least_one_can_write({'lusfjdiaurghalskug', })
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.at_least_one_can_write(set())
+        assert exc_info.value.code == 43
+
+
+def test_who_can_write_can_read():
+    workspace_name = 'test_who_can_write_can_read_workspace'
+    username = 'test_who_can_write_can_read_user'
+
+    publications.who_can_write_can_read(set(), set())
+    publications.who_can_write_can_read({username, }, {username, })
+    publications.who_can_write_can_read({username, workspace_name}, {username, })
+    publications.who_can_write_can_read({username, settings.RIGHTS_EVERYONE_ROLE}, {username, })
+    publications.who_can_write_can_read({username, settings.RIGHTS_EVERYONE_ROLE}, {username, settings.RIGHTS_EVERYONE_ROLE, })
+    publications.who_can_write_can_read({settings.RIGHTS_EVERYONE_ROLE, }, {settings.RIGHTS_EVERYONE_ROLE, })
+    publications.who_can_write_can_read({settings.RIGHTS_EVERYONE_ROLE, }, {settings.RIGHTS_EVERYONE_ROLE, username, })
+    publications.who_can_write_can_read({settings.RIGHTS_EVERYONE_ROLE, }, {settings.RIGHTS_EVERYONE_ROLE, workspace_name, })
+    publications.who_can_write_can_read({settings.RIGHTS_EVERYONE_ROLE, username, }, {settings.RIGHTS_EVERYONE_ROLE, })
+    publications.who_can_write_can_read({settings.RIGHTS_EVERYONE_ROLE, username, }, set())
+    publications.who_can_write_can_read({workspace_name, }, {workspace_name, })
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.who_can_write_can_read(set(), {workspace_name, })
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.who_can_write_can_read(set(), {username, })
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.who_can_write_can_read(set(), {settings.RIGHTS_EVERYONE_ROLE, })
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.who_can_write_can_read(username, {settings.RIGHTS_EVERYONE_ROLE, })
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.who_can_write_can_read(username, {workspace_name, })
+        assert exc_info.value.code == 43
+
+
+def test_i_can_still_write():
+    workspace_name = 'test_i_can_still_write_workspace'
+    username = 'test_who_can_write_can_read_user'
+
+    publications.i_can_still_write(None, {settings.RIGHTS_EVERYONE_ROLE, })
+    publications.i_can_still_write(None, {username, settings.RIGHTS_EVERYONE_ROLE, })
+    publications.i_can_still_write(username, {settings.RIGHTS_EVERYONE_ROLE, })
+    publications.i_can_still_write(username, {workspace_name, settings.RIGHTS_EVERYONE_ROLE, })
+    publications.i_can_still_write(username, {workspace_name, username, })
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.i_can_still_write(None, set())
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.i_can_still_write(None, {workspace_name, })
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.i_can_still_write(username, set())
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.i_can_still_write(username, {workspace_name, })
+        assert exc_info.value.code == 43
+
+
+def test_owner_can_still_write():
+    workspace_name = 'test_owner_can_still_write_workspace'
+    username = 'test_owner_can_still_write_user'
+
+    publications.owner_can_still_write(None, set())
+    publications.owner_can_still_write(None, {settings.RIGHTS_EVERYONE_ROLE, })
+    publications.owner_can_still_write(None, {username, })
+    publications.owner_can_still_write(username, {settings.RIGHTS_EVERYONE_ROLE, })
+    publications.owner_can_still_write(username, {username, })
+    publications.owner_can_still_write(username, {username, workspace_name, })
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.owner_can_still_write(username, set())
+        assert exc_info.value.code == 43
+
+    with pytest.raises(LaymanError) as exc_info:
+        publications.owner_can_still_write(username, {workspace_name, })
+        assert exc_info.value.code == 43
+
+
 def test_clear_roles():
     workspace_name = 'test_clear_roles_workspace'
     username = 'test_clear_roles_user'
@@ -345,6 +477,71 @@ def case_test_update_rights(username,
                                 [username, settings.RIGHTS_EVERYONE_ROLE, ],
                                 )
 
+        with pytest.raises(LaymanError) as exc_info:
+            case_test_update_rights(username,
+                                    publication_insert_info,
+                                    {"access_rights": {"read": {username2, },
+                                                       "write": {username2, },
+                                                       },
+                                     'actor_name': username2},
+                                    [username, username2, ],
+                                    [username, username2, ],
+                                    )
+            assert exc_info.value.code == 43
+
+        with pytest.raises(LaymanError) as exc_info:
+            case_test_update_rights(username,
+                                    publication_insert_info,
+                                    {"access_rights": {"read": {username, },
+                                                       },
+                                     'actor_name': username},
+                                    [username, username2, ],
+                                    [username, username2, ],
+                                    )
+            assert exc_info.value.code == 43
+
+        with pytest.raises(LaymanError) as exc_info:
+            case_test_update_rights(username,
+                                    publication_insert_info,
+                                    {"access_rights": {"read": {username, },
+                                                       },
+                                     'actor_name': username},
+                                    [username, username2, ],
+                                    [username, username2, ],
+                                    )
+            assert exc_info.value.code == 43
+
+        case_test_update_rights(username,
+                                publication_insert_info,
+                                {"access_rights": {"read": {username, },
+                                                   "write": {username, },
+                                                   },
+                                 'actor_name': username},
+                                [username, ],
+                                [username, ],
+                                )
+        with pytest.raises(LaymanError) as exc_info:
+            case_test_update_rights(username,
+                                    publication_insert_info,
+                                    {"access_rights": {"write": {username, username2, },
+                                                       },
+                                     'actor_name': username},
+                                    [username, username2, ],
+                                    [username, username2, username2, ],
+                                    )
+            assert exc_info.value.code == 43
+
+        with pytest.raises(LaymanError) as exc_info:
+            case_test_update_rights(username,
+                                    publication_insert_info,
+                                    {"access_rights": {"write": {settings.RIGHTS_EVERYONE_ROLE, },
+                                                       },
+                                     'actor_name': username},
+                                    [username, username2, ],
+                                    [settings.RIGHTS_EVERYONE_ROLE, ],
+                                    )
+            assert exc_info.value.code == 43
+
         publications.delete_publication(username, publication_insert_info["publ_type_name"], publication_insert_info["name"])
 
 

src/layman/error_list.py

@@ -43,4 +43,5 @@
     40: (404, 'Username does not exist.'),
     41: (409, 'Username is in conflict with LAYMAN_GS_USER. To resolve this conflict, you can create new GeoServer user with another name to become new LAYMAN_GS_USER, give him LAYMAN_GS_ROLE and ADMIN roles, remove the old LAYMAN_GS_USER user at GeoServer, change environment settings LAYMAN_GS_USER and LAYMAN_GS_PASSWORD, and restart Layman'),
     42: (409, 'LAYMAN_PRIME_SCHEMA is in conflict with existing username.'),
+    43: (400, 'Wrong access rights.')
 }

src/layman/gs_wfs_proxy_test.py

@@ -1,11 +1,6 @@
 import requests
 from urllib.parse import urljoin
 
-import sys
-
-
-del sys.modules['layman']
-
 from layman import app
 from layman import settings
 from layman.layer import db