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

added permission to run privileged scripts w/ burnettk

This commit is contained in:
jasquat 2022-12-21 11:24:38 -05:00
parent 6b886663f1
commit 71c4a2fdc4
7 changed files with 127 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

6
src/spiffworkflow_backend/config/permissions/development.yml
View File

@ -16,7 +16,7 @@ groups:
alex,
dan,
mike,
jason,
jason@sartography.com,
jarrad,
elizabeth,
jon,
@ -29,7 +29,7 @@ groups:
alex,
dan,
mike,
jason,
jason@sartography.com,
amir,
jarrad,
elizabeth,
@ -46,7 +46,7 @@ groups:
fin,
fin1,
harmeet,
jason,
jason@sartography.com,
sasha,
manuchehr,
lead,

2
src/spiffworkflow_backend/routes/user.py
View File

@ -105,7 +105,7 @@ def verify_token(
) from e
if (
user_info is not None and "error" not in user_info
user_info is not None and "error" not in user_info and 'iss' in user_info
): # not sure what to test yet
user_model = (
UserModel.query.filter(UserModel.service == user_info["iss"])

4
src/spiffworkflow_backend/scripts/add_permission.py
View File

@ -14,6 +14,10 @@ from spiffworkflow_backend.services.group_service import GroupService
class AddPermission(Script):
"""AddUserToGroup."""
@staticmethod
def requires_privileged_permissions() -> bool:
return True
def get_description(self) -> str:
"""Get_description."""
return """Add a permission to a group. ex: add_permission("read", "test/*", "Editors") """

47
src/spiffworkflow_backend/scripts/script.py
View File

@ -1,5 +1,7 @@
"""Script."""
from __future__ import annotations
from spiffworkflow_backend.models.process_instance import ProcessInstanceModel, ProcessInstanceNotFoundError
from spiffworkflow_backend.services.authorization_service import AuthorizationService
import importlib
import os
@ -20,6 +22,10 @@ from spiffworkflow_backend.models.script_attributes_context import (
SCRIPT_SUB_CLASSES = None
class ScriptUnauthorizedForUserError(Exception):
pass
class Script:
"""Provides an abstract class that defines how scripts should work, this must be extended in all Script Tasks."""
@ -43,6 +49,10 @@ class Script:
+ "does not properly implement the run function.",
)
@staticmethod
def requires_privileged_permissions() -> bool:
return True
@staticmethod
def generate_augmented_list(
script_attributes_context: ScriptAttributesContext,
@ -71,18 +81,41 @@ class Script:
that we created.
"""
instance = subclass()
return lambda *ar, **kw: subclass.run(
instance,
script_attributes_context,
*ar,
**kw,
)
def run_subclass(*ar: Any, **kw: Any) -> Any:
if subclass.requires_privileged_permissions():
script_function_name = get_script_function_name(subclass)
uri = f"/v1.0/can-run-privileged-script/{script_function_name}"
process_instance = ProcessInstanceModel.query.filter_by(id=script_attributes_context.process_instance_id).first()
if process_instance is None:
raise ProcessInstanceNotFoundError(
f"Could not find a process instance with id '{script_attributes_context.process_instance_id}' "
f"when running script '{script_function_name}'"
)
user = process_instance.process_initiator
has_permission = AuthorizationService.user_has_permission(
user=user, permission="create", target_uri=uri
)
if not has_permission:
raise ScriptUnauthorizedForUserError(
f"User {user.username} does not have access to run privileged script '{script_function_name}'"
)
return subclass.run(
instance,
script_attributes_context,
*ar,
**kw,
)
return run_subclass
def get_script_function_name(subclass: type[Script]) -> str:
return subclass.__module__.split(".")[-1]
execlist = {}
subclasses = Script.get_all_subclasses()
for x in range(len(subclasses)):
subclass = subclasses[x]
execlist[subclass.__module__.split(".")[-1]] = make_closure(
execlist[get_script_function_name(subclass)] = make_closure(
subclass, script_attributes_context=script_attributes_context
)
return execlist

4
src/spiffworkflow_backend/services/process_model_service.py
View File

@ -224,10 +224,10 @@ class ProcessModelService(FileSystemService):
new_process_model_list = []
for process_model in process_models:
uri = f"/v1.0/process-instances/{process_model.id.replace('/', ':')}"
result = AuthorizationService.user_has_permission(
has_permission = AuthorizationService.user_has_permission(
user=user, permission="create", target_uri=uri
)
if result:
if has_permission:
new_process_model_list.append(process_model)
return new_process_model_list

39
tests/data/script_add_permission/add_permission.bpmn Normal file
View File

@ -0,0 +1,39 @@
<?xml version="1.0" encoding="UTF-8"?>
<bpmn:definitions xmlns:bpmn="http://www.omg.org/spec/BPMN/20100524/MODEL" xmlns:bpmndi="http://www.omg.org/spec/BPMN/20100524/DI" xmlns:dc="http://www.omg.org/spec/DD/20100524/DC" xmlns:di="http://www.omg.org/spec/DD/20100524/DI" id="Definitions_96f6665" targetNamespace="http://bpmn.io/schema/bpmn" exporter="Camunda Modeler" exporterVersion="3.0.0-dev">
<bpmn:process id="Process_02u675m" isExecutable="true">
<bpmn:startEvent id="StartEvent_1">
<bpmn:outgoing>Flow_01cweoc</bpmn:outgoing>
</bpmn:startEvent>
<bpmn:sequenceFlow id="Flow_01cweoc" sourceRef="StartEvent_1" targetRef="add_permission_script" />
<bpmn:endEvent id="Event_11584qn">
<bpmn:incoming>Flow_1xle2yo</bpmn:incoming>
</bpmn:endEvent>
<bpmn:sequenceFlow id="Flow_1xle2yo" sourceRef="add_permission_script" targetRef="Event_11584qn" />
<bpmn:scriptTask id="add_permission_script" name="Add Permission">
<bpmn:incoming>Flow_01cweoc</bpmn:incoming>
<bpmn:outgoing>Flow_1xle2yo</bpmn:outgoing>
<bpmn:script>add_permission('read', '/v1.0/test_permission_uri', "test_group")</bpmn:script>
</bpmn:scriptTask>
</bpmn:process>
<bpmndi:BPMNDiagram id="BPMNDiagram_1">
<bpmndi:BPMNPlane id="BPMNPlane_1" bpmnElement="Process_02u675m">
<bpmndi:BPMNShape id="_BPMNShape_StartEvent_2" bpmnElement="StartEvent_1">
<dc:Bounds x="179" y="159" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Event_11584qn_di" bpmnElement="Event_11584qn">
<dc:Bounds x="432" y="159" width="36" height="36" />
</bpmndi:BPMNShape>
<bpmndi:BPMNShape id="Activity_1ymj79t_di" bpmnElement="add_permission_script">
<dc:Bounds x="270" y="137" width="100" height="80" />
</bpmndi:BPMNShape>
<bpmndi:BPMNEdge id="Flow_01cweoc_di" bpmnElement="Flow_01cweoc">
<di:waypoint x="215" y="177" />
<di:waypoint x="270" y="177" />
</bpmndi:BPMNEdge>
<bpmndi:BPMNEdge id="Flow_1xle2yo_di" bpmnElement="Flow_1xle2yo">
<di:waypoint x="370" y="177" />
<di:waypoint x="432" y="177" />
</bpmndi:BPMNEdge>
</bpmndi:BPMNPlane>
</bpmndi:BPMNDiagram>
</bpmn:definitions>

38
tests/spiffworkflow_backend/scripts/test_add_permission.py
View File

@ -1,6 +1,11 @@
"""Test_get_localtime."""
from flask.app import Flask
from flask_bpmn.api.api_error import ApiError
import pytest
from spiffworkflow_backend.scripts.script import ScriptUnauthorizedForUserError
from tests.spiffworkflow_backend.helpers.test_data import load_test_spec
from flask.testing import FlaskClient
from spiffworkflow_backend.services.process_instance_processor import ProcessInstanceProcessor
from tests.spiffworkflow_backend.helpers.base_test import BaseTest
from spiffworkflow_backend.models.group import GroupModel
@ -58,3 +63,36 @@ class TestAddPermission(BaseTest):
assert group is not None
assert permission_target is not None
assert len(permission_assignments) == 1
def test_add_permission_script_through_bpmn(
self,
app: Flask,
client: FlaskClient,
with_db_and_bpmn_file_cleanup: None,
) -> None:
basic_user = self.find_or_create_user("basic_user")
privileged_user = self.find_or_create_user("privileged_user")
self.add_permissions_to_user(
privileged_user,
target_uri="/v1.0/can-run-privileged-script/*",
permission_names=["create"],
)
process_model = load_test_spec(
process_model_id="add_permission",
process_model_source_directory="script_add_permission",
)
process_instance = self.create_process_instance_from_process_model(
process_model=process_model, user=basic_user
)
processor = ProcessInstanceProcessor(process_instance)
with pytest.raises(ApiError) as exception:
processor.do_engine_steps(save=True)
assert "ScriptUnauthorizedForUserError" in str(exception)
process_instance = self.create_process_instance_from_process_model(
process_model=process_model, user=privileged_user
)
processor = ProcessInstanceProcessor(process_instance)
processor.do_engine_steps(save=True)
assert process_instance.status == "complete"