Fix API key role descriptors rewrite bug for upgraded clusters (#62917)

x-pack/plugin/core/src/main/java/org/elasticsearch/xpack/core/security/SecurityContext.java

@@ -14,6 +14,7 @@
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
+import org.elasticsearch.common.xcontent.XContentBuilder;
 import org.elasticsearch.common.xcontent.XContentHelper;
 import org.elasticsearch.common.xcontent.XContentType;
 import org.elasticsearch.node.Node;
@@ -167,19 +168,36 @@ public void executeAfterRewritingAuthentication(Consumer<StoredContext> consumer
 
     private Map<String, Object> rewriteMetadataForApiKeyRoleDescriptors(Version streamVersion, Authentication authentication) {
         Map<String, Object> metadata = authentication.getMetadata();
-        if (authentication.getAuthenticationType() == AuthenticationType.API_KEY
-            && authentication.getVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)
-            && streamVersion.before(VERSION_API_KEY_ROLES_AS_BYTES)) {
-            metadata = new HashMap<>(metadata);
-            metadata.put(
-                API_KEY_ROLE_DESCRIPTORS_KEY,
-                XContentHelper.convertToMap(
-                    (BytesReference) metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY), false, XContentType.JSON).v2());
-            metadata.put(
-                API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
-                XContentHelper.convertToMap(
-                    (BytesReference) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY), false, XContentType.JSON).v2());
+        if (authentication.getAuthenticationType() == AuthenticationType.API_KEY) {
+            if (authentication.getVersion().onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)
+                && streamVersion.before(VERSION_API_KEY_ROLES_AS_BYTES)) {
+                metadata = new HashMap<>(metadata);
+                metadata.put(API_KEY_ROLE_DESCRIPTORS_KEY,
+                    convertRoleDescriptorsBytesToMap((BytesReference) metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY)));
+                metadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                    convertRoleDescriptorsBytesToMap((BytesReference) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY)));
+            } else if (authentication.getVersion().before(VERSION_API_KEY_ROLES_AS_BYTES)
+                && streamVersion.onOrAfter(VERSION_API_KEY_ROLES_AS_BYTES)) {
+                metadata = new HashMap<>(metadata);
+                metadata.put(API_KEY_ROLE_DESCRIPTORS_KEY,
+                    convertRoleDescriptorsMapToBytes((Map<String, Object>)metadata.get(API_KEY_ROLE_DESCRIPTORS_KEY)));
+                metadata.put(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY,
+                    convertRoleDescriptorsMapToBytes((Map<String, Object>) metadata.get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY)));
+            }
         }
         return metadata;
     }
+
+    private Map<String, Object> convertRoleDescriptorsBytesToMap(BytesReference roleDescriptorsBytes) {
+        return XContentHelper.convertToMap(roleDescriptorsBytes, false, XContentType.JSON).v2();
+    }
+
+    private BytesReference convertRoleDescriptorsMapToBytes(Map<String, Object> roleDescriptorsMap) {
+        try(XContentBuilder builder = XContentBuilder.builder(XContentType.JSON.xContent())) {
+            builder.map(roleDescriptorsMap);
+            return BytesReference.bytes(builder);
+        } catch (IOException e) {
+            throw new UncheckedIOException(e);
+        }
+    }
 }

x-pack/plugin/security/src/test/java/org/elasticsearch/xpack/security/SecurityContextTests.java

@@ -7,6 +7,7 @@
 
 import org.elasticsearch.Version;
 import org.elasticsearch.common.bytes.BytesArray;
+import org.elasticsearch.common.bytes.BytesReference;
 import org.elasticsearch.common.settings.Settings;
 import org.elasticsearch.common.util.concurrent.ThreadContext;
 import org.elasticsearch.common.util.concurrent.ThreadContext.StoredContext;
@@ -136,7 +137,7 @@ public void testExecuteAfterRewritingAuthentication() throws IOException {
         assertEquals(original, securityContext.getAuthentication());
     }
 
