sartography
/
spiffworkflow-backend
mirror of https://github.com/sartography/spiffworkflow-backend.git
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

added fix to SpiffWorkflow to deepcopy operation params before evaluating them w/ burnettk

This commit is contained in:
jasquat 2023-01-05 17:29:28 -05:00
parent 158cbb4bfd
commit 4b8b5fc403
4 changed files with 44 additions and 27 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

21
src/spiffworkflow_backend/routes/process_instances_controller.py
View File

@ -1,7 +1,5 @@
"""APIs for dealing with process groups, process models, and process instances."""
import json
from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.models.process_model import ProcessModelInfo
from typing import Any
from typing import Dict
from typing import Optional
@ -33,6 +31,7 @@ from spiffworkflow_backend.models.process_instance_metadata import (
from spiffworkflow_backend.models.process_instance_report import (
ProcessInstanceReportModel,
)
from spiffworkflow_backend.models.process_model import ProcessModelInfo
from spiffworkflow_backend.models.spec_reference import SpecReferenceCache
from spiffworkflow_backend.models.spec_reference import SpecReferenceNotFoundError
from spiffworkflow_backend.models.spiff_logging import SpiffLoggingModel
@ -45,6 +44,7 @@ from spiffworkflow_backend.routes.process_api_blueprint import _get_process_mode
from spiffworkflow_backend.routes.process_api_blueprint import (
_un_modify_modified_process_model_id,
)
from spiffworkflow_backend.services.authorization_service import AuthorizationService
from spiffworkflow_backend.services.error_handling_service import ErrorHandlingService
from spiffworkflow_backend.services.git_service import GitCommandError
from spiffworkflow_backend.services.git_service import GitService
@ -578,25 +578,30 @@ def process_instance_reset(
def process_instance_find_by_id(
process_instance_id: int,
) -> flask.wrappers.Response:
"""Process_instance_find_by_id."""
process_instance = _find_process_instance_by_id_or_raise(process_instance_id)
modified_process_model_identifier = ProcessModelInfo.modify_process_identifier_for_path_param(process_instance.process_model_identifier)
process_instance_uri = f'/process-instances/{modified_process_model_identifier}/{process_instance.id}'
modified_process_model_identifier = (
ProcessModelInfo.modify_process_identifier_for_path_param(
process_instance.process_model_identifier
)
)
process_instance_uri = (
f"/process-instances/{modified_process_model_identifier}/{process_instance.id}"
)
has_permission = AuthorizationService.user_has_permission(
user=g.user,
permission='read',
permission="read",
target_uri=process_instance_uri,
)
uri_type = None
if not has_permission:
process_instance = _find_process_instance_for_me_or_raise(process_instance_id)
uri_type = 'for-me'
uri_type = "for-me"
response_json = {
"process_instance": process_instance,
"uri_type": uri_type,
}
return make_response(jsonify(response_json), 200)

33
src/spiffworkflow_backend/services/authorization_service.py
View File

@ -626,6 +626,7 @@ class AuthorizationService:
@classmethod
def set_basic_permissions(cls) -> list[PermissionToAssign]:
"""Set_basic_permissions."""
permissions_to_assign: list[PermissionToAssign] = []
permissions_to_assign.append(
PermissionToAssign(
@ -661,7 +662,10 @@ class AuthorizationService:
return permissions_to_assign
@classmethod
def set_process_group_permissions(cls, target: str, permission_set: str) -> list[PermissionToAssign]:
def set_process_group_permissions(
cls, target: str, permission_set: str
) -> list[PermissionToAssign]:
"""Set_process_group_permissions."""
permissions_to_assign: list[PermissionToAssign] = []
process_group_identifier = (
target.removeprefix("PG:").replace("/", ":").removeprefix(":")
@ -673,16 +677,16 @@ class AuthorizationService:
f"/process-groups/{process_related_path_segment}",
f"/process-models/{process_related_path_segment}",
]
permissions_to_assign = (
permissions_to_assign
+ cls.get_permissions_to_assign(
permission_set, process_related_path_segment, target_uris
)
permissions_to_assign = permissions_to_assign + cls.get_permissions_to_assign(
permission_set, process_related_path_segment, target_uris
)
return permissions_to_assign
@classmethod
def set_process_model_permissions(cls, target: str, permission_set: str) -> list[PermissionToAssign]:
def set_process_model_permissions(
cls, target: str, permission_set: str
) -> list[PermissionToAssign]:
"""Set_process_model_permissions."""
permissions_to_assign: list[PermissionToAssign] = []
process_model_identifier = (
target.removeprefix("PM:").replace("/", ":").removeprefix(":")
@ -693,11 +697,8 @@ class AuthorizationService:
process_related_path_segment = "*"
target_uris = [f"/process-models/{process_related_path_segment}"]
permissions_to_assign = (
permissions_to_assign
+ cls.get_permissions_to_assign(
permission_set, process_related_path_segment, target_uris
)
permissions_to_assign = permissions_to_assign + cls.get_permissions_to_assign(
permission_set, process_related_path_segment, target_uris
)
return permissions_to_assign
@ -731,9 +732,13 @@ class AuthorizationService:
permissions = ["create", "read", "update", "delete"]
if target.startswith("PG:"):
permissions_to_assign += cls.set_process_group_permissions(target, permission_set)
permissions_to_assign += cls.set_process_group_permissions(
target, permission_set
)
elif target.startswith("PM:"):
permissions_to_assign += cls.set_process_model_permissions(target, permission_set)
permissions_to_assign += cls.set_process_model_permissions(
target, permission_set
)
elif permission_set == "start":
raise InvalidPermissionError(
"Permission 'start' is only available for macros PM and PG."

1
tests/spiffworkflow_backend/helpers/base_test.py
View File

@ -354,6 +354,7 @@ class BaseTest:
assert has_permission is expected_result
def modify_process_identifier_for_path_param(self, identifier: str) -> str:
"""Modify_process_identifier_for_path_param."""
return ProcessModelInfo.modify_process_identifier_for_path_param(identifier)
def un_modify_modified_process_identifier_for_path_param(

16
tests/spiffworkflow_backend/integration/test_process_instances_controller.py
View File

@ -1,13 +1,15 @@
"""Test_users_controller."""
from flask.app import Flask
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask.testing import FlaskClient
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from spiffworkflow_backend.models.user import UserModel
class TestProcessInstancesController(BaseTest):
"""TestProcessInstancesController."""
def test_find_by_id(
self,
app: Flask,
@ -16,8 +18,12 @@ class TestProcessInstancesController(BaseTest):
with_super_admin_user: UserModel,
) -> None:
"""Test_user_search_returns_a_user."""
user_one = self.create_user_with_permission(username="user_one", target_uri="/process-instances/find-by-id/*")
user_two = self.create_user_with_permission(username="user_two", target_uri="/process-instances/find-by-id/*")
user_one = self.create_user_with_permission(
username="user_one", target_uri="/process-instances/find-by-id/*"
)
user_two = self.create_user_with_permission(
username="user_two", target_uri="/process-instances/find-by-id/*"
)
process_model = load_test_spec(
process_model_id="group/sample",
@ -34,7 +40,7 @@ class TestProcessInstancesController(BaseTest):
)
assert response.status_code == 200
assert response.json
assert response.json['id'] == process_instance.id
assert response.json["id"] == process_instance.id
response = client.get(
f"/v1.0/process-instances/find-by-id/{process_instance.id}",
@ -48,4 +54,4 @@ class TestProcessInstancesController(BaseTest):
)
assert response.status_code == 200
assert response.json
assert response.json['id'] == process_instance.id
assert response.json["id"] == process_instance.id