Moved functions out of iris_engine.access_control.utils into the business layer

Author: c8y3

source/app/blueprints/access_controls.py

@@ -48,7 +48,7 @@
 from app.datamgmt.manage.manage_access_control_db import user_has_client_access
 from app.datamgmt.manage.manage_users_db import get_user
 from app.iris_engine.access_control.iris_user import iris_current_user
-from app.iris_engine.access_control.utils import ac_fast_check_user_has_case_access
+from app.business.access_controls import ac_fast_check_user_has_case_access
 from app.iris_engine.access_control.utils import ac_get_effective_permissions_of_user
 from app.iris_engine.utils.tracker import track_activity
 from app.models.authorization import Permissions

source/app/blueprints/rest/case/case_routes.py

@@ -38,9 +38,8 @@
 from app.datamgmt.manage.manage_groups_db import get_group_with_members
 from app.datamgmt.manage.manage_users_db import get_user
 from app.datamgmt.manage.manage_users_db import get_users_list_restricted_from_case
-from app.datamgmt.manage.manage_users_db import set_user_case_access
+from app.business.access_controls import set_user_case_access, ac_fast_check_user_has_case_access
 from app.business.cases import cases_export_to_json
-from app.iris_engine.access_control.utils import ac_fast_check_user_has_case_access
 from app.iris_engine.access_control.utils import ac_set_case_access_for_users
 from app.iris_engine.utils.tracker import track_activity
 from app.models.models import CaseStatus

source/app/business/access_controls.py

@@ -15,9 +15,101 @@
 #  You should have received a copy of the GNU Lesser General Public License
 #  along with this program; if not, write to the Free Software Foundation,
 #  Inc., 51 Franklin Street, Fifth Floor, Boston, MA  02110-1301, USA.
+from sqlalchemy import and_
+
+from app import db
+from app.datamgmt.manage.manage_access_control_db import get_case_effective_access
+from app.datamgmt.manage.manage_access_control_db import check_ua_case_client
 from app.iris_engine.access_control.iris_user import iris_current_user
-from app.iris_engine.access_control.utils import ac_fast_check_user_has_case_access
+from app.logger import logger
+from app.models.authorization import UserCaseAccess
+from app.models.authorization import UserCaseEffectiveAccess
+from app.models.authorization import CaseAccessLevel
+from app.models.authorization import ac_flag_match_mask
 
 
 def ac_fast_check_current_user_has_case_access(cid, access_level):
     return ac_fast_check_user_has_case_access(iris_current_user.id, cid, access_level)
+
+
+def set_user_case_access(user_id, case_id, access_level):
+
+    uca = UserCaseAccess.query.filter(
+        UserCaseAccess.user_id == user_id,
+        UserCaseAccess.case_id == case_id
+    ).all()
+
+    if len(uca) > 1:
+        for u in uca:
+            db.session.delete(u)
+        db.session.commit()
+        uca = None
+
+    if not uca:
+        uca = UserCaseAccess()
+        uca.user_id = user_id
+        uca.case_id = case_id
+        uca.access_level = access_level
+        db.session.add(uca)
+    else:
+        uca[0].access_level = access_level
+
+    db.session.commit()
+
+    set_case_effective_access_for_user(user_id, case_id, access_level)
+
+
+def set_case_effective_access_for_user(user_id, case_id, access_level: int):
+    """
+    Set a case access from a user
+    """
+
+    uac = UserCaseEffectiveAccess.query.where(and_(
+        UserCaseEffectiveAccess.user_id == user_id,
+        UserCaseEffectiveAccess.case_id == case_id
+    )).all()
+
+    if len(uac) > 1:
+        logger.error(f'Multiple access found for user {user_id} and case {case_id}')
+        for u in uac:
+            db.session.delete(u)
+        db.session.commit()
+
+        uac = UserCaseEffectiveAccess()
+        uac.user_id = user_id
+        uac.case_id = case_id
+        uac.access_level = access_level
+        db.session.add(uac)
+
+    elif len(uac) == 1:
+        uac = uac[0]
+        uac.access_level = access_level
+
+    db.session.commit()
+
+
+def ac_fast_check_user_has_case_access(user_id, cid, expected_access_levels: list[CaseAccessLevel]):
+    """
+    Checks the user has access to the case with at least one of the access_level
+    if the user has access, returns the access level of the user to the case
+    Returns None otherwise
+    """
+    access_level = get_case_effective_access(user_id, cid)
+
+    if not access_level:
+        # The user has no direct access, check if he is part of the client
+        access_level = check_ua_case_client(user_id, cid)
+        if not access_level:
+            return None
+        set_case_effective_access_for_user(user_id, cid, access_level)
+
+        return access_level
+
+    if ac_flag_match_mask(access_level, CaseAccessLevel.deny_all.value):
+        return None
+
+    for acl in expected_access_levels:
+        if ac_flag_match_mask(access_level, acl.value):
+            return access_level
+
+    return None

source/app/datamgmt/manage/manage_users_db.py

@@ -34,7 +34,6 @@
 from app.iris_engine.access_control.utils import ac_auto_update_user_effective_access
 from app.iris_engine.access_control.utils import ac_get_detailed_effective_permissions_from_groups
 from app.iris_engine.access_control.utils import ac_remove_case_access_from_user
-from app.iris_engine.access_control.utils import ac_set_case_access_for_user
 from app.models.cases import Cases
 from app.models.models import Client
 from app.models.models import UserActivity
@@ -445,33 +444,6 @@ def remove_case_access_from_user(user_id, case_id):
     return True, 'Case access removed'
 
 
-def set_user_case_access(user_id, case_id, access_level):
-
-    uca = UserCaseAccess.query.filter(
-        UserCaseAccess.user_id == user_id,
-        UserCaseAccess.case_id == case_id
-    ).all()
-
-    if len(uca) > 1:
-        for u in uca:
-            db.session.delete(u)
-        db.session.commit()
-        uca = None
-
-    if not uca:
-        uca = UserCaseAccess()
-        uca.user_id = user_id
-        uca.case_id = case_id
-        uca.access_level = access_level
-        db.session.add(uca)
-    else:
-        uca[0].access_level = access_level
-
-    db.session.commit()
-
-    ac_set_case_access_for_user(user_id, case_id, access_level)
-
-
 def get_user_details(user_id, include_api_key=False):
 
     user = User.query.filter(User.id == user_id).first()

source/app/iris_engine/access_control/utils.py

@@ -2,10 +2,9 @@
 from sqlalchemy import and_
 
 from app import db
+from app.business.access_controls import set_case_effective_access_for_user
 from app.logger import logger
 from app.iris_engine.access_control.iris_user import iris_current_user
-from app.datamgmt.manage.manage_access_control_db import check_ua_case_client
-from app.datamgmt.manage.manage_access_control_db import get_case_effective_access
 from app.models.cases import Cases
 from app.models.models import Client
 from app.models.authorization import CaseAccessLevel
@@ -285,33 +284,6 @@ def ac_trace_effective_user_permissions(user_id):
     return perms
 
 
-def ac_fast_check_user_has_case_access(user_id, cid, expected_access_levels: list[CaseAccessLevel]):
-    """
-    Checks the user has access to the case with at least one of the access_level
-    if the user has access, returns the access level of the user to the case
-    Returns None otherwise
-    """
-    access_level = get_case_effective_access(user_id, cid)
-
-    if not access_level:
-        # The user has no direct access, check if he is part of the client
-        access_level = check_ua_case_client(user_id, cid)
-        if not access_level:
-            return None
-        ac_set_case_access_for_user(user_id, cid, access_level)
-
-        return access_level
-
-    if ac_flag_match_mask(access_level, CaseAccessLevel.deny_all.value):
-        return None
-
-    for acl in expected_access_levels:
-        if ac_flag_match_mask(access_level, acl.value):
-            return access_level
-
-    return None
-
-
 def ac_recompute_effective_ac_from_users_list(users_list):
     """
     Recompute all users effective access of users
@@ -558,44 +530,15 @@ def ac_set_case_access_for_users(users, case_id, access_level):
         user_id = user.get('id')
         if user_id == iris_current_user.id:
             logs = "It's done, but I excluded you from the list of users to update, Dave"
-            ac_set_case_access_for_user(user.get('id'), case_id, CaseAccessLevel.full_access.value)
+            set_case_effective_access_for_user(user.get('id'), case_id, CaseAccessLevel.full_access.value)
             continue
 
-        ac_set_case_access_for_user(user.get('id'), case_id, access_level)
+        set_case_effective_access_for_user(user.get('id'), case_id, access_level)
 
     db.session.commit()
     return True, logs
 
 
-def ac_set_case_access_for_user(user_id, case_id, access_level: int):
-    """
-    Set a case access from a user
-    """
-
-    uac = UserCaseEffectiveAccess.query.where(and_(
-        UserCaseEffectiveAccess.user_id == user_id,
-        UserCaseEffectiveAccess.case_id == case_id
-    )).all()
-
-    if len(uac) > 1:
-        logger.error(f'Multiple access found for user {user_id} and case {case_id}')
-        for u in uac:
-            db.session.delete(u)
-        db.session.commit()
-
-        uac = UserCaseEffectiveAccess()
-        uac.user_id = user_id
-        uac.case_id = case_id
-        uac.access_level = access_level
-        db.session.add(uac)
-
-    elif len(uac) == 1:
-        uac = uac[0]
-        uac.access_level = access_level
-
-    db.session.commit()
-
-
 def ac_get_fast_user_cases_access(user_id):
     ucea = UserCaseEffectiveAccess.query.with_entities(
         UserCaseEffectiveAccess.case_id