-    public void testExecuteAfterRewritingAuthenticationShouldRewriteApiKeyMetadataForBwc() throws IOException {
+    public void testExecuteAfterRewritingAuthenticationWillConditionallyRewriteNewApiKeyMetadata() throws IOException {
         User user = new User("test", null, new User("authUser"));
         RealmRef authBy = new RealmRef("_es_api_key", "_es_api_key", "node1");
         final Map<String, Object> metadata = org.elasticsearch.common.collect.Map.of(
@@ -147,6 +148,7 @@ API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, new BytesArray("{\"limitedBy role\": {\"cl
             AuthenticationType.API_KEY, metadata);
         original.writeToContext(threadContext);
 
+        // If target is old node, rewrite new style API key metadata to old format
         securityContext.executeAfterRewritingAuthentication(originalCtx -> {
             Authentication authentication = securityContext.getAuthentication();
             assertEquals(org.elasticsearch.common.collect.Map.of("a role",
@@ -156,9 +158,15 @@ API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY, new BytesArray("{\"limitedBy role\": {\"cl
                 org.elasticsearch.common.collect.Map.of("cluster", org.elasticsearch.common.collect.List.of("all"))),
                 authentication.getMetadata().get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY));
         }, Version.V_7_8_0);
+
+        // If target is new node, no need to rewrite the new style API key metadata
+        securityContext.executeAfterRewritingAuthentication(originalCtx -> {
+            Authentication authentication = securityContext.getAuthentication();
+            assertSame(metadata, authentication.getMetadata());
+        }, VersionUtils.randomVersionBetween(random(), VERSION_API_KEY_ROLES_AS_BYTES, Version.CURRENT));
     }
 
-    public void testExecuteAfterRewritingAuthenticationShouldNotRewriteApiKeyMetadataForOldAuthenticationObject() throws IOException {
+    public void testExecuteAfterRewritingAuthenticationWillConditionallyRewriteOldApiKeyMetadata() throws IOException {
         User user = new User("test", null, new User("authUser"));
         RealmRef authBy = new RealmRef("_es_api_key", "_es_api_key", "node1");
         final Map<String, Object> metadata = org.elasticsearch.common.collect.Map.of(
@@ -170,9 +178,19 @@ public void testExecuteAfterRewritingAuthenticationShouldNotRewriteApiKeyMetadat
         final Authentication original = new Authentication(user, authBy, authBy, Version.V_7_8_0, AuthenticationType.API_KEY, metadata);
         original.writeToContext(threadContext);
 
+        // If target is old node, no need to rewrite old style API key metadata
         securityContext.executeAfterRewritingAuthentication(originalCtx -> {
             Authentication authentication = securityContext.getAuthentication();
             assertSame(metadata, authentication.getMetadata());
-        }, randomFrom(VERSION_API_KEY_ROLES_AS_BYTES, Version.V_7_8_0));
+        }, Version.V_7_8_0);
+
+        // If target is new old, ensure old map style API key metadata is rewritten to bytesreference
+        securityContext.executeAfterRewritingAuthentication(originalCtx -> {
+            Authentication authentication = securityContext.getAuthentication();
+            assertEquals("{\"a role\":{\"cluster\":[\"all\"]}}",
+                ((BytesReference)authentication.getMetadata().get(API_KEY_ROLE_DESCRIPTORS_KEY)).utf8ToString());
+            assertEquals("{\"limitedBy role\":{\"cluster\":[\"all\"]}}",
+                ((BytesReference)authentication.getMetadata().get(API_KEY_LIMITED_ROLE_DESCRIPTORS_KEY)).utf8ToString());
+        }, VersionUtils.randomVersionBetween(random(), VERSION_API_KEY_ROLES_AS_BYTES, Version.CURRENT));
     }
 }

x-pack/qa/full-cluster-restart/build.gradle

@@ -62,6 +62,10 @@ for (Version bwcVersion : BuildParams.bwcVersions.indexCompatible) {
       setting 'xpack.security.transport.ssl.certificate', 'testnode.crt'
       keystore 'xpack.security.transport.ssl.secure_key_passphrase', 'testnode'
       setting 'logger.org.elasticsearch.xpack.watcher', 'DEBUG'
+
+      if (bwcVersion.onOrAfter('6.7.0')) {
+        setting 'xpack.security.authc.api_key.enabled', 'true'
+      }
     }
   }
 

x-pack/qa/full-cluster-restart/src/test/java/org/elasticsearch/xpack/restart/FullClusterRestartIT.java

@@ -10,6 +10,7 @@
 import org.apache.http.util.EntityUtils;
 import org.elasticsearch.Version;
 import org.elasticsearch.client.Request;
