Skip to content

Commit 4ae4e3a

Browse files
authored
Tests for RCS with multile fulfilling clusters (#94904)
This PR adds a new abstract test suite to test common CCS scenarios under the RCS with two fulfilling clusters. The PR adds two concrete instantiations: mixed model (one cluster uses configurable security, other basic) and same model (both clusters use configurable security).
1 parent 67d155b commit 4ae4e3a

File tree

4 files changed

+526
-21
lines changed

4 files changed

+526
-21
lines changed

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

Lines changed: 53 additions & 21 deletions
Original file line numberDiff line numberDiff line change
@@ -37,6 +37,7 @@
3737
import static org.hamcrest.Matchers.equalTo;
3838
import static org.hamcrest.Matchers.hasKey;
3939
import static org.hamcrest.Matchers.is;
40+
import static org.hamcrest.Matchers.nullValue;
4041

4142
public abstract class AbstractRemoteClusterSecurityTestCase extends ESRestTestCase {
4243

@@ -77,9 +78,13 @@ public static void initFulfillingClusterClient() {
7778
if (fulfillingClusterClient != null) {
7879
return;
7980
}
80-
assert fulfillingCluster != null;
81-
final int numberOfFcNodes = fulfillingCluster.getHttpAddresses().split(",").length;
82-
final String url = fulfillingCluster.getHttpAddress(randomIntBetween(0, numberOfFcNodes - 1));
81+
fulfillingClusterClient = buildRestClient(fulfillingCluster);
82+
}
83+
84+
static RestClient buildRestClient(ElasticsearchCluster targetCluster) {
85+
assert targetCluster != null;
86+
final int numberOfFcNodes = targetCluster.getHttpAddresses().split(",").length;
87+
final String url = targetCluster.getHttpAddress(randomIntBetween(0, numberOfFcNodes - 1));
8388

8489
final int portSeparator = url.lastIndexOf(':');
8590
final var httpHost = new HttpHost(url.substring(0, portSeparator), Integer.parseInt(url.substring(portSeparator + 1)), "http");
@@ -90,7 +95,7 @@ public static void initFulfillingClusterClient() {
9095
throw new UncheckedIOException(e);
9196
}
9297
builder.setStrictDeprecationMode(true);
93-
fulfillingClusterClient = builder.build();
98+
return builder.build();
9499
}
95100

96101
@AfterClass
@@ -111,6 +116,10 @@ protected Settings restClientSettings() {
111116

112117
protected static Map<String, Object> createCrossClusterAccessApiKey(String indicesPrivilegesJson) {
113118
initFulfillingClusterClient();
119+
return createCrossClusterAccessApiKey(fulfillingClusterClient, indicesPrivilegesJson);
120+
}
121+
122+
static Map<String, Object> createCrossClusterAccessApiKey(RestClient targetClusterClient, String indicesPrivilegesJson) {
114123
// Create API key on FC
115124
final var createApiKeyRequest = new Request("POST", "/_security/api_key");
116125
createApiKeyRequest.setJsonEntity(Strings.format("""
@@ -124,7 +133,7 @@ protected static Map<String, Object> createCrossClusterAccessApiKey(String indic
124133
}
125134
}""", indicesPrivilegesJson));
126135
try {
127-
final Response createApiKeyResponse = performRequestAgainstFulfillingCluster(createApiKeyRequest);
136+
final Response createApiKeyResponse = performRequestWithAdminUser(targetClusterClient, createApiKeyRequest);
128137
assertOK(createApiKeyResponse);
129138
return responseAsMap(createApiKeyResponse);
130139
} catch (IOException e) {
@@ -133,44 +142,67 @@ protected static Map<String, Object> createCrossClusterAccessApiKey(String indic
133142
}
134143

135144
protected void configureRemoteClusters() throws Exception {
136-
// This method assume the cross cluster access API key is already configured in keystore
137-
configureRemoteClusters(randomBoolean());
145+
configureRemoteCluster(fulfillingCluster, randomBoolean());
138146
}
139147

140-
/**
141-
* Returns API key ID of cross cluster access API key.
142-
*/
143148
protected void configureRemoteClusters(boolean isProxyMode) throws Exception {
144-
// This method assume the cross cluster access API key is already configured in keystore
149+
configureRemoteCluster(fulfillingCluster, isProxyMode);
150+
}
151+
152+
protected void configureRemoteCluster(ElasticsearchCluster targetFulfillingCluster, boolean isProxyMode) throws Exception {
153+
configureRemoteCluster("my_remote_cluster", targetFulfillingCluster, false, isProxyMode, false);
154+
}
155+
156+
protected void configureRemoteCluster(
157+
String clusterAlias,
158+
ElasticsearchCluster targetFulfillingCluster,
159+
boolean basicSecurity,
160+
boolean isProxyMode,
161+
boolean skipUnavailable
162+
) throws Exception {
163+
// For configurable remote cluster security, this method assumes the cross cluster access API key is already configured in keystore
145164
final Settings.Builder builder = Settings.builder();
165+
final String remoteClusterEndpoint = basicSecurity
166+
? targetFulfillingCluster.getTransportEndpoint(0)
167+
: targetFulfillingCluster.getRemoteClusterServerEndpoint(0);
146168
if (isProxyMode) {
147-
builder.put("cluster.remote.my_remote_cluster.mode", "proxy")
148-
.put("cluster.remote.my_remote_cluster.proxy_address", fulfillingCluster.getRemoteClusterServerEndpoint(0));
169+
builder.put("cluster.remote." + clusterAlias + ".mode", "proxy")
170+
.put("cluster.remote." + clusterAlias + ".proxy_address", remoteClusterEndpoint);
149171
} else {
150-
builder.put("cluster.remote.my_remote_cluster.mode", "sniff")
151-
.putList("cluster.remote.my_remote_cluster.seeds", fulfillingCluster.getRemoteClusterServerEndpoint(0));
172+
builder.put("cluster.remote." + clusterAlias + ".mode", "sniff")
173+
.putList("cluster.remote." + clusterAlias + ".seeds", remoteClusterEndpoint);
152174
}
175+
builder.put("cluster.remote." + clusterAlias + ".skip_unavailable", skipUnavailable);
153176
updateClusterSettings(builder.build());
154177

155178
// Ensure remote cluster is connected
156-
final int numberOfFcNodes = fulfillingCluster.getHttpAddresses().split(",").length;
179+
final int numberOfFcNodes = targetFulfillingCluster.getHttpAddresses().split(",").length;
157180
final Request remoteInfoRequest = new Request("GET", "/_remote/info");
158181
assertBusy(() -> {
159182
final Response remoteInfoResponse = adminClient().performRequest(remoteInfoRequest);
160183
assertOK(remoteInfoResponse);
161184
final Map<String, Object> remoteInfoMap = responseAsMap(remoteInfoResponse);
162-
assertThat(remoteInfoMap, hasKey("my_remote_cluster"));
163-
assertThat(ObjectPath.eval("my_remote_cluster.connected", remoteInfoMap), is(true));
185+
assertThat(remoteInfoMap, hasKey(clusterAlias));
186+
assertThat(ObjectPath.eval(clusterAlias + ".connected", remoteInfoMap), is(true));
164187
if (false == isProxyMode) {
165-
assertThat(ObjectPath.eval("my_remote_cluster.num_nodes_connected", remoteInfoMap), equalTo(numberOfFcNodes));
188+
assertThat(ObjectPath.eval(clusterAlias + ".num_nodes_connected", remoteInfoMap), equalTo(numberOfFcNodes));
189+
}
190+
final String credentialsValue = ObjectPath.eval(clusterAlias + ".cluster_credentials", remoteInfoMap);
191+
if (basicSecurity) {
192+
assertThat(credentialsValue, nullValue());
193+
} else {
194+
assertThat(credentialsValue, equalTo("::es_redacted::"));
166195
}
167-
assertThat(ObjectPath.eval("my_remote_cluster.cluster_credentials", remoteInfoMap), equalTo("::es_redacted::"));
168196
});
169197
}
170198

171199
protected static Response performRequestAgainstFulfillingCluster(Request request) throws IOException {
200+
return performRequestWithAdminUser(fulfillingClusterClient, request);
201+
}
202+
203+
protected static Response performRequestWithAdminUser(RestClient targetFulfillingClusterClient, Request request) throws IOException {
172204
request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", basicAuthHeaderValue(USER, PASS)));
173-
return fulfillingClusterClient.performRequest(request);
205+
return targetFulfillingClusterClient.performRequest(request);
174206
}
175207

176208
// TODO centralize common usage of this across all tests
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,226 @@
1+
/*
2+
* Copyright Elasticsearch B.V. and/or licensed to Elasticsearch B.V. under one
3+
* or more contributor license agreements. Licensed under the Elastic License
4+
* 2.0; you may not use this file except in compliance with the Elastic License
5+
* 2.0.
6+
*/
7+
8+
package org.elasticsearch.xpack.remotecluster;
9+
10+
import org.elasticsearch.action.search.SearchResponse;
11+
import org.elasticsearch.client.Request;
12+
import org.elasticsearch.client.RequestOptions;
13+
import org.elasticsearch.client.Response;
14+
import org.elasticsearch.client.ResponseException;
15+
import org.elasticsearch.client.RestClient;
16+
import org.elasticsearch.core.IOUtils;
17+
import org.elasticsearch.core.Strings;
18+
import org.elasticsearch.search.SearchHit;
19+
import org.elasticsearch.test.cluster.ElasticsearchCluster;
20+
import org.elasticsearch.xcontent.ObjectPath;
21+
import org.junit.AfterClass;
22+
import org.junit.BeforeClass;
23+
24+
import java.io.IOException;
25+
import java.util.Arrays;
26+
import java.util.List;
27+
import java.util.Locale;
28+
import java.util.Map;
29+
import java.util.stream.Collectors;
30+
31+
import static org.hamcrest.Matchers.containsInAnyOrder;
32+
import static org.hamcrest.Matchers.equalTo;
33+
import static org.hamcrest.Matchers.hasKey;
34+
import static org.hamcrest.Matchers.is;
35+
36+
public abstract class AbstractRemoteClusterSecurityWithMultipleRemotesRestIT extends AbstractRemoteClusterSecurityTestCase {
37+
38+
protected static ElasticsearchCluster otherFulfillingCluster;
39+
protected static RestClient otherFulfillingClusterClient;
40+
41+
@BeforeClass
42+
public static void initOtherFulfillingClusterClient() {
43+
if (otherFulfillingClusterClient != null) {
44+
return;
45+
}
46+
otherFulfillingClusterClient = buildRestClient(otherFulfillingCluster);
47+
}
48+
49+
@AfterClass
50+
public static void closeOtherFulfillingClusterClient() throws IOException {
51+
IOUtils.close(otherFulfillingClusterClient);
52+
}
53+
54+
public void testCrossClusterSearch() throws Exception {
55+
configureRemoteClusters();
56+
configureRolesOnClusters();
57+
58+
// Fulfilling cluster
59+
{
60+
// Index some documents, so we can attempt to search them from the querying cluster
61+
final Request bulkRequest = new Request("POST", "/_bulk?refresh=true");
62+
bulkRequest.setJsonEntity(Strings.format("""
63+
{ "index": { "_index": "cluster1_index1" } }
64+
{ "name": "doc1" }
65+
{ "index": { "_index": "cluster1_index2" } }
66+
{ "name": "doc2" }\n"""));
67+
assertOK(performRequestAgainstFulfillingCluster(bulkRequest));
68+
}
69+
70+
// Other fulfilling cluster
71+
{
72+
// Index some documents, so we can attempt to search them from the querying cluster
73+
final Request bulkRequest = new Request("POST", "/_bulk?refresh=true");
74+
bulkRequest.setJsonEntity(Strings.format("""
75+
{ "index": { "_index": "cluster2_index1" } }
76+
{ "name": "doc1" }
77+
{ "index": { "_index": "cluster2_index2" } }
78+
{ "name": "doc2" }\n"""));
79+
assertOK(performRequestAgainstOtherFulfillingCluster(bulkRequest));
80+
}
81+
82+
// Query cluster
83+
{
84+
// Index some documents, to use them in a multi-cluster search
85+
final var indexDocRequest = new Request("POST", "/local_index/_doc?refresh=true");
86+
indexDocRequest.setJsonEntity("{\"name\": \"doc1\"}");
87+
assertOK(client().performRequest(indexDocRequest));
88+
89+
// Search across local cluster and both remotes
90+
searchAndAssertIndicesFound(
91+
String.format(
92+
Locale.ROOT,
93+
"/local_index,%s:%s/_search?ccs_minimize_roundtrips=%s",
94+
randomFrom("my_remote_*", "*"),
95+
randomFrom("*_index1", "*"),
96+
randomBoolean()
97+
),
98+
"cluster1_index1",
99+
"cluster2_index1",
100+
"local_index"
101+
);
102+
103+
// Search across both remotes using cluster alias wildcard
104+
searchAndAssertIndicesFound(
105+
String.format(
106+
Locale.ROOT,
107+
"/%s:%s/_search?ccs_minimize_roundtrips=%s",
108+
randomFrom("my_remote_*", "*"),
109+
randomFrom("*_index1", "*"),
110+
randomBoolean()
111+
),
112+
"cluster1_index1",
113+
"cluster2_index1"
114+
);
115+
116+
// Search across both remotes using explicit cluster aliases
117+
searchAndAssertIndicesFound(
118+
String.format(
119+
Locale.ROOT,
120+
"/my_remote_cluster:%s,my_remote_cluster_2:%s/_search?ccs_minimize_roundtrips=%s",
121+
randomFrom("cluster1_index1", "*_index1", "*"),
122+
randomFrom("cluster2_index1", "*_index1", "*"),
123+
randomBoolean()
124+
),
125+
"cluster1_index1",
126+
"cluster2_index1"
127+
);
128+
129+
// Search single remote
130+
final boolean searchFirstCluster = randomBoolean();
131+
final String index1 = searchFirstCluster ? "cluster1_index1" : "cluster2_index1";
132+
searchAndAssertIndicesFound(
133+
String.format(
134+
Locale.ROOT,
135+
"/%s:%s/_search?ccs_minimize_roundtrips=%s",
136+
searchFirstCluster ? "my_remote_cluster" : "my_remote_cluster_2",
137+
randomFrom(index1, "*_index1", "*"),
138+
randomBoolean()
139+
),
140+
index1
141+
);
142+
143+
// To simplify the test setup, we only ever (randomly) set skip_unavailable on the other remote, not on both,
144+
// i.e. the first remote cluster always has skip_unavailable = false.
145+
// This impacts below failure scenarios; in some cases, skipping the other remote results in overall request success
146+
final boolean skipUnavailableOnOtherCluster = isSkipUnavailable("my_remote_cluster_2");
147+
148+
// Search when one cluster throws 403
149+
// No permissions for this index name, so searching for it on either remote will result in 403
150+
final String missingIndex = "missingIndex";
151+
final boolean missingIndexOnFirstCluster = randomBoolean();
152+
// Make sure we search for missing index on at least one remote, possibly both
153+
final boolean missingIndexOnSecondCluster = false == missingIndexOnFirstCluster || randomBoolean();
154+
final String searchPath1 = String.format(
155+
Locale.ROOT,
156+
"/my_remote_cluster:%s,my_remote_cluster_2:%s/_search?ccs_minimize_roundtrips=%s",
157+
missingIndexOnFirstCluster ? missingIndex : randomFrom("cluster1_index1", "*_index1", "*"),
158+
missingIndexOnSecondCluster ? missingIndex : randomFrom("cluster2_index1", "*_index1", "*"),
159+
randomBoolean()
160+
);
161+
if (skipUnavailableOnOtherCluster && false == missingIndexOnFirstCluster) {
162+
// 403 from other cluster is skipped, so we get a result
163+
searchAndAssertIndicesFound(searchPath1, "cluster1_index1");
164+
} else {
165+
searchAndExpect403(searchPath1);
166+
}
167+
168+
// Search with cluster alias wildcard matching both remotes, where index is authorized on one but not the other
169+
final String index2 = randomFrom("cluster1_index1", "cluster2_index1");
170+
final String searchPath2 = String.format(
171+
Locale.ROOT,
172+
"/my_remote_cluster*:%s/_search?ccs_minimize_roundtrips=%s",
173+
index2,
174+
randomBoolean()
175+
);
176+
if (skipUnavailableOnOtherCluster && index2.equals("cluster1_index1")) {
177+
// 403 from other cluster is skipped, so we get a result
178+
searchAndAssertIndicesFound(searchPath2, index2);
179+
} else {
180+
searchAndExpect403(searchPath2);
181+
}
182+
183+
// Search when both clusters throw 403; in this case we always fail because first cluster is not skipped
184+
searchAndExpect403(String.format(Locale.ROOT, "/*:%s/_search?ccs_minimize_roundtrips=%s", "missingIndex", randomBoolean()));
185+
}
186+
}
187+
188+
private static boolean isSkipUnavailable(String clusterAlias) throws IOException {
189+
final Request remoteInfoRequest = new Request("GET", "/_remote/info");
190+
final Response remoteInfoResponse = adminClient().performRequest(remoteInfoRequest);
191+
assertOK(remoteInfoResponse);
192+
final Map<String, Object> remoteInfoMap = responseAsMap(remoteInfoResponse);
193+
assertThat(remoteInfoMap, hasKey(clusterAlias));
194+
assertThat(ObjectPath.eval(clusterAlias + ".connected", remoteInfoMap), is(true));
195+
return ObjectPath.eval(clusterAlias + ".skip_unavailable", remoteInfoMap);
196+
}
197+
198+
private static void searchAndExpect403(String searchPath) {
199+
final ResponseException exception = expectThrows(
200+
ResponseException.class,
201+
() -> performRequestWithRemoteSearchUser(new Request("GET", searchPath))
202+
);
203+
assertThat(exception.getResponse().getStatusLine().getStatusCode(), equalTo(403));
204+
}
205+
206+
protected abstract void configureRolesOnClusters() throws IOException;
207+
208+
static void searchAndAssertIndicesFound(String searchPath, String... expectedIndices) throws IOException {
209+
final Response response = performRequestWithRemoteSearchUser(new Request("GET", searchPath));
210+
assertOK(response);
211+
final SearchResponse searchResponse = SearchResponse.fromXContent(responseAsParser(response));
212+
final List<String> actualIndices = Arrays.stream(searchResponse.getHits().getHits())
213+
.map(SearchHit::getIndex)
214+
.collect(Collectors.toList());
215+
assertThat(actualIndices, containsInAnyOrder(expectedIndices));
216+
}
217+
218+
static Response performRequestWithRemoteSearchUser(final Request request) throws IOException {
219+
request.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", basicAuthHeaderValue(REMOTE_SEARCH_USER, PASS)));
220+
return client().performRequest(request);
221+
}
222+
223+
static Response performRequestAgainstOtherFulfillingCluster(Request putRoleRequest) throws IOException {
224+
return performRequestWithAdminUser(otherFulfillingClusterClient, putRoleRequest);
225+
}
226+
}

0 commit comments

Comments
 (0)