Skip to content

Commit 807dc6c

Browse files
albertzaharovitsalbertzaharovits
authored
[8.14] Fix task cancellation authz on fulfilling cluster (#109357) (#109422)
This fixes task cancellation actions (i.e. internal:admin/tasks/cancel_child and internal:admin/tasks/ban) not being authorized by the fulfilling cluster. This can result in orphaned tasks on the fulfilling cluster. Backport of #109357
1 parent 1af4d9c commit 807dc6c

File tree

6 files changed

+221
-2
lines changed

6 files changed

+221
-2
lines changed

‎docs/changelog/109357.yaml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,5 @@
1+
pr: 109357
2+
summary: Fix task cancellation authz on fulfilling cluster
3+
area: Authorization
4+
type: bug
5+
issues: []

‎server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java‎

Lines changed: 14 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -47,7 +47,9 @@
4747

4848
public class TaskCancellationService {
4949
public static final String BAN_PARENT_ACTION_NAME = "internal:admin/tasks/ban";
50+
public static final String REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME = "cluster:internal/admin/tasks/ban";
5051
public static final String CANCEL_CHILD_ACTION_NAME = "internal:admin/tasks/cancel_child";
52+
public static final String REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME = "cluster:internal/admin/tasks/cancel_child";
5153
public static final TransportVersion VERSION_SUPPORTING_CANCEL_CHILD_ACTION = TransportVersions.V_8_8_0;
5254
private static final Logger logger = LogManager.getLogger(TaskCancellationService.class);
5355
private final TransportService transportService;
@@ -64,12 +66,24 @@ public TaskCancellationService(TransportService transportService) {
6466
BanParentTaskRequest::new,
6567
new BanParentRequestHandler()
6668
);
69+
transportService.registerRequestHandler(
70+
REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
71+
EsExecutors.DIRECT_EXECUTOR_SERVICE,
72+
BanParentTaskRequest::new,
73+
new BanParentRequestHandler()
74+
);
6775
transportService.registerRequestHandler(
6876
CANCEL_CHILD_ACTION_NAME,
6977
EsExecutors.DIRECT_EXECUTOR_SERVICE,
7078
CancelChildRequest::new,
7179
new CancelChildRequestHandler()
7280
);
81+
transportService.registerRequestHandler(
82+
REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
83+
EsExecutors.DIRECT_EXECUTOR_SERVICE,
84+
CancelChildRequest::new,
85+
new CancelChildRequestHandler()
86+
);
7387
}
7488

7589
private String localNodeId() {

‎x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolver.java‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -24,6 +24,7 @@
2424
import org.elasticsearch.common.Strings;
2525
import org.elasticsearch.common.util.Maps;
2626
import org.elasticsearch.core.Nullable;
27+
import org.elasticsearch.tasks.TaskCancellationService;
2728
import org.elasticsearch.transport.RemoteClusterService;
2829
import org.elasticsearch.transport.TransportRequest;
2930
import org.elasticsearch.xpack.core.action.XPackInfoAction;
@@ -178,6 +179,8 @@ public class ClusterPrivilegeResolver {
178179
private static final Set<String> CROSS_CLUSTER_SEARCH_PATTERN = Set.of(
179180
RemoteClusterService.REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME,
180181
RemoteClusterNodesAction.TYPE.name(),
182+
TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
183+
TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
181184
XPackInfoAction.NAME,
182185
// esql enrich
183186
"cluster:monitor/xpack/enrich/esql/resolve_policy",
@@ -187,6 +190,8 @@ public class ClusterPrivilegeResolver {
187190
private static final Set<String> CROSS_CLUSTER_REPLICATION_PATTERN = Set.of(
188191
RemoteClusterService.REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME,
189192
RemoteClusterNodesAction.TYPE.name(),
193+
TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
194+
TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
190195
XPackInfoAction.NAME,
191196
ClusterStateAction.NAME
192197
);

‎x-pack/plugin/security/qa/multi-cluster/src/javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRestIT.java‎

Lines changed: 190 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -7,20 +7,25 @@
77

88
package org.elasticsearch.xpack.remotecluster;
99

10+
import org.apache.http.util.EntityUtils;
11+
import org.elasticsearch.Build;
1012
import org.elasticsearch.action.search.SearchResponse;
1113
import org.elasticsearch.client.Request;
1214
import org.elasticsearch.client.RequestOptions;
1315
import org.elasticsearch.client.Response;
1416
import org.elasticsearch.client.ResponseException;
1517
import org.elasticsearch.common.UUIDs;
1618
import org.elasticsearch.common.settings.Settings;
19+
import org.elasticsearch.common.xcontent.XContentHelper;
1720
import org.elasticsearch.core.Strings;
1821
import org.elasticsearch.search.SearchHit;
1922
import org.elasticsearch.search.SearchResponseUtils;
2023
import org.elasticsearch.test.cluster.ElasticsearchCluster;
24+
import org.elasticsearch.test.cluster.local.distribution.DistributionType;
2125
import org.elasticsearch.test.cluster.util.resource.Resource;
2226
import org.elasticsearch.test.junit.RunnableTestRuleAdapter;
2327
import org.elasticsearch.xcontent.ObjectPath;
28+
import org.elasticsearch.xcontent.json.JsonXContent;
2429
import org.junit.ClassRule;
2530
import org.junit.rules.RuleChain;
2631
import org.junit.rules.TestRule;
@@ -36,6 +41,7 @@
3641
import java.util.concurrent.atomic.AtomicBoolean;
3742
import java.util.concurrent.atomic.AtomicInteger;
3843
import java.util.concurrent.atomic.AtomicReference;
44+
import java.util.function.Consumer;
3945
import java.util.stream.Collectors;
4046

4147
import static org.hamcrest.Matchers.anEmptyMap;
@@ -58,6 +64,7 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe
5864

5965
static {
6066
fulfillingCluster = ElasticsearchCluster.local()
67+
.distribution(DistributionType.DEFAULT)
6168
.name("fulfilling-cluster")
6269
.nodes(3)
6370
.apply(commonClusterConfig)
@@ -73,6 +80,7 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe
7380
.build();
7481

7582
queryCluster = ElasticsearchCluster.local()
83+
.distribution(DistributionType.DEFAULT)
7684
.name("query-cluster")
7785
.apply(commonClusterConfig)
7886
.setting("xpack.security.remote_cluster_client.ssl.enabled", () -> String.valueOf(SSL_ENABLED_REF.get()))
@@ -137,6 +145,168 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe
137145
INVALID_SECRET_LENGTH.set(randomValueOtherThan(22, () -> randomIntBetween(0, 99)));
138146
})).around(fulfillingCluster).around(queryCluster);
139147

148+
public void testTaskCancellation() throws Exception {
149+
assumeTrue("[error_query] is only available in snapshot builds", Build.current().isSnapshot());
150+
configureRemoteCluster();
151+
152+
final String indexName = "index_fulfilling";
153+
final String roleName = "taskCancellationRoleName";
154+
final String userName = "taskCancellationUsername";
155+
try {
156+
// create some index on the fulfilling cluster, to be searched from the querying cluster
157+
{
158+
Request bulkRequest = new Request("POST", "/_bulk?refresh=true");
159+
bulkRequest.setJsonEntity(Strings.format("""
160+
{ "index": { "_index": "%s" } }
161+
{ "foo": "bar" }
162+
""", indexName));
163+
assertOK(performRequestAgainstFulfillingCluster(bulkRequest));
164+
}
165+
166+
// Create user and role with privileges for remote indices
167+
var putRoleRequest = new Request("PUT", "/_security/role/" + roleName);
168+
putRoleRequest.setJsonEntity(Strings.format("""
169+
{
170+
"remote_indices": [
171+
{
172+
"names": ["%s"],
173+
"privileges": ["read", "read_cross_cluster"],
174+
"clusters": ["my_remote_cluster"]
175+
}
176+
]
177+
}""", indexName));
178+
assertOK(adminClient().performRequest(putRoleRequest));
179+
var putUserRequest = new Request("PUT", "/_security/user/" + userName);
180+
putUserRequest.setJsonEntity(Strings.format("""
181+
{
182+
"password": "%s",
183+
"roles" : ["%s"]
184+
}""", PASS, roleName));
185+
assertOK(adminClient().performRequest(putUserRequest));
186+
var submitAsyncSearchRequest = new Request(
187+
"POST",
188+
Strings.format(
189+
"/%s:%s/_async_search?ccs_minimize_roundtrips=%s",
190+
randomFrom("my_remote_cluster", "*", "my_remote_*"),
191+
indexName,
192+
randomBoolean()
193+
)
194+
);
195+
196+
// submit a stalling remote async search
197+
submitAsyncSearchRequest.setJsonEntity("""
198+
{
199+
"query": {
200+
"error_query": {
201+
"indices": [
202+
{
203+
"name": "*:*",
204+
"error_type": "exception",
205+
"stall_time_seconds": 60
206+
}
207+
]
208+
}
209+
}
210+
}""");
211+
String asyncSearchOpaqueId = "async-search-opaque-id-" + randomUUID();
212+
submitAsyncSearchRequest.setOptions(
213+
RequestOptions.DEFAULT.toBuilder()
214+
.addHeader("Authorization", headerFromRandomAuthMethod(userName, PASS))
215+
.addHeader("X-Opaque-Id", asyncSearchOpaqueId)
216+
);
217+
Response submitAsyncSearchResponse = client().performRequest(submitAsyncSearchRequest);
218+
assertOK(submitAsyncSearchResponse);
219+
Map<String, Object> submitAsyncSearchResponseMap = XContentHelper.convertToMap(
220+
JsonXContent.jsonXContent,
221+
EntityUtils.toString(submitAsyncSearchResponse.getEntity()),
222+
false
223+
);
224+
assertThat(submitAsyncSearchResponseMap.get("is_running"), equalTo(true));
225+
String asyncSearchId = (String) submitAsyncSearchResponseMap.get("id");
226+
assertThat(asyncSearchId, notNullValue());
227+
// wait for the tasks to show up on the querying cluster
228+
assertBusy(() -> {
229+
try {
230+
Response queryingClusterTasks = adminClient().performRequest(new Request("GET", "/_tasks"));
231+
assertOK(queryingClusterTasks);
232+
Map<String, Object> responseMap = XContentHelper.convertToMap(
233+
JsonXContent.jsonXContent,
234+
EntityUtils.toString(queryingClusterTasks.getEntity()),
235+
false
236+
);
237+
AtomicBoolean someTasks = new AtomicBoolean(false);
238+
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> {
239+
// search tasks should not be cancelled at this point (but some transitory ones might be,
240+
// e.g. for action "indices:admin/seq_no/global_checkpoint_sync")
241+
if (task.get("action") instanceof String action && action.contains("indices:data/read/search")) {
242+
assertThat(task.get("cancelled"), equalTo(false));
243+
someTasks.set(true);
244+
}
245+
});
246+
assertTrue(someTasks.get());
247+
} catch (IOException e) {
248+
throw new RuntimeException(e);
249+
}
250+
});
251+
// wait for the tasks to show up on the fulfilling cluster
252+
assertBusy(() -> {
253+
try {
254+
Response fulfillingClusterTasks = performRequestAgainstFulfillingCluster(new Request("GET", "/_tasks"));
255+
assertOK(fulfillingClusterTasks);
256+
Map<String, Object> responseMap = XContentHelper.convertToMap(
257+
JsonXContent.jsonXContent,
258+
EntityUtils.toString(fulfillingClusterTasks.getEntity()),
259+
false
260+
);
261+
AtomicBoolean someTasks = new AtomicBoolean(false);
262+
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> {
263+
// search tasks should not be cancelled at this point (but some transitory ones might be,
264+
// e.g. for action "indices:admin/seq_no/global_checkpoint_sync")
265+
if (task.get("action") instanceof String action && action.contains("indices:data/read/search")) {
266+
assertThat(task.get("cancelled"), equalTo(false));
267+
someTasks.set(true);
268+
}
269+
});
270+
assertTrue(someTasks.get());
271+
} catch (IOException e) {
272+
throw new RuntimeException(e);
273+
}
274+
});
275+
// delete the stalling async search
276+
var deleteAsyncSearchRequest = new Request("DELETE", Strings.format("/_async_search/%s", asyncSearchId));
277+
deleteAsyncSearchRequest.setOptions(
278+
RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", headerFromRandomAuthMethod(userName, PASS))
279+
);
280+
assertOK(client().performRequest(deleteAsyncSearchRequest));
281+
// ensure any remaining tasks are all cancelled on the querying cluster
282+
{
283+
Response queryingClusterTasks = adminClient().performRequest(new Request("GET", "/_tasks"));
284+
assertOK(queryingClusterTasks);
285+
Map<String, Object> responseMap = XContentHelper.convertToMap(
286+
JsonXContent.jsonXContent,
287+
EntityUtils.toString(queryingClusterTasks.getEntity()),
288+
false
289+
);
290+
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> assertThat(task.get("cancelled"), equalTo(true)));
291+
}
292+
// ensure any remaining tasks are all cancelled on the fulfilling cluster
293+
{
294+
Response fulfillingClusterTasks = performRequestAgainstFulfillingCluster(new Request("GET", "/_tasks"));
295+
assertOK(fulfillingClusterTasks);
296+
Map<String, Object> responseMap = XContentHelper.convertToMap(
297+
JsonXContent.jsonXContent,
298+
EntityUtils.toString(fulfillingClusterTasks.getEntity()),
299+
false
300+
);
301+
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> assertThat(task.get("cancelled"), equalTo(true)));
302+
}
303+
} finally {
304+
assertOK(adminClient().performRequest(new Request("DELETE", "/_security/user/" + userName)));
305+
assertOK(adminClient().performRequest(new Request("DELETE", "/_security/role/" + roleName)));
306+
assertOK(performRequestAgainstFulfillingCluster(new Request("DELETE", indexName)));
307+
}
308+
}
309+
140310
public void testCrossClusterSearch() throws Exception {
141311
configureRemoteCluster();
142312
final String crossClusterAccessApiKeyId = (String) API_KEY_MAP_REF.get().get("id");
@@ -446,4 +616,24 @@ private Response performRequestWithLocalSearchUser(final Request request) throws
446616
);
447617
return client().performRequest(request);
448618
}
619+
620+
@SuppressWarnings("unchecked")
621+
private static void selectTasksWithOpaqueId(
622+
Map<String, Object> tasksResponse,
623+
String opaqueId,
624+
Consumer<Map<String, Object>> taskConsumer
625+
) {
626+
Map<String, Map<String, Object>> nodes = (Map<String, Map<String, Object>>) tasksResponse.get("nodes");
627+
for (Map<String, Object> node : nodes.values()) {
628+
Map<String, Map<String, Object>> tasks = (Map<String, Map<String, Object>>) node.get("tasks");
629+
for (Map<String, Object> task : tasks.values()) {
630+
if (task.get("headers") != null) {
631+
Map<String, Object> headers = (Map<String, Object>) task.get("headers");
632+
if (opaqueId.equals(headers.get("X-Opaque-Id"))) {
633+
taskConsumer.accept(task);
634+
}
635+
}
636+
}
637+
}
638+
}
449639
}

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java‎