+import org.elasticsearch.client.RequestOptions;
 import org.elasticsearch.client.Response;
 import org.elasticsearch.client.ResponseException;
 import org.elasticsearch.client.RestClient;
@@ -58,6 +59,7 @@
 import static org.hamcrest.Matchers.containsString;
 import static org.hamcrest.Matchers.equalTo;
 import static org.hamcrest.Matchers.everyItem;
+import static org.hamcrest.Matchers.greaterThan;
 import static org.hamcrest.Matchers.greaterThanOrEqualTo;
 import static org.hamcrest.Matchers.hasEntry;
 import static org.hamcrest.Matchers.hasItems;
@@ -211,14 +213,7 @@ public void testWatcher() throws Exception {
             }
 
             // Wait for watcher to actually start....
-            Map<String, Object> startWatchResponse = entityAsMap(client().performRequest(new Request("POST", "_watcher/_start")));
-            assertThat(startWatchResponse.get("acknowledged"), equalTo(Boolean.TRUE));
-            assertBusy(() -> {
-                Map<String, Object> statsWatchResponse = entityAsMap(client().performRequest(new Request("GET", "_watcher/stats")));
-                List<?> states = ((List<?>) statsWatchResponse.get("stats"))
-                    .stream().map(o -> ((Map<?, ?>) o).get("watcher_state")).collect(Collectors.toList());
-                assertThat(states, everyItem(is("started")));
-            });
+            startWatcher();
 
             try {
                 assertOldTemplatesAreDeleted();
@@ -228,15 +223,72 @@ public void testWatcher() throws Exception {
                 /* Shut down watcher after every test because watcher can be a bit finicky about shutting down when the node shuts
                  * down. This makes super sure it shuts down *and* causes the test to fail in a sensible spot if it doesn't shut down.
                  */
-                Map<String, Object> stopWatchResponse = entityAsMap(client().performRequest(new Request("POST", "_watcher/_stop")));
-                assertThat(stopWatchResponse.get("acknowledged"), equalTo(Boolean.TRUE));
+                stopWatcher();
+            }
+        }
+    }
+
+    @SuppressWarnings("unchecked")
+    public void testWatcherWithApiKey() throws Exception {
+        assumeTrue("API key is available since 6.7.0", getOldClusterVersion().onOrAfter(Version.V_6_7_0));
+        if (isRunningAgainstOldCluster()) {
+            final Request createApiKeyRequest = new Request("PUT", "/_security/api_key");
+            createApiKeyRequest.setJsonEntity("{\"name\":\"key-1\",\"role_descriptors\":" +
+                "{\"r\":{\"cluster\":[\"all\"],\"indices\":[{\"names\":[\"*\"],\"privileges\":[\"all\"]}]}}}");
+            final Response response = client().performRequest(createApiKeyRequest);
+            final Map<String, Object> createApiKeyResponse = entityAsMap(response);
+
+            Request createWatchWithApiKeyRequest = new Request("PUT", getWatcherEndpoint() + "/watch/watch_with_api_key");
+            createWatchWithApiKeyRequest.setJsonEntity(loadWatch("logging-watch.json"));
+            final byte[] keyBytes =
+                (createApiKeyResponse.get("id") + ":" + createApiKeyResponse.get("api_key")).getBytes(StandardCharsets.UTF_8);
+            final String authHeader = "ApiKey " + Base64.getEncoder().encodeToString(keyBytes);
+            createWatchWithApiKeyRequest.setOptions(RequestOptions.DEFAULT.toBuilder().addHeader("Authorization", authHeader));
+            client().performRequest(createWatchWithApiKeyRequest);
+
+            assertBusy(() -> {
+                final Request getRequest = new Request("GET", getWatcherEndpoint() + "/watch/watch_with_api_key");
+                getRequest.addParameter("filter_path", "status");
+                final Map<String, Object> getWatchStatusResponse = entityAsMap(client().performRequest(getRequest));
+                final Map<String, Object> status = (Map<String, Object>) getWatchStatusResponse.get("status");
+                assertEquals("executed", status.get("execution_state"));
+            });
+
+        } else {
+            logger.info("testing against {}", getOldClusterVersion());
+            try {
+                waitForYellow(".watches,.watcher-history*");
+            } catch (ResponseException e) {
+                String rsp = toStr(client().performRequest(new Request("GET", "/_cluster/state")));
+                logger.info("cluster_state_response=\n{}", rsp);
+                throw e;
+            }
+
+            // Wait for watcher to actually start....
+            startWatcher();
+
+            try {
+                final Request getWatchStatusRequest = new Request("GET", "_watcher/watch/watch_with_api_key");
+                getWatchStatusRequest.addParameter("filter_path", "status");
+                if (getOldClusterVersion().before(Version.V_7_0_0)) {
+                    getWatchStatusRequest.setOptions(
+                        expectWarnings(
+                            SEARCH_INPUT_TYPES_DEPRECATION_MESSAGE
+                        )
+                    );
+                }
+                final Map<String, Object> getWatchStatusResponse = entityAsMap(client().performRequest(getWatchStatusRequest));
+                final Map<String, Object> status = (Map<String, Object>) getWatchStatusResponse.get("status");
+                final int version = (int) status.get("version");
+
                 assertBusy(() -> {
-                    Map<String, Object> statsStoppedWatchResponse = entityAsMap(client().performRequest(
-                        new Request("GET", "_watcher/stats")));
-                    List<?> states = ((List<?>) statsStoppedWatchResponse.get("stats"))
-                        .stream().map(o -> ((Map<?, ?>) o).get("watcher_state")).collect(Collectors.toList());
-                    assertThat(states, everyItem(is("stopped")));
+                    final Map<String, Object> newGetWatchStatusResponse = entityAsMap(client().performRequest(getWatchStatusRequest));
+                    final Map<String, Object> newStatus = (Map<String, Object>) newGetWatchStatusResponse.get("status");
+                    assertThat((int) newStatus.get("version"), greaterThan(version + 2));
+                    assertEquals("executed", newStatus.get("execution_state"));
                 });
+            } finally {
+                stopWatcher();
             }
         }
     }
