Merge pull request !2925 from rhodecode-enterprise-ce fix/special_case_for_duplicated_permissions

fix: adds a special case - when there are duplicated permissions, allows removing them

rhodecode/apps/repository/tests/test_repo_permissions.py

@@ -18,6 +18,10 @@
 
 import pytest
 
+from rhodecode.model.db import UserRepoToPerm, Permission
+from rhodecode.model.meta import Session
+from rhodecode.model.repo import RepoModel
+from rhodecode.model.user import UserModel
 from rhodecode.tests.utils import permission_update_data_generator
 from rhodecode.tests.routes import route_path
 
@@ -50,3 +54,47 @@ class TestRepoPermissionsView(object):
         response = self.app.post(route_path("edit_repo_perms", repo_name=repo_name), form_data).follow()
 
         assert "Repository access permissions updated" in response
+
+    def test_special_case_duplicated_permissions_no_error(self, csrf_token, user_util):
+        repo = user_util.create_repo()
+        repo_name = repo.repo_name
+        user = user_util.create_user()
+        user_id = user.user_id
+        username = user.username
+
+        self._crete_duplicated_permissions(username, repo_name)
+
+        # grant new
+        form_data = permission_update_data_generator(
+            csrf_token, default="repository.write", grant=[(user_id, "repository.write", username, "user")]
+        )
+
+        response = self.app.post(route_path("edit_repo_perms", repo_name=repo_name), form_data).follow()
+
+        assert "Found duplicate permissions. Remove duplicated permissions before doing any other action." in response
+
+    def test_special_case_duplicated_permissions_revoke_duplicated_permissions(self, csrf_token, user_util):
+        repo = user_util.create_repo()
+        repo_name = repo.repo_name
+        user = user_util.create_user()
+        user_id = user.user_id
+        username = user.username
+
+        self._crete_duplicated_permissions(username, repo_name)
+
+        # revoke duplicate
+        form_data = permission_update_data_generator(csrf_token, default="repository.read", revoke=[(user_id, "user")])
+
+        response = self.app.post(route_path("edit_repo_perms", repo_name=repo_name), form_data).follow()
+
+        assert "Repository access permissions updated" in response
+
+    def _crete_duplicated_permissions(self, username, repo_name):
+        for perm in ["repository.write", "repository.read"]:
+            obj = UserRepoToPerm()
+            obj.repository = RepoModel().get_by_repo_name(repo_name)
+            obj.user = UserModel().get_by_username(username)
+            obj.permission = Permission().get_by_key(perm)
+            Session().add(obj)
+
+        Session().commit()

rhodecode/apps/repository/views/repo_permissions.py

@@ -19,6 +19,7 @@
 import logging
 
 from pyramid.httpexceptions import HTTPFound
+from sqlalchemy.exc import MultipleResultsFound
 
 from rhodecode.apps._base import RepoAppView
 from rhodecode.lib import helpers as h
@@ -68,9 +69,28 @@ class RepoSettingsPermissionsView(RepoAppView):
         # into the form for private repos
         data["repo_private"] = self.db_repo.private
         form = RepoPermsForm(self.request.translate)().to_python(data)
-        changes = RepoModel().update_permissions(
-            self.db_repo_name, form["perm_additions"], form["perm_updates"], form["perm_deletions"]
-        )
+        try:
+            additions_ = form["perm_additions"]
+            updates_ = form["perm_updates"]
+            deletions_ = form["perm_deletions"]
+            deletions_, updates_ = self._remove_duplicates(deletions_, updates_)
+            updates_, additions_ = self._remove_duplicates(updates_, additions_)
+            changes = RepoModel().update_permissions(self.db_repo_name, additions_, updates_, deletions_)
+        except MultipleResultsFound:
+            # some clients managed to create duplicated permissions, this is possible only:
+            # - client migrated from the older version where such a bug existed
+            # - manually edited DB entries
+            # this is a rare case, and ideally should not exist
+            error_message = "Found duplicate permissions. Remove duplicated permissions before doing any other action."
+            h.flash(_(error_message), category="error")
+            audit_logger.store_web(
+                "repo.edit.permissions.error",
+                action_data={"error": error_message},
+                user=self._rhodecode_user,
+                repo=self.db_repo,
+            )
+            Session().commit()
+            raise HTTPFound(h.route_path("edit_repo_perms", repo_name=self.db_repo_name))
 
         action_data = {
             "added": changes["added"],
@@ -92,6 +112,20 @@ class RepoSettingsPermissionsView(RepoAppView):
 
         raise HTTPFound(h.route_path("edit_repo_perms", repo_name=self.db_repo_name))
 
+    def _remove_duplicates(
+        self, prioritized: list[tuple[str, str, str]], lst: list[tuple[str, str, str]]
+    ) -> tuple[list, list]:
+        """
+        if there are duplicates in prioritized list, remove them from lst
+        """
+        filtered_list = []
+        prioritized_ids = set(perm[0] for perm in prioritized)
+        for id_, *data in lst:
+            if id_ not in prioritized_ids:
+                filtered_list.append((id_, *data))
+
+        return prioritized, filtered_list
+
     @LoginRequired()
     @HasRepoPermissionAnyDecorator("repository.admin")
     @CSRFRequired()

rhodecode/lib/audit_logger.py

@@ -69,6 +69,7 @@ ACTIONS_V1 = {
     "repo.fork": {"data": {}},
     "repo.edit": {"old_data": {}},
     "repo.edit.permissions": {},
+    "repo.edit.permissions.error": {},
     "repo.edit.permissions.branch": {},
     "repo.archive": {"old_data": {}},
     "repo.delete": {"old_data": {}},

rhodecode/model/repo.py

@@ -907,14 +907,16 @@ class RepoModel(BaseModel):
         user = self._get_user(user)
         repo = self._get_repo(repo)
 
-        obj = (
+        objs = (
             self.sa.query(UserRepoToPerm)
             .filter(UserRepoToPerm.repository == repo)
             .filter(UserRepoToPerm.user == user)
-            .scalar()
+            .all()
         )
-        if obj:
-            self.sa.delete(obj)
+        if objs:
+            for obj in objs:
+                self.sa.delete(obj)
+
             log.debug("Revoked perm on %s on %s", repo, user)
             action_logger_generic(
                 "revoked permission from user: {} on repo: {}".format(user, repo), namespace="security.repo"