Skip to content

Commit

Permalink
Fix task cancellation authz on fulfilling cluster (elastic#109357)
Browse files Browse the repository at this point in the history 
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.
  • Loading branch information
@albertzaharovits
albertzaharovits committed Jun 6, 2024
1 parent aba35e1 commit 6dcb6a2
Show file tree
Hide file tree
Showing 6 changed files with 221 additions and 2 deletions.
5 changes: 5 additions & 0 deletions docs/changelog/109357.yaml
View file
Original file line number Diff line number Diff line change
@@ -0,0 +1,5 @@
pr: 109357
summary: Fix task cancellation authz on fulfilling cluster
area: Authorization
type: bug
issues: []
14 changes: 14 additions & 0 deletions server/src/main/java/org/elasticsearch/tasks/TaskCancellationService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -47,7 +47,9 @@

public class TaskCancellationService {
public static final String BAN_PARENT_ACTION_NAME = "internal:admin/tasks/ban";
public static final String REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME = "cluster:internal/admin/tasks/ban";
public static final String CANCEL_CHILD_ACTION_NAME = "internal:admin/tasks/cancel_child";
public static final String REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME = "cluster:internal/admin/tasks/cancel_child";
public static final TransportVersion VERSION_SUPPORTING_CANCEL_CHILD_ACTION = TransportVersions.V_8_8_0;
private static final Logger logger = LogManager.getLogger(TaskCancellationService.class);
private final TransportService transportService;
Expand All @@ -64,12 +66,24 @@ public TaskCancellationService(TransportService transportService) {
BanParentTaskRequest::new,
new BanParentRequestHandler()
);
transportService.registerRequestHandler(
REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
EsExecutors.DIRECT_EXECUTOR_SERVICE,
BanParentTaskRequest::new,
new BanParentRequestHandler()
);
transportService.registerRequestHandler(
CANCEL_CHILD_ACTION_NAME,
EsExecutors.DIRECT_EXECUTOR_SERVICE,
CancelChildRequest::new,
new CancelChildRequestHandler()
);
transportService.registerRequestHandler(
REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
EsExecutors.DIRECT_EXECUTOR_SERVICE,
CancelChildRequest::new,
new CancelChildRequestHandler()
);
}

private String localNodeId() {
Expand Down
5 changes: 5 additions & 0 deletions .../java/org/elasticsearch/xpack/core/security/authz/privilege/ClusterPrivilegeResolver.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -24,6 +24,7 @@
import org.elasticsearch.common.Strings;
import org.elasticsearch.common.util.Maps;
import org.elasticsearch.core.Nullable;
import org.elasticsearch.tasks.TaskCancellationService;
import org.elasticsearch.transport.RemoteClusterService;
import org.elasticsearch.transport.TransportRequest;
import org.elasticsearch.xpack.core.action.XPackInfoAction;
Expand Down Expand Up @@ -178,6 +179,8 @@ public class ClusterPrivilegeResolver {
private static final Set<String> CROSS_CLUSTER_SEARCH_PATTERN = Set.of(
RemoteClusterService.REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME,
RemoteClusterNodesAction.TYPE.name(),
TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
XPackInfoAction.NAME,
// esql enrich
"cluster:monitor/xpack/enrich/esql/resolve_policy",
Expand All @@ -187,6 +190,8 @@ public class ClusterPrivilegeResolver {
private static final Set<String> CROSS_CLUSTER_REPLICATION_PATTERN = Set.of(
RemoteClusterService.REMOTE_CLUSTER_HANDSHAKE_ACTION_NAME,
RemoteClusterNodesAction.TYPE.name(),
TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME,
XPackInfoAction.NAME,
ClusterStateAction.NAME
);
Expand Down
190 changes: 190 additions & 0 deletions .../javaRestTest/java/org/elasticsearch/xpack/remotecluster/RemoteClusterSecurityRestIT.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -7,20 +7,24 @@

package org.elasticsearch.xpack.remotecluster;

import org.apache.http.util.EntityUtils;
import org.elasticsearch.action.search.SearchResponse;
import org.elasticsearch.client.Request;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.Response;
import org.elasticsearch.client.ResponseException;
import org.elasticsearch.common.UUIDs;
import org.elasticsearch.common.settings.Settings;
import org.elasticsearch.common.xcontent.XContentHelper;
import org.elasticsearch.core.Strings;
import org.elasticsearch.search.SearchHit;
import org.elasticsearch.search.SearchResponseUtils;
import org.elasticsearch.test.cluster.ElasticsearchCluster;
import org.elasticsearch.test.cluster.local.distribution.DistributionType;
import org.elasticsearch.test.cluster.util.resource.Resource;
import org.elasticsearch.test.junit.RunnableTestRuleAdapter;
import org.elasticsearch.xcontent.ObjectPath;
import org.elasticsearch.xcontent.json.JsonXContent;
import org.junit.ClassRule;
import org.junit.rules.RuleChain;
import org.junit.rules.TestRule;
Expand All @@ -36,6 +40,7 @@
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicReference;
import java.util.function.Consumer;
import java.util.stream.Collectors;

import static org.hamcrest.Matchers.anEmptyMap;
Expand All @@ -58,6 +63,7 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe

static {
fulfillingCluster = ElasticsearchCluster.local()
.distribution(DistributionType.DEFAULT)
.name("fulfilling-cluster")
.nodes(3)
.apply(commonClusterConfig)
Expand All @@ -73,6 +79,7 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe
.build();

queryCluster = ElasticsearchCluster.local()
.distribution(DistributionType.DEFAULT)
.name("query-cluster")
.apply(commonClusterConfig)
.setting("xpack.security.remote_cluster_client.ssl.enabled", () -> String.valueOf(SSL_ENABLED_REF.get()))
Expand Down Expand Up @@ -137,6 +144,169 @@ public class RemoteClusterSecurityRestIT extends AbstractRemoteClusterSecurityTe
INVALID_SECRET_LENGTH.set(randomValueOtherThan(22, () -> randomIntBetween(0, 99)));
})).around(fulfillingCluster).around(queryCluster);

public void testTaskCancellation() throws Exception {
assumeTrue("[error_query] is only available in snapshot builds", Build.current().isSnapshot());
configureRemoteCluster();

final String indexName = "index_fulfilling";
final String roleName = "taskCancellationRoleName";
final String userName = "taskCancellationUsername";
try {
// create some index on the fulfilling cluster, to be searched from the querying cluster
{
Request bulkRequest = new Request("POST", "/_bulk?refresh=true");
bulkRequest.setJsonEntity(Strings.format("""
{ "index": { "_index": "%s" } }
{ "foo": "bar" }
""", indexName));
assertOK(performRequestAgainstFulfillingCluster(bulkRequest));
}

// Create user and role with privileges for remote indices
var putRoleRequest = new Request("PUT", "/_security/role/" + roleName);
putRoleRequest.setJsonEntity(Strings.format("""
{
"description": "Role with privileges for remote index for the test of task cancellation.",
"remote_indices": [
{
"names": ["%s"],
"privileges": ["read", "read_cross_cluster"],
"clusters": ["my_remote_cluster"]
}
]
}""", indexName));
assertOK(adminClient().performRequest(putRoleRequest));
var putUserRequest = new Request("PUT", "/_security/user/" + userName);
putUserRequest.setJsonEntity(Strings.format("""
{
"password": "%s",
"roles" : ["%s"]
}""", PASS, roleName));
assertOK(adminClient().performRequest(putUserRequest));
var submitAsyncSearchRequest = new Request(
"POST",
Strings.format(
"/%s:%s/_async_search?ccs_minimize_roundtrips=%s",
randomFrom("my_remote_cluster", "*", "my_remote_*"),
indexName,
randomBoolean()
)
);

// submit a stalling remote async search
submitAsyncSearchRequest.setJsonEntity("""
{
"query": {
"error_query": {
"indices": [
{
"name": "*:*",
"error_type": "exception",
"stall_time_seconds": 60
}
]
}
}
}""");
String asyncSearchOpaqueId = "async-search-opaque-id-" + randomUUID();
submitAsyncSearchRequest.setOptions(
RequestOptions.DEFAULT.toBuilder()
.addHeader("Authorization", headerFromRandomAuthMethod(userName, PASS))
.addHeader("X-Opaque-Id", asyncSearchOpaqueId)
);
Response submitAsyncSearchResponse = client().performRequest(submitAsyncSearchRequest);
assertOK(submitAsyncSearchResponse);
Map<String, Object> submitAsyncSearchResponseMap = XContentHelper.convertToMap(
JsonXContent.jsonXContent,
EntityUtils.toString(submitAsyncSearchResponse.getEntity()),
false
);
assertThat(submitAsyncSearchResponseMap.get("is_running"), equalTo(true));
String asyncSearchId = (String) submitAsyncSearchResponseMap.get("id");
assertThat(asyncSearchId, notNullValue());
// wait for the tasks to show up on the querying cluster
assertBusy(() -> {
try {
Response queryingClusterTasks = adminClient().performRequest(new Request("GET", "/_tasks"));
assertOK(queryingClusterTasks);
Map<String, Object> responseMap = XContentHelper.convertToMap(
JsonXContent.jsonXContent,
EntityUtils.toString(queryingClusterTasks.getEntity()),
false
);
AtomicBoolean someTasks = new AtomicBoolean(false);
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> {
// search tasks should not be cancelled at this point (but some transitory ones might be,
// e.g. for action "indices:admin/seq_no/global_checkpoint_sync")
if (task.get("action") instanceof String action && action.contains("indices:data/read/search")) {
assertThat(task.get("cancelled"), equalTo(false));
someTasks.set(true);
}
});
assertTrue(someTasks.get());
} catch (IOException e) {
throw new RuntimeException(e);
}
});
// wait for the tasks to show up on the fulfilling cluster
assertBusy(() -> {
try {
Response fulfillingClusterTasks = performRequestAgainstFulfillingCluster(new Request("GET", "/_tasks"));
assertOK(fulfillingClusterTasks);
Map<String, Object> responseMap = XContentHelper.convertToMap(
JsonXContent.jsonXContent,
EntityUtils.toString(fulfillingClusterTasks.getEntity()),
false
);
AtomicBoolean someTasks = new AtomicBoolean(false);
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> {
// search tasks should not be cancelled at this point (but some transitory ones might be,
// e.g. for action "indices:admin/seq_no/global_checkpoint_sync")
if (task.get("action") instanceof String action && action.contains("indices:data/read/search")) {
assertThat(task.get("cancelled"), equalTo(false));
someTasks.set(true);
}
});
assertTrue(someTasks.get());
} catch (IOException e) {
throw new RuntimeException(e);
}
});
// delete the stalling async search
var deleteAsyncSearchRequest = new Request("DELETE", Strings.format("/_async_search/%s", asyncSearchId));
deleteAsyncSearchRequest.setOptions(
RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", headerFromRandomAuthMethod(userName, PASS))
);
assertOK(client().performRequest(deleteAsyncSearchRequest));
// ensure any remaining tasks are all cancelled on the querying cluster
{
Response queryingClusterTasks = adminClient().performRequest(new Request("GET", "/_tasks"));
assertOK(queryingClusterTasks);
Map<String, Object> responseMap = XContentHelper.convertToMap(
JsonXContent.jsonXContent,
EntityUtils.toString(queryingClusterTasks.getEntity()),
false
);
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> assertThat(task.get("cancelled"), equalTo(true)));
}
// ensure any remaining tasks are all cancelled on the fulfilling cluster
{
Response fulfillingClusterTasks = performRequestAgainstFulfillingCluster(new Request("GET", "/_tasks"));
assertOK(fulfillingClusterTasks);
Map<String, Object> responseMap = XContentHelper.convertToMap(
JsonXContent.jsonXContent,
EntityUtils.toString(fulfillingClusterTasks.getEntity()),
false
);
selectTasksWithOpaqueId(responseMap, asyncSearchOpaqueId, task -> assertThat(task.get("cancelled"), equalTo(true)));
}
} finally {
assertOK(adminClient().performRequest(new Request("DELETE", "/_security/user/" + userName)));
assertOK(adminClient().performRequest(new Request("DELETE", "/_security/role/" + roleName)));
assertOK(performRequestAgainstFulfillingCluster(new Request("DELETE", indexName)));
}
}