Lines changed: 1 addition & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -530,7 +530,7 @@ private void authorizeAction(
530530
)
531531
);
532532
} else {
533-
logger.warn("denying access as action [{}] is not an index or cluster action", action);
533+
logger.warn("denying access for [{}] as action [{}] is not an index or cluster action", authentication, action);
534534
auditTrail.accessDenied(requestId, authentication, action, request, authzInfo);
535535
listener.onFailure(actionDenied(authentication, authzInfo, action, request));
536536
}

‎x-pack/plugin/security/src/main/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java‎

Lines changed: 6 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -23,6 +23,7 @@
2323
import org.elasticsearch.license.LicenseUtils;
2424
import org.elasticsearch.license.XPackLicenseState;
2525
import org.elasticsearch.tasks.Task;
26+
import org.elasticsearch.tasks.TaskCancellationService;
2627
import org.elasticsearch.threadpool.ThreadPool;
2728
import org.elasticsearch.transport.RemoteConnectionManager;
2829
import org.elasticsearch.transport.RemoteConnectionManager.RemoteClusterAliasWithCredentials;
@@ -81,7 +82,11 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
8182
"internal:data/read/esql/open_exchange",
8283
"cluster:internal:data/read/esql/open_exchange",
8384
"internal:data/read/esql/exchange",
84-
"cluster:internal:data/read/esql/exchange"
85+
"cluster:internal:data/read/esql/exchange",
86+
TaskCancellationService.BAN_PARENT_ACTION_NAME,
87+
TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
88+
TaskCancellationService.CANCEL_CHILD_ACTION_NAME,
89+
TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME
8590
);
8691

8792
private final AuthenticationService authcService;

0 commit comments

Comments
 (0)