Skip to content

Commit dad8d31

Browse files
owaiskazi19owaiskazi19
authored
[Backport-2.x] Added user level access control based on backend roles (#838) (#847)
* Added user level access control based on backend roles (#838) * Implemented backend role filtering for Flow Framework Signed-off-by: owaiskazi19 <[email protected]> * Spotless Fixes Signed-off-by: owaiskazi19 <[email protected]> * Added secured integ tests Signed-off-by: owaiskazi19 <[email protected]> * Fixed threadcontext and an integ test Signed-off-by: owaiskazi19 <[email protected]> * Added javadocs and fixed checkstyle Signed-off-by: owaiskazi19 <[email protected]> * Added backend role filtering for reprovisioning API Signed-off-by: owaiskazi19 <[email protected]> * Fixed exceptions Signed-off-by: owaiskazi19 <[email protected]> * Updated CHANGELOG Signed-off-by: owaiskazi19 <[email protected]> * Fixed forbidden APIs in tests Signed-off-by: owaiskazi19 <[email protected]> * Added secured integ tests for reprovision workflow Signed-off-by: owaiskazi19 <[email protected]> * Fixed checkstyle violation Signed-off-by: owaiskazi19 <[email protected]> * Added more tests and resolved PR comments Signed-off-by: Owais <[email protected]> * Addressed additional PR Comments Signed-off-by: Owais <[email protected]> * Updated the javadoc Signed-off-by: Owais <[email protected]> --------- Signed-off-by: owaiskazi19 <[email protected]> Signed-off-by: Owais <[email protected]> (cherry picked from commit 60458a6) * Updated HttpHost import Signed-off-by: Owais <[email protected]> --------- Signed-off-by: Owais <[email protected]>
1 parent f032867 commit dad8d31

31 files changed

+2362
-457
lines changed

‎CHANGELOG.md

+1
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@ Inspired from [Keep a Changelog](https://keepachangelog.com/en/1.1.0/)
1515
## [Unreleased 2.x](https://github.com/opensearch-project/flow-framework/compare/2.14...2.x)
1616
### Features
1717
- Adds reprovision API to support updating search pipelines, ingest pipelines index settings ([#804](https://github.com/opensearch-project/flow-framework/pull/804))
18+
- Adds user level access control based on backend roles ([#838](https://github.com/opensearch-project/flow-framework/pull/838))
1819

1920
### Enhancements
2021
### Bug Fixes

‎src/main/java/org/opensearch/flowframework/FlowFrameworkPlugin.java

+20-2
Original file line numberDiff line numberDiff line change
@@ -56,6 +56,7 @@
5656
import org.opensearch.flowframework.transport.SearchWorkflowStateAction;
5757
import org.opensearch.flowframework.transport.SearchWorkflowStateTransportAction;
5858
import org.opensearch.flowframework.transport.SearchWorkflowTransportAction;
59+
import org.opensearch.flowframework.transport.handler.SearchHandler;
5960
import org.opensearch.flowframework.util.EncryptorUtils;
6061
import org.opensearch.flowframework.workflow.WorkflowProcessSorter;
6162
import org.opensearch.flowframework.workflow.WorkflowStepFactory;
@@ -84,6 +85,7 @@
8485
import static org.opensearch.flowframework.common.CommonValue.PROVISION_WORKFLOW_THREAD_POOL;
8586
import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_STATE_INDEX;
8687
import static org.opensearch.flowframework.common.CommonValue.WORKFLOW_THREAD_POOL;
88+
import static org.opensearch.flowframework.common.FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES;
8789
import static org.opensearch.flowframework.common.FlowFrameworkSettings.FLOW_FRAMEWORK_ENABLED;
8890
import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOWS;
8991
import static org.opensearch.flowframework.common.FlowFrameworkSettings.MAX_WORKFLOW_STEPS;
@@ -135,7 +137,16 @@ public Collection<Object> createComponents(
135137
);
136138
WorkflowProcessSorter workflowProcessSorter = new WorkflowProcessSorter(workflowStepFactory, threadPool, flowFrameworkSettings);
137139

138-
return List.of(workflowStepFactory, workflowProcessSorter, encryptorUtils, flowFrameworkIndicesHandler, flowFrameworkSettings);
140+
SearchHandler searchHandler = new SearchHandler(settings, clusterService, client, FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES);
141+
142+
return List.of(
143+
workflowStepFactory,
144+
workflowProcessSorter,
145+
encryptorUtils,
146+
flowFrameworkIndicesHandler,
147+
searchHandler,
148+
flowFrameworkSettings
149+
);
139150
}
140151

141152
@Override
@@ -179,7 +190,14 @@ public List<RestHandler> getRestHandlers(
179190

180191
@Override
181192
public List<Setting<?>> getSettings() {
182-
return List.of(FLOW_FRAMEWORK_ENABLED, MAX_WORKFLOWS, MAX_WORKFLOW_STEPS, WORKFLOW_REQUEST_TIMEOUT, TASK_REQUEST_RETRY_DURATION);
193+
return List.of(
194+
FLOW_FRAMEWORK_ENABLED,
195+
MAX_WORKFLOWS,
196+
MAX_WORKFLOW_STEPS,
197+
WORKFLOW_REQUEST_TIMEOUT,
198+
TASK_REQUEST_RETRY_DURATION,
199+
FILTER_BY_BACKEND_ROLES
200+
);
183201
}
184202

185203
@Override

‎src/main/java/org/opensearch/flowframework/common/FlowFrameworkSettings.java

+8
Original file line numberDiff line numberDiff line change
@@ -75,6 +75,14 @@ public class FlowFrameworkSettings {
7575
Setting.Property.Dynamic
7676
);
7777

78+
/** This setting sets the backend role filtering */
79+
public static final Setting<Boolean> FILTER_BY_BACKEND_ROLES = Setting.boolSetting(
80+
"plugins.flow_framework.filter_by_backend_roles",
81+
false,
82+
Setting.Property.NodeScope,
83+
Setting.Property.Dynamic
84+
);
85+
7886
/**
7987
* Instantiate this class.
8088
*

‎src/main/java/org/opensearch/flowframework/transport/CreateWorkflowTransportAction.java

+87-2
Original file line numberDiff line numberDiff line change
@@ -16,12 +16,15 @@
1616
import org.opensearch.action.support.ActionFilters;
1717
import org.opensearch.action.support.HandledTransportAction;
1818
import org.opensearch.client.Client;
19+
import org.opensearch.cluster.service.ClusterService;
1920
import org.opensearch.common.inject.Inject;
21+
import org.opensearch.common.settings.Settings;
2022
import org.opensearch.common.unit.TimeValue;
2123
import org.opensearch.common.util.concurrent.ThreadContext;
2224
import org.opensearch.commons.authuser.User;
2325
import org.opensearch.core.action.ActionListener;
2426
import org.opensearch.core.rest.RestStatus;
27+
import org.opensearch.core.xcontent.NamedXContentRegistry;
2528
import org.opensearch.flowframework.common.CommonValue;
2629
import org.opensearch.flowframework.common.FlowFrameworkSettings;
2730
import org.opensearch.flowframework.exception.FlowFrameworkException;
@@ -49,7 +52,10 @@
4952
import static org.opensearch.flowframework.common.CommonValue.GLOBAL_CONTEXT_INDEX;
5053
import static org.opensearch.flowframework.common.CommonValue.PROVISIONING_PROGRESS_FIELD;
5154
import static org.opensearch.flowframework.common.CommonValue.STATE_FIELD;
55+
import static org.opensearch.flowframework.common.FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES;
56+
import static org.opensearch.flowframework.util.ParseUtils.checkFilterByBackendRoles;
5257
import static org.opensearch.flowframework.util.ParseUtils.getUserContext;
58+
import static org.opensearch.flowframework.util.ParseUtils.getWorkflow;
5359

5460
/**
5561
* Transport Action to index or update a use case template within the Global Context
@@ -63,6 +69,9 @@ public class CreateWorkflowTransportAction extends HandledTransportAction<Workfl
6369
private final Client client;
6470
private final FlowFrameworkSettings flowFrameworkSettings;
6571
private final PluginsService pluginsService;
72+
private volatile Boolean filterByEnabled;
73+
private final ClusterService clusterService;
74+
private final NamedXContentRegistry xContentRegistry;
6675

6776
/**
6877
* Instantiates a new CreateWorkflowTransportAction
@@ -73,6 +82,9 @@ public class CreateWorkflowTransportAction extends HandledTransportAction<Workfl
7382
* @param flowFrameworkSettings Plugin settings
7483
* @param client The client used to make the request to OS
7584
* @param pluginsService The plugin service
85+
* @param clusterService the cluster service
86+
* @param xContentRegistry the named content registry
87+
* @param settings the plugin settings
7688
*/
7789
@Inject
7890
public CreateWorkflowTransportAction(
@@ -82,20 +94,93 @@ public CreateWorkflowTransportAction(
8294
FlowFrameworkIndicesHandler flowFrameworkIndicesHandler,
8395
FlowFrameworkSettings flowFrameworkSettings,
8496
Client client,
85-
PluginsService pluginsService
97+
PluginsService pluginsService,
98+
ClusterService clusterService,
99+
NamedXContentRegistry xContentRegistry,
100+
Settings settings
86101
) {
87102
super(CreateWorkflowAction.NAME, transportService, actionFilters, WorkflowRequest::new);
88103
this.workflowProcessSorter = workflowProcessSorter;
89104
this.flowFrameworkIndicesHandler = flowFrameworkIndicesHandler;
90105
this.flowFrameworkSettings = flowFrameworkSettings;
91106
this.client = client;
92107
this.pluginsService = pluginsService;
108+
filterByEnabled = FILTER_BY_BACKEND_ROLES.get(settings);
109+
this.clusterService = clusterService;
110+
clusterService.getClusterSettings().addSettingsUpdateConsumer(FILTER_BY_BACKEND_ROLES, it -> filterByEnabled = it);
111+
this.xContentRegistry = xContentRegistry;
93112
}
94113

95114
@Override
96115
protected void doExecute(Task task, WorkflowRequest request, ActionListener<WorkflowResponse> listener) {
97-
98116
User user = getUserContext(client);
117+
String workflowId = request.getWorkflowId();
118+
try {
119+
resolveUserAndExecute(user, workflowId, listener, () -> createExecute(request, user, listener));
120+
} catch (Exception e) {
121+
logger.error("Failed to create workflow", e);
122+
listener.onFailure(e);
123+
}
124+
}
125+
126+
/**
127+
* Resolve user and execute the workflow function
128+
* @param requestedUser the user making the request
129+
* @param workflowId the workflow id
130+
* @param listener the action listener
131+
* @param function the workflow function to execute
132+
*/
133+
private void resolveUserAndExecute(
134+
User requestedUser,
135+
String workflowId,
136+
ActionListener<WorkflowResponse> listener,
137+
Runnable function
138+
) {
139+
try {
140+
// Check if user has backend roles
141+
// When filter by is enabled, block users creating/updating workflows who do not have backend roles.
142+
if (filterByEnabled == Boolean.TRUE) {
143+
try {
144+
checkFilterByBackendRoles(requestedUser);
145+
} catch (FlowFrameworkException e) {
146+
logger.error(e.getMessage(), e);
147+
listener.onFailure(e);
148+
return;
149+
}
150+
}
151+
if (workflowId != null) {
152+
// requestedUser == null means security is disabled or user is superadmin. In this case we don't need to
153+
// check if request user have access to the workflow or not. But we still need to get current workflow for
154+
// this case, so we can keep current workflow's user data.
155+
boolean filterByBackendRole = requestedUser == null ? false : filterByEnabled;
156+
// Update workflow request, check if user has permissions to update the workflow
157+
// Get workflow and verify backend roles
158+
getWorkflow(requestedUser, workflowId, filterByBackendRole, listener, function, client, clusterService, xContentRegistry);
159+
} else {
160+
// Create Workflow. No need to get current workflow.
161+
function.run();
162+
}
163+
} catch (Exception e) {
164+
String errorMessage = "Failed to create or update workflow";
165+
if (e instanceof FlowFrameworkException) {
166+
listener.onFailure(e);
167+
} else {
168+
listener.onFailure(new FlowFrameworkException(errorMessage, ExceptionsHelper.status(e)));
169+
}
170+
}
171+
}
172+
173+
/**
174+
* Execute the create or update request
175+
* 1. Validate workflows if requested
176+
* 2. Create or update global context index
177+
* 3. Create or update state index
178+
* 4. Create or update provisioning progress index
179+
* @param request the workflow request
180+
* @param user the user making the request
181+
* @param listener the action listener
182+
*/
183+
private void createExecute(WorkflowRequest request, User user, ActionListener<WorkflowResponse> listener) {
99184
Instant creationTime = Instant.now();
100185
Template templateWithUser = new Template(
101186
request.getTemplate().name(),

‎src/main/java/org/opensearch/flowframework/transport/DeleteWorkflowTransportAction.java

+61-14
Original file line numberDiff line numberDiff line change
@@ -16,17 +16,24 @@
1616
import org.opensearch.action.support.ActionFilters;
1717
import org.opensearch.action.support.HandledTransportAction;
1818
import org.opensearch.client.Client;
19+
import org.opensearch.cluster.service.ClusterService;
1920
import org.opensearch.common.inject.Inject;
21+
import org.opensearch.common.settings.Settings;
2022
import org.opensearch.common.util.concurrent.ThreadContext;
23+
import org.opensearch.commons.authuser.User;
2124
import org.opensearch.core.action.ActionListener;
2225
import org.opensearch.core.rest.RestStatus;
26+
import org.opensearch.core.xcontent.NamedXContentRegistry;
2327
import org.opensearch.flowframework.exception.FlowFrameworkException;
2428
import org.opensearch.flowframework.indices.FlowFrameworkIndicesHandler;
2529
import org.opensearch.tasks.Task;
2630
import org.opensearch.transport.TransportService;
2731

2832
import static org.opensearch.flowframework.common.CommonValue.CLEAR_STATUS;
2933
import static org.opensearch.flowframework.common.CommonValue.GLOBAL_CONTEXT_INDEX;
34+
import static org.opensearch.flowframework.common.FlowFrameworkSettings.FILTER_BY_BACKEND_ROLES;
35+
import static org.opensearch.flowframework.util.ParseUtils.getUserContext;
36+
import static org.opensearch.flowframework.util.ParseUtils.resolveUserAndExecute;
3037

3138
/**
3239
* Transport action to retrieve a use case template within the Global Context
@@ -37,50 +44,90 @@ public class DeleteWorkflowTransportAction extends HandledTransportAction<Workfl
3744

3845
private final FlowFrameworkIndicesHandler flowFrameworkIndicesHandler;
3946
private final Client client;
47+
private volatile Boolean filterByEnabled;
48+
private final ClusterService clusterService;
49+
private final NamedXContentRegistry xContentRegistry;
4050

4151
/**
4252
* Instantiates a new DeleteWorkflowTransportAction instance
4353
* @param transportService the transport service
4454
* @param actionFilters action filters
4555
* @param flowFrameworkIndicesHandler The Flow Framework indices handler
4656
* @param client the OpenSearch Client
57+
* @param clusterService the cluster service
58+
* @param xContentRegistry contentRegister to parse get response
59+
* @param settings the plugin settings
4760
*/
4861
@Inject
4962
public DeleteWorkflowTransportAction(
5063
TransportService transportService,
5164
ActionFilters actionFilters,
5265
FlowFrameworkIndicesHandler flowFrameworkIndicesHandler,
53-
Client client
66+
Client client,
67+
ClusterService clusterService,
68+
NamedXContentRegistry xContentRegistry,
69+
Settings settings
5470
) {
5571
super(DeleteWorkflowAction.NAME, transportService, actionFilters, WorkflowRequest::new);
5672
this.flowFrameworkIndicesHandler = flowFrameworkIndicesHandler;
5773
this.client = client;
74+
filterByEnabled = FILTER_BY_BACKEND_ROLES.get(settings);
75+
this.xContentRegistry = xContentRegistry;
76+
this.clusterService = clusterService;
77+
clusterService.getClusterSettings().addSettingsUpdateConsumer(FILTER_BY_BACKEND_ROLES, it -> filterByEnabled = it);
5878
}
5979

6080
@Override
6181
protected void doExecute(Task task, WorkflowRequest request, ActionListener<DeleteResponse> listener) {
6282
if (flowFrameworkIndicesHandler.doesIndexExist(GLOBAL_CONTEXT_INDEX)) {
6383
String workflowId = request.getWorkflowId();
64-
DeleteRequest deleteRequest = new DeleteRequest(GLOBAL_CONTEXT_INDEX, workflowId);
84+
User user = getUserContext(client);
6585

6686
ThreadContext.StoredContext context = client.threadPool().getThreadContext().stashContext();
67-
logger.info("Deleting workflow doc: {}", workflowId);
68-
client.delete(deleteRequest, ActionListener.runBefore(listener, context::restore));
6987

70-
// Whether to force deletion of corresponding state
71-
final boolean clearStatus = Booleans.parseBoolean(request.getParams().get(CLEAR_STATUS), false);
72-
ActionListener<DeleteResponse> stateListener = ActionListener.wrap(response -> {
73-
logger.info("Deleted workflow state doc: {}", workflowId);
74-
}, exception -> { logger.info("Failed to delete workflow state doc: {}", workflowId, exception); });
75-
flowFrameworkIndicesHandler.canDeleteWorkflowStateDoc(workflowId, clearStatus, canDelete -> {
76-
if (Boolean.TRUE.equals(canDelete)) {
77-
flowFrameworkIndicesHandler.deleteFlowFrameworkSystemIndexDoc(workflowId, stateListener);
78-
}
79-
}, stateListener);
88+
resolveUserAndExecute(
89+
user,
90+
workflowId,
91+
filterByEnabled,
92+
listener,
93+
() -> executeDeleteRequest(request, listener, context),
94+
client,
95+
clusterService,
96+
xContentRegistry
97+
);
98+
8099
} else {
81100
String errorMessage = "There are no templates in the global context";
82101
logger.error(errorMessage);
83102
listener.onFailure(new FlowFrameworkException(errorMessage, RestStatus.NOT_FOUND));
84103
}
85104
}
105+
106+
/**
107+
* Executes the delete request
108+
* @param request the workflow request
109+
* @param listener the action listener
110+
* @param context the thread context
111+
*/
112+
private void executeDeleteRequest(
113+
WorkflowRequest request,
114+
ActionListener<DeleteResponse> listener,
115+
ThreadContext.StoredContext context
116+
) {
117+
String workflowId = request.getWorkflowId();
118+
DeleteRequest deleteRequest = new DeleteRequest(GLOBAL_CONTEXT_INDEX, workflowId);
119+
logger.info("Deleting workflow doc: {}", workflowId);
120+
client.delete(deleteRequest, ActionListener.runBefore(listener, context::restore));
121+
122+
// Whether to force deletion of corresponding state
123+
final boolean clearStatus = Booleans.parseBoolean(request.getParams().get(CLEAR_STATUS), false);
124+
ActionListener<DeleteResponse> stateListener = ActionListener.wrap(response -> {
125+
logger.info("Deleted workflow state doc: {}", workflowId);
126+
}, exception -> { logger.info("Failed to delete workflow state doc: {}", workflowId, exception); });
127+
flowFrameworkIndicesHandler.canDeleteWorkflowStateDoc(workflowId, clearStatus, canDelete -> {
128+
if (Boolean.TRUE.equals(canDelete)) {
129+
flowFrameworkIndicesHandler.deleteFlowFrameworkSystemIndexDoc(workflowId, stateListener);
130+
}
131+
}, stateListener);
132+
}
86133
}

0 commit comments

Comments
 (0)