update permission check to hopefully improve performance (#523)

Co-authored-by: burnettk <burnettk@users.noreply.github.com>

spiffworkflow-backend/src/spiffworkflow_backend/models/permission_assignment.py

@@ -41,6 +41,7 @@ class PermissionAssignmentModel(SpiffworkflowBaseDBModel):
     id = db.Column(db.Integer, primary_key=True)
     principal_id = db.Column(ForeignKey(PrincipalModel.id), nullable=False, index=True)
     permission_target_id = db.Column(ForeignKey(PermissionTargetModel.id), nullable=False, index=True)  # type: ignore
+    permission_target = db.relationship(PermissionTargetModel, backref="permission_assignments")
     grant_type = db.Column(db.String(50), nullable=False)
     permission = db.Column(db.String(50), nullable=False)
 

spiffworkflow-backend/src/spiffworkflow_backend/routes/process_api_blueprint.py

@@ -12,6 +12,8 @@ from flask.wrappers import Response
 
 from spiffworkflow_backend.exceptions.api_error import ApiError
 from spiffworkflow_backend.exceptions.process_entity_not_found_error import ProcessEntityNotFoundError
+from spiffworkflow_backend.models.permission_assignment import PermissionAssignmentModel
+from spiffworkflow_backend.models.permission_target import PermissionTargetModel
 from spiffworkflow_backend.models.principal import PrincipalModel
 from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
 from spiffworkflow_backend.models.process_instance_file_data import ProcessInstanceFileDataModel
@@ -24,6 +26,7 @@ from spiffworkflow_backend.services.git_service import GitService
 from spiffworkflow_backend.services.process_caller_service import ProcessCallerService
 from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
 from spiffworkflow_backend.services.process_model_service import ProcessModelService
+from spiffworkflow_backend.services.user_service import UserService
 
 process_api_blueprint = Blueprint("process_api", __name__)
 
@@ -40,6 +43,16 @@ def permissions_check(body: dict[str, dict[str, list[str]]]) -> flask.wrappers.R
     response_dict: dict[str, dict[str, bool]] = {}
     requests_to_check = body["requests_to_check"]
 
+    user = g.user
+    principals = UserService.all_principals_for_user(user)
+    principal_ids = [p.id for p in principals]
+
+    permission_assignments = (
+        PermissionAssignmentModel.query.filter(PermissionAssignmentModel.principal_id.in_(principal_ids))
+        .join(PermissionTargetModel)
+        .all()
+    )
+
     for target_uri, http_methods in requests_to_check.items():
         if target_uri not in response_dict:
             response_dict[target_uri] = {}
@@ -47,8 +60,8 @@ def permissions_check(body: dict[str, dict[str, list[str]]]) -> flask.wrappers.R
         for http_method in http_methods:
             permission_string = AuthorizationService.get_permission_from_http_method(http_method)
             if permission_string:
-                has_permission = AuthorizationService.user_has_permission(
-                    user=g.user,
+                has_permission = AuthorizationService.permission_assignments_include(
+                    permission_assignments=permission_assignments,
                     permission=permission_string,
                     target_uri=target_uri,
                 )

spiffworkflow-backend/src/spiffworkflow_backend/services/authorization_service.py

@@ -198,6 +198,33 @@ class AuthorizationService:
 
         return cls.has_permission(principals, permission, target_uri)
 
+    @classmethod
+    def permission_assignments_include(
+        cls, permission_assignments: list[PermissionAssignmentModel], permission: str, target_uri: str
+    ) -> bool:
+        uri_with_percent = re.sub(r"\*", "%", target_uri)
+        target_uri_normalized = uri_with_percent.removeprefix(V1_API_PATH_PREFIX)
+        for permission_assignment in permission_assignments:
+            print(f"permission_assignment.permission_target.uri: {permission_assignment.permission_target.uri}")
+            if permission_assignment.permission == permission and cls.target_uri_matches_actual_uri(
+                permission_assignment.permission_target.uri, target_uri_normalized
+            ):
+                # we might have to rethink this to actually support deny
+                if permission_assignment.grant_type == "permit":
+                    return True
+                elif permission_assignment.grant_type == "deny":
+                    return False
+                return True
+        return False
+
+    @classmethod
+    def target_uri_matches_actual_uri(cls, target_uri: str, actual_uri: str) -> bool:
+        if target_uri.endswith("%"):
+            return actual_uri.startswith(target_uri.removesuffix("%")) or actual_uri == target_uri.removesuffix(
+                "%"
+            ).removesuffix("/")
+        return actual_uri == target_uri
+
     @classmethod
     def delete_all_permissions(cls) -> None:
         """Delete_all_permissions_and_recreate.  EXCEPT For permissions for the current user?"""

spiffworkflow-backend/src/spiffworkflow_backend/services/user_service.py

@@ -7,6 +7,7 @@ from spiffworkflow_backend.models.db import db
 from spiffworkflow_backend.models.group import GroupModel
 from spiffworkflow_backend.models.human_task import HumanTaskModel
 from spiffworkflow_backend.models.human_task_user import HumanTaskUserModel
+from spiffworkflow_backend.models.principal import MissingPrincipalError
 from spiffworkflow_backend.models.principal import PrincipalModel
 from spiffworkflow_backend.models.user import UserModel
 from spiffworkflow_backend.models.user_group_assignment import UserGroupAssignmentModel
@@ -194,3 +195,16 @@ class UserService:
                         )
                     )
         return unique_permission_assignments