@@ -651,6 +703,29 @@ private void waitForHits(String indexName, int expectedHits) throws Exception {
         }, 30, TimeUnit.SECONDS);
     }
 
+    private void startWatcher() throws Exception {
+        Map<String, Object> startWatchResponse = entityAsMap(client().performRequest(new Request("POST", "_watcher/_start")));
+        assertThat(startWatchResponse.get("acknowledged"), equalTo(Boolean.TRUE));
+        assertBusy(() -> {
+            Map<String, Object> statsWatchResponse = entityAsMap(client().performRequest(new Request("GET", "_watcher/stats")));
+            List<?> states = ((List<?>) statsWatchResponse.get("stats"))
+                .stream().map(o -> ((Map<?, ?>) o).get("watcher_state")).collect(Collectors.toList());
+            assertThat(states, everyItem(is("started")));
+        });
+    }
+
+    private void stopWatcher() throws Exception {
+        Map<String, Object> stopWatchResponse = entityAsMap(client().performRequest(new Request("POST", "_watcher/_stop")));
+        assertThat(stopWatchResponse.get("acknowledged"), equalTo(Boolean.TRUE));
+        assertBusy(() -> {
+            Map<String, Object> statsStoppedWatchResponse = entityAsMap(client().performRequest(
+                new Request("GET", "_watcher/stats")));
+            List<?> states = ((List<?>) statsStoppedWatchResponse.get("stats"))
+                .stream().map(o -> ((Map<?, ?>) o).get("watcher_state")).collect(Collectors.toList());
+            assertThat(states, everyItem(is("stopped")));
+        });
+    }
+
     static String toStr(Response response) throws IOException {
         return EntityUtils.toString(response.getEntity());
     }

x-pack/qa/full-cluster-restart/src/test/resources/org/elasticsearch/xpack/restart/logging-watch.json

@@ -0,0 +1,27 @@
+{
+  "trigger" : {
+    "schedule": {
+      "interval": "1s"
+    }
+  },
+  "input" : {
+    "search" : {
+      "timeout": "100s",
+      "request" : {
+        "indices" : [ ".watches" ],
+        "body" : {
+          "query" : { "match_all" : {}},
+          "size": 1
+        }
+      }
+    }
+  },
+  "throttle_period_in_millis": 500,
+  "actions" : {
+    "log" : {
+      "logging" : {
+        "text" : "watcher search result: {{ctx.payload}}"
+      }
+    }
+  }
+}