public void testCrossClusterSearch() throws Exception {
configureRemoteCluster();
final String crossClusterAccessApiKeyId = (String) API_KEY_MAP_REF.get().get("id");
Expand Down Expand Up @@ -446,4 +616,24 @@ private Response performRequestWithLocalSearchUser(final Request request) throws
);
return client().performRequest(request);
}

@SuppressWarnings("unchecked")
private static void selectTasksWithOpaqueId(
Map<String, Object> tasksResponse,
String opaqueId,
Consumer<Map<String, Object>> taskConsumer
) {
Map<String, Map<String, Object>> nodes = (Map<String, Map<String, Object>>) tasksResponse.get("nodes");
for (Map<String, Object> node : nodes.values()) {
Map<String, Map<String, Object>> tasks = (Map<String, Map<String, Object>>) node.get("tasks");
for (Map<String, Object> task : tasks.values()) {
if (task.get("headers") != null) {
Map<String, Object> headers = (Map<String, Object>) task.get("headers");
if (opaqueId.equals(headers.get("X-Opaque-Id"))) {
taskConsumer.accept(task);
}
}
}
}
}
}
2 changes: 1 addition & 1 deletion ...n/security/src/main/java/org/elasticsearch/xpack/security/authz/AuthorizationService.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -530,7 +530,7 @@ private void authorizeAction(
)
);
} else {
logger.warn("denying access as action [{}] is not an index or cluster action", action);
logger.warn("denying access for [{}] as action [{}] is not an index or cluster action", authentication, action);
auditTrail.accessDenied(requestId, authentication, action, request, authzInfo);
listener.onFailure(actionDenied(authentication, authzInfo, action, request));
}
Expand Down
7 changes: 6 additions & 1 deletion ...n/java/org/elasticsearch/xpack/security/transport/SecurityServerTransportInterceptor.java
View file
Original file line number Diff line number Diff line change
Expand Up @@ -23,6 +23,7 @@
import org.elasticsearch.license.LicenseUtils;
import org.elasticsearch.license.XPackLicenseState;
import org.elasticsearch.tasks.Task;
import org.elasticsearch.tasks.TaskCancellationService;
import org.elasticsearch.threadpool.ThreadPool;
import org.elasticsearch.transport.RemoteConnectionManager;
import org.elasticsearch.transport.RemoteConnectionManager.RemoteClusterAliasWithCredentials;
Expand Down Expand Up @@ -81,7 +82,11 @@ public class SecurityServerTransportInterceptor implements TransportInterceptor
"internal:data/read/esql/open_exchange",
"cluster:internal:data/read/esql/open_exchange",
"internal:data/read/esql/exchange",
"cluster:internal:data/read/esql/exchange"
"cluster:internal:data/read/esql/exchange",
TaskCancellationService.BAN_PARENT_ACTION_NAME,
TaskCancellationService.REMOTE_CLUSTER_BAN_PARENT_ACTION_NAME,
TaskCancellationService.CANCEL_CHILD_ACTION_NAME,
TaskCancellationService.REMOTE_CLUSTER_CANCEL_CHILD_ACTION_NAME
);

private final AuthenticationService authcService;
Expand Down

0 comments on commit 6dcb6a2

Please sign in to comment.