+
+    @classmethod
+    def all_principals_for_user(cls, user: UserModel) -> list[PrincipalModel]:
+        if user.principal is None:
+            raise MissingPrincipalError(f"Missing principal for user with id: {user.id}")
+        principals = [user.principal]
+
+        for group in user.groups:
+            if group.principal is None:
+                raise MissingPrincipalError(f"Missing principal for group with id: {group.id}")
+            principals.append(group.principal)
+
+        return principals

spiffworkflow-backend/tests/spiffworkflow_backend/helpers/base_test.py

@@ -14,6 +14,7 @@ from spiffworkflow_backend.models.db import db
 from spiffworkflow_backend.models.message_instance import MessageInstanceModel
 from spiffworkflow_backend.models.permission_assignment import Permission
 from spiffworkflow_backend.models.permission_target import PermissionTargetModel
+from spiffworkflow_backend.models.principal import PrincipalModel
 from spiffworkflow_backend.models.process_group import ProcessGroup
 from spiffworkflow_backend.models.process_group import ProcessGroupSchema
 from spiffworkflow_backend.models.process_instance import ProcessInstanceModel
@@ -322,6 +323,14 @@ class BaseTest:
         target_uri: str = PermissionTargetModel.URI_ALL,
         permission_names: list[str] | None = None,
     ) -> UserModel:
+        principal = user.principal
+        cls.add_permissions_to_principal(principal, target_uri=target_uri, permission_names=permission_names)
+        return user
+
+    @classmethod
+    def add_permissions_to_principal(
+        cls, principal: PrincipalModel, target_uri: str, permission_names: list[str] | None
+    ) -> None:
         permission_target = AuthorizationService.find_or_create_permission_target(target_uri)
 
         if permission_names is None:
@@ -329,11 +338,10 @@ class BaseTest:
 
         for permission in permission_names:
             AuthorizationService.create_permission_for_principal(
-                principal=user.principal,
+                principal=principal,
                 permission_target=permission_target,
                 permission=permission,
             )
-        return user
 
     def assert_user_has_permission(
         self,

spiffworkflow-backend/tests/spiffworkflow_backend/integration/test_process_api.py

@@ -26,6 +26,7 @@ from spiffworkflow_backend.models.reference_cache import ReferenceCacheModel
 from spiffworkflow_backend.models.task import TaskModel  # noqa: F401
 from spiffworkflow_backend.models.user import UserModel
 from spiffworkflow_backend.services.file_system_service import FileSystemService
+from spiffworkflow_backend.services.group_service import GroupService
 from spiffworkflow_backend.services.process_caller_service import ProcessCallerService
 from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
 from spiffworkflow_backend.services.process_instance_service import ProcessInstanceService
@@ -94,6 +95,39 @@ class TestProcessApi(BaseTest):
         assert response.json is not None
         assert response.json == expected_response_body
 
+    def test_permissions_check_with_wildcard_permissions_through_group(
+        self,
+        app: Flask,
+        client: FlaskClient,
+        with_db_and_bpmn_file_cleanup: None,
+    ) -> None:
+        user = self.find_or_create_user()
+        group = GroupService.find_or_create_group("test_group")
+        principal = group.principal
+        GroupService.add_user_to_group(user, group.identifier)
+        self.add_permissions_to_principal(principal, target_uri="/v1.0/process-groups/%", permission_names=["read"])
+        request_body = {
+            "requests_to_check": {
+                "/v1.0/process-groups": ["GET", "POST"],
+                "/v1.0/process-models": ["GET"],
+            }
+        }
+        expected_response_body = {
+            "results": {
+                "/v1.0/process-groups": {"GET": True, "POST": False},
+                "/v1.0/process-models": {"GET": False},
+            }
+        }
+        response = client.post(
+            "/v1.0/permissions-check",
+            headers=self.logged_in_headers(user),
+            content_type="application/json",
+            data=json.dumps(request_body),
+        )
+        assert response.status_code == 200
+        assert response.json is not None
+        assert response.json == expected_response_body
+
     def test_process_model_create(
         self,
         app: Flask,

spiffworkflow-backend/tests/spiffworkflow_backend/unit/test_authorization_service.py

@@ -442,6 +442,16 @@ class TestAuthorizationService(BaseTest):
         self.assert_user_has_permission(user_two, "read", "/v1.0/process-groups/hey2:yo")
         self.assert_user_has_permission(user_two, "create", "/v1.0/process-groups/hey2:yo")
 
+    def test_target_uri_matches_actual_uri(self, app: Flask, with_db_and_bpmn_file_cleanup: None) -> None:
+        # exact match
+        assert AuthorizationService.target_uri_matches_actual_uri("/process-groups/hey", "/process-groups/hey")
+        # wildcard
+        assert AuthorizationService.target_uri_matches_actual_uri("/process-groups/%", "/process-groups/hey")
+        # wildcard is magical
+        assert AuthorizationService.target_uri_matches_actual_uri("/process-groups/%", "/process-groups")
+        # no match, since prefix doesn't match. wildcard isn't that magical.
+        assert AuthorizationService.target_uri_matches_actual_uri("/process-groups/%", "/process-models") is False
+
     def _expected_basic_permissions(self) -> list[tuple[str, str]]:
         return sorted(